private[streaming] object WriteAheadLogBasedBlockHandler{def checkpointDirToLogDir(checkpointDir:String, streamId: Int): String = {new Path(checkpointDir,new Path("receivedData", streamId.toString)).toString}
}

def createLogForReceiver(sparkConf: SparkConf,fileWalLogDirectory: String,fileWalHadoopConf: Configuration): WriteAheadLog = {createLog(false, sparkConf, fileWalLogDirectory, fileWalHadoopConf)
}

private def createLog(isDriver: Boolean,sparkConf: SparkConf,fileWalLogDirectory: String,fileWalHadoopConf: Configuration): WriteAheadLog = {val classNameOption= if (isDriver) {sparkConf.getOption(DRIVER_WAL_CLASS_CONF_KEY)} else {sparkConf.getOption(RECEIVER_WAL_CLASS_CONF_KEY)}val wal =classNameOption.map { className =>try {instantiateClass(Utils.classForName(className).asInstanceOf[Class[_ <: WriteAheadLog]], sparkConf)} catch {case NonFatal(e) =>throw new SparkException(s"Couldnot create a write ahead log of class $className", e)}}.getOrElse {new FileBasedWriteAheadLog(sparkConf,fileWalLogDirectory, fileWalHadoopConf,getRollingIntervalSecs(sparkConf,isDriver), getMaxFailures(sparkConf, isDriver),shouldCloseFileAfterWrite(sparkConf,isDriver))}if (isBatchingEnabled(sparkConf,isDriver)) {new BatchedWriteAheadLog(wal,sparkConf)} else {wal}
}

def storeBlock(blockId: StreamBlockId, block:ReceivedBlock): ReceivedBlockStoreResult = {var numRecords= None: Option[Long]// Serialize the block so that it can be inserted intobothval serializedBlock= block match {case ArrayBufferBlock(arrayBuffer)=>numRecords = Some(arrayBuffer.size.toLong)blockManager.dataSerialize(blockId,arrayBuffer.iterator)case IteratorBlock(iterator)=>val countIterator= new CountingIterator(iterator)val serializedBlock= blockManager.dataSerialize(blockId, countIterator)numRecords = countIterator.countserializedBlockcase ByteBufferBlock(byteBuffer)=>byteBuffercase _=>throw new Exception(s"Could notpush $blockId to block manager, unexpected block type")}// Store the block in block managerval storeInBlockManagerFuture= Future {val putResult=blockManager.putBytes(blockId,serializedBlock, effectiveStorageLevel, tellMaster = true)if (!putResult.map{ _._1 }.contains(blockId)) {throw new SparkException(s"Could not store $blockId to block manager with storage level $storageLevel")}}// Store the block in write ahead logval storeInWriteAheadLogFuture= Future {writeAheadLog.write(serializedBlock, clock.getTimeMillis())}// Combine the futures, wait for both to complete, andreturn the write ahead log record handleval combinedFuture= storeInBlockManagerFuture.zip(storeInWriteAheadLogFuture).map(_._2)val walRecordHandle= Await.result(combinedFuture, blockStoreTimeout)WriteAheadLogBasedStoreResult(blockId,numRecords, walRecordHandle)
}