//设置package包名称以及导入依赖的类
package elo.scala.web

import io.netty.buffer.ByteBuf
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


class DiscardServerHandler extends ChannelInboundHandlerAdapter { // (1)

  override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef): Unit = { // (2)
    // Discard the received data silently.
//    ((ByteBuf) msg).release(); // (3)
    msg.asInstanceOf[ByteBuf].release()
  }

  override def exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) { // (4)
    // Close the connection when an exception is raised.
    cause.printStackTrace()
    ctx.close()
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.netty4.http

import com.twitter.finagle.http.codec.ChannelBufferUsageTracker
import io.netty.buffer.ByteBuf
import io.netty.channel._

private[http] class ByteBufManager(usageTracker: ChannelBufferUsageTracker)
  extends ChannelDuplexHandler {

  private[this] var bufferUsage = 0L

  override def write(ctx: ChannelHandlerContext, msg: scala.Any, promise: ChannelPromise): Unit = {
    clearBufferUsage()
    super.write(ctx, msg, promise)
  }

  override def channelRead(ctx: ChannelHandlerContext, msg: scala.Any): Unit = {
    msg match {
      case bb: ByteBuf => increaseBufferUsage(bb.capacity)
      case _ => ()
    }
    super.channelRead(ctx, msg)
  }

  override def close(ctx: ChannelHandlerContext, future: ChannelPromise): Unit = {
    clearBufferUsage()
    super.close(ctx, future)
  }

  private[this] def increaseBufferUsage(size: Long) = {
    // Don't change the order of the following statements, as usageTracker may throw an exception.
    usageTracker.increase(size)
    bufferUsage += size
  }

  private[this] def clearBufferUsage() = {
    // Don't change the order of the following statements, as usageTracker may throw an exception.
    usageTracker.decrease(bufferUsage)
    bufferUsage = 0
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.thrift.transport.netty4

import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


@Sharable
private[netty4] object ThriftByteBufToArrayDecoder extends ChannelInboundHandlerAdapter {
  override def channelRead(
    ctx: ChannelHandlerContext,
    msg: scala.Any
  ): Unit = msg match {
    case buffer: ByteBuf =>
      // toArray takes ownership of the buffer
      val array = toArray(buffer)
      ctx.fireChannelRead(array)

    case _ => throw new IllegalArgumentException("no byte buffer")
  }

  // takes ownership of the passed `ByteBuf`
  private def toArray(buffer: ByteBuf): Array[Byte] = {
    val array = new Array[Byte](buffer.readableBytes())
    buffer.readBytes(array)
    buffer.release()    // If you love it, set it free.
    array
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.thrift.transport.netty4

import com.twitter.finagle.thrift.Protocols
import com.twitter.finagle.thrift.transport.AbstractBufferedTransportDecoderTest
import io.netty.buffer.{ByteBuf, Unpooled}
import io.netty.channel.embedded.EmbeddedChannel

class ThriftBufferedTransportDecoderTest extends AbstractBufferedTransportDecoderTest {

  private def getArray(buf: ByteBuf): Array[Byte] = {
    val out = new Array[Byte](buf.readableBytes())
    buf.readBytes(out)
    assert(buf.readableBytes() == 0)
    out
  }
  private def getDecoder = new ThriftBufferedTransportDecoder(Protocols.factory())

  def decode(arrays: Seq[Array[Byte]]): Vector[Array[Byte]] = {
    val data = arrays.map(Unpooled.wrappedBuffer(_))
    val channel = new EmbeddedChannel()

    channel.pipeline().addLast(getDecoder)
    channel.writeInbound(data:_*)

    var acc = Vector.empty[Array[Byte]]
    while (!channel.inboundMessages().isEmpty) {
      acc :+= getArray(channel.readInbound[ByteBuf]())
    }

    acc
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.netty4.channel

import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


@Sharable
object DirectToHeapInboundHandler extends ChannelInboundHandlerAdapter  {
  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = msg match {
    case bb: ByteBuf if bb.isDirect =>
      val heapBuf = ctx.alloc().heapBuffer(bb.readableBytes, bb.capacity)
      heapBuf.writeBytes(bb)

      bb.release()
      ctx.fireChannelRead(heapBuf)
    case _ => ctx.fireChannelRead(msg)
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.netty4.channel

import com.twitter.finagle.stats.StatsReceiver
import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


@Sharable
private[netty4] class RecvBufferSizeStatsHandler(stats: StatsReceiver)
  extends ChannelInboundHandlerAdapter {

  private[this] val receiveBufferBytes = stats.stat("receive_buffer_bytes")

  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = {
    msg match {
      case bb: ByteBuf => receiveBufferBytes.add(bb.readableBytes().toFloat)
      case _ => // NOOP
    }

    ctx.fireChannelRead(msg)
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.netty4.codec

import com.twitter.finagle.Failure
import com.twitter.finagle.netty4.{BufAsByteBuf, ByteBufAsBuf}
import com.twitter.io.Buf
import io.netty.buffer.ByteBuf
import io.netty.channel.{ChannelPromise, ChannelHandlerContext, ChannelDuplexHandler}


private[finagle] class BufCodec extends ChannelDuplexHandler {
  override def write(ctx: ChannelHandlerContext, msg: Any, p: ChannelPromise): Unit =
    msg match {
      case buf: Buf => ctx.write(BufAsByteBuf.Owned(buf), p)
      case typ => p.setFailure(Failure(
        s"unexpected type ${typ.getClass.getSimpleName} when encoding to ByteBuf"))
    }

  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit =
    msg match {
      case bb: ByteBuf => ctx.fireChannelRead(ByteBufAsBuf.Owned(bb))
      case typ => ctx.fireExceptionCaught(Failure(
          s"unexpected type ${typ.getClass.getSimpleName} when encoding to Buf"))
    }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.netty4.channel

import io.netty.buffer.ByteBuf
import io.netty.channel.embedded.EmbeddedChannel
import org.junit.runner.RunWith
import org.scalacheck.Gen
import org.scalatest.junit.JUnitRunner
import org.scalatest.{FunSuite, OneInstancePerTest}
import org.scalatest.prop.GeneratorDrivenPropertyChecks

@RunWith(classOf[JUnitRunner])
class DirectToHeapInboundHandlerTest extends FunSuite
  with GeneratorDrivenPropertyChecks
  with OneInstancePerTest {

  val channel = new EmbeddedChannel(DirectToHeapInboundHandler)

  test("converts direct to heap") {
    forAll(Gen.alphaStr.suchThat(_.nonEmpty)) { s =>
      val in = channel.alloc.directBuffer(s.length)
      in.setBytes(0, s.getBytes("UTF-8"))

      channel.writeInbound(in)

      val out = channel.readInbound[ByteBuf]

      assert(!out.isDirect)
      assert(in == out)
    }
  }

  test("skips non-ByteBufs") {
    forAll { s: String =>
      channel.writeInbound(s)
      assert(channel.readInbound[String] == s)
    }
  }

  test("works when readIndex is not zero") {
    val in = channel.alloc.directBuffer(4)
    in.writeBytes(Array[Byte](0x1, 0x2, 0x3, 0x4))
    in.readerIndex(1)

    channel.writeInbound(in)

    val out = channel.readInbound[ByteBuf]
    assert(!out.isDirect)
    assert(out.readByte() == 0x2)
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.netty4.codec

import com.twitter.finagle.Failure
import com.twitter.io.{Buf, Charsets}
import io.netty.buffer.{ByteBuf, Unpooled}
import io.netty.channel.embedded.EmbeddedChannel
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner

@RunWith(classOf[JUnitRunner])
class BufCodecTest extends FunSuite {
  test("decode") {
    val ch = new EmbeddedChannel(new BufCodec)
    val q = ch.inboundMessages

    ch.writeInbound(Unpooled.wrappedBuffer("hello".getBytes(Charsets.Utf8)))
    assert(q.size == 1)
    assert(q.poll() == Buf.Utf8("hello"))
    assert(q.size == 0)

    intercept[Failure] { ch.writeInbound(new Object) }
  }

  test("encode") {
    val ch = new EmbeddedChannel(new BufCodec)
    val q = ch.outboundMessages

    ch.writeOutbound(Buf.Utf8("hello"))
    assert(q.size == 1)
    assert(q.peek().isInstanceOf[ByteBuf])
    val bb = q.poll().asInstanceOf[ByteBuf]
    assert(bb.toString(Charsets.Utf8) == "hello")
    assert(q.size == 0)

    val channelFuture = ch.write(new Object)
    assert(channelFuture.cause.isInstanceOf[Failure])
  }
} 

//设置package包名称以及导入依赖的类
package tictactoe.net

import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.{ByteToMessageDecoder, MessageToByteEncoder}


object LoginCodec {
  case class LoginRequest(username: String, password: String)
  case class LoginResponse(responseCode: Int)

  type Ctx = ChannelHandlerContext
  type Buffer = ByteBuf
  type Output = java.util.List[AnyRef]

  final class LoginRequestEncoder extends MessageToByteEncoder[LoginRequest] {
    override def encode(ctx: Ctx, msg: LoginRequest, out: Buffer): Unit = {
      out.writeCString(msg.username)
      out.writeCString(msg.password)
    }
  }

  final class LoginResponseDecoder extends ByteToMessageDecoder {
    override def decode(ctx: Ctx, in: Buffer, out: Output): Unit = {
      (in.readableBytes() compare 1) signum match {
        case 0 | 1 => out.add(LoginResponse(in.readUnsignedByte()))
        case -1 => // nothing
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.wavesplatform.network

import java.net.{InetAddress, InetSocketAddress}

import com.google.common.base.Charsets
import io.netty.buffer.ByteBuf

case class Handshake(
    applicationName: String,
    applicationVersion: (Int, Int, Int),
    nodeName: String,
    nodeNonce: Long,
    declaredAddress: Option[InetSocketAddress]) {
  def encode(out: ByteBuf): out.type = {
    out.writeByte(applicationName.length)
    out.writeBytes(applicationName.getBytes(Charsets.UTF_8))
    out.writeInt(applicationVersion._1)
    out.writeInt(applicationVersion._2)
    out.writeInt(applicationVersion._3)
    out.writeByte(nodeName.length)
    out.writeBytes(nodeName.getBytes(Charsets.UTF_8))
    out.writeLong(nodeNonce)
    declaredAddress match {
      case None => out.writeInt(0)
      case Some(addr) =>
        val addressBytes = addr.getAddress.getAddress
        out.writeInt(addressBytes.length + 4)
        out.writeBytes(addressBytes)
        out.writeInt(addr.getPort)
    }
    out.writeLong(System.currentTimeMillis() / 1000)

    out
  }
}

object Handshake {
  def decode(in: ByteBuf): Handshake = {
    val appNameSize = in.readByte()
    val appName = in.readSlice(appNameSize).toString(Charsets.UTF_8)
    val appVersion = (in.readInt(), in.readInt(), in.readInt())
    val nodeNameSize = in.readByte()
    val nodeName = in.readSlice(nodeNameSize).toString(Charsets.UTF_8)
    val nonce = in.readLong()
    val declaredAddressLength = in.readInt()
    // 0 for no declared address, 8 for ipv4 address + port, 20 for ipv6 address + port
    require(declaredAddressLength == 0 || declaredAddressLength == 8 || declaredAddressLength == 20,
      s"invalid declared address length: $declaredAddressLength")
    val isa = if (declaredAddressLength == 0) None else {
      val addressBytes = new Array[Byte](declaredAddressLength - 4)
      in.readBytes(addressBytes)
      val address = InetAddress.getByAddress(addressBytes)
      val port = in.readInt()
      Some(new InetSocketAddress(address, port))
    }
    in.readLong() // time is ignored

    Handshake(appName, appVersion, nodeName, nonce, isa)
  }
} 

//设置package包名称以及导入依赖的类
package com.bwsw.sj.engine.input

import java.util.concurrent._

import com.bwsw.sj.engine.core.engine.TaskRunner
import com.bwsw.sj.engine.input.connection.tcp.server.InputStreamingServer
import com.bwsw.sj.engine.input.task.InputTaskManager
import com.bwsw.sj.engine.input.task.engine.InputTaskEngineFactory
import com.bwsw.sj.engine.input.task.reporting.InputStreamingPerformanceMetrics
import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandlerContext
import org.slf4j.LoggerFactory

import scala.collection.convert.decorateAsScala._



object InputTaskRunner extends {override val threadName = "InputTaskRunner-%d"} with TaskRunner {

  private val logger = LoggerFactory.getLogger(this.getClass)
  private val queueSize = 1000

  def main(args: Array[String]) {
    try {
      val bufferForEachContext = (new ConcurrentHashMap[ChannelHandlerContext, ByteBuf]()).asScala
      val channelContextQueue = new ArrayBlockingQueue[ChannelHandlerContext](queueSize)

      val manager: InputTaskManager = new InputTaskManager()
      logger.info(s"Task: ${manager.taskName}. Start preparing of task runner for an input module\n")

      val performanceMetrics = new InputStreamingPerformanceMetrics(manager)

      val inputTaskEngineFactory = new InputTaskEngineFactory(manager, performanceMetrics, channelContextQueue, bufferForEachContext)

      val inputTaskEngine = inputTaskEngineFactory.createInputTaskEngine()

      val inputStreamingServer = new InputStreamingServer(
        manager.agentsHost,
        manager.entryPort,
        inputTaskEngine.executor,
        channelContextQueue,
        bufferForEachContext
      )

      logger.info(s"Task: ${manager.taskName}. Preparing finished. Launch task\n")

      executorService.submit(inputTaskEngine)
      executorService.submit(performanceMetrics)
      executorService.submit(inputStreamingServer)

      executorService.take().get()
    } catch {
      case assertionError: Error => handleException(assertionError)
      case exception: Exception => handleException(exception)
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.bwsw.sj.engine.input.connection.tcp.server

import java.util.concurrent.ArrayBlockingQueue

import com.bwsw.sj.engine.core.input.InputStreamingExecutor
import io.netty.buffer.ByteBuf
import io.netty.channel.socket.SocketChannel
import io.netty.channel.{ChannelHandlerContext, ChannelInitializer}
import io.netty.handler.codec.string.StringEncoder
import io.netty.handler.logging.{LogLevel, LoggingHandler}

import scala.collection.concurrent


class InputStreamingChannelInitializer(executor: InputStreamingExecutor,
                                       channelContextQueue: ArrayBlockingQueue[ChannelHandlerContext],
                                       bufferForEachContext: concurrent.Map[ChannelHandlerContext, ByteBuf])
  extends ChannelInitializer[SocketChannel] {

  def initChannel(channel: SocketChannel) = {
    val pipeline = channel.pipeline()

    pipeline.addLast("logger", new LoggingHandler(LogLevel.WARN))
    pipeline.addLast("encoder", new StringEncoder())
    pipeline.addLast("handler", new InputStreamingServerHandler(executor, channelContextQueue, bufferForEachContext))
  }
} 

//设置package包名称以及导入依赖的类
package com.bwsw.sj.engine.input.connection.tcp.server

import java.util.concurrent.{Callable, ArrayBlockingQueue}

import com.bwsw.sj.engine.core.input.InputStreamingExecutor
import io.netty.bootstrap.ServerBootstrap
import io.netty.buffer.ByteBuf
import io.netty.channel._
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.nio.NioServerSocketChannel
import io.netty.handler.logging.{LogLevel, LoggingHandler}
import org.slf4j.LoggerFactory

import scala.collection.concurrent



class InputStreamingServer(host: String,
                           port: Int,
                           executor: InputStreamingExecutor,
                           channelContextQueue: ArrayBlockingQueue[ChannelHandlerContext],
                           bufferForEachContext: concurrent.Map[ChannelHandlerContext, ByteBuf]) extends Callable[Unit] {

  private val logger = LoggerFactory.getLogger(this.getClass)

  override def call() = {
    logger.info(s"Launch input streaming server on: '$host:$port'\n")
    val bossGroup: EventLoopGroup = new NioEventLoopGroup()
    val workerGroup = new NioEventLoopGroup()
    try {
      val bootstrapServer = new ServerBootstrap()
      bootstrapServer.group(bossGroup, workerGroup)
        .channel(classOf[NioServerSocketChannel])
        .handler(new LoggingHandler(LogLevel.INFO))
        .childHandler(new InputStreamingChannelInitializer(executor,  channelContextQueue, bufferForEachContext))

      bootstrapServer.bind(host, port).sync().channel().closeFuture().sync()
    } finally {
      workerGroup.shutdownGracefully()
      bossGroup.shutdownGracefully()
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.bwsw.tstreamstransactionserver.netty

import io.netty.buffer.ByteBuf
import io.netty.buffer.Unpooled._



  def fromByteArray(bytes: Array[Byte]): Message = {
    val buffer = java.nio.ByteBuffer.wrap(bytes)
    val id     = buffer.getLong
    val protocol = buffer.get
    val token = buffer.getInt
    val method = buffer.get()
    val isFireAndForgetMethod = buffer.get()
    val length = buffer.getInt
    val message = {
      val bytes = new Array[Byte](buffer.limit() - headerFieldSize - lengthFieldSize)
      buffer.get(bytes)
      bytes
    }
    Message(id, length, protocol, message, token, method, isFireAndForgetMethod)
  }

  def fromByteBuf(buf: ByteBuf): Message = {
    val id       = buf.readLong()
    val protocol = buf.readByte()
    val token    = buf.readInt()
    val method   = buf.readByte()
    val isFireAndForgetMethod = buf.readByte()
    val length   = buf.readInt()
    val message = {
      val bytes = new Array[Byte](buf.readableBytes())
      buf.readBytes(bytes)
      bytes
    }
    Message(id, length, protocol, message, token, method, isFireAndForgetMethod)
  }

} 

//设置package包名称以及导入依赖的类
package com.bwsw.tstreamstransactionserver.netty.server

import com.bwsw.tstreamstransactionserver.netty.Message
import com.bwsw.tstreamstransactionserver.options.ServerOptions.TransportOptions
import io.netty.buffer.ByteBuf
import io.netty.channel.socket.SocketChannel
import io.netty.channel.{ChannelInitializer, SimpleChannelInboundHandler}
import io.netty.handler.codec.LengthFieldBasedFrameDecoder
import io.netty.handler.codec.bytes.ByteArrayEncoder



class ServerInitializer(serverHandler: => SimpleChannelInboundHandler[ByteBuf]) extends ChannelInitializer[SocketChannel] {
  override def initChannel(ch: SocketChannel): Unit = {
    ch.pipeline()
      .addLast(new ByteArrayEncoder())
      .addLast(new LengthFieldBasedFrameDecoder(
        Int.MaxValue,
        //packageTransmissionOpts.maxDataPackageSize max packageTransmissionOpts.maxMetadataPackageSize,
        Message.headerFieldSize,
        Message.lengthFieldSize)
      )
      .addLast(serverHandler)
  }
} 

//设置package包名称以及导入依赖的类
package strd.net.http

import io.netty.buffer.ByteBuf
import io.netty.handler.codec.http.HttpHeaders.Names._
import io.netty.handler.codec.http.HttpResponseStatus
import io.netty.handler.codec.http.HttpResponseStatus._


case class HttpResp(status      : HttpResponseStatus = OK,
                    body        : Option[ByteBuf] = None,
                    headers     : MultiStringMap = Map.empty) {

  def +(kv: (String, Seq[String])) = copy(headers = headers + kv)
  def :+(kv: (String, String)) = copy(headers = headers :+ kv)
}

object HttpResp {
  val EMPTY       = HttpResp(NO_CONTENT)
  val BAD_REQ     = HttpResp(BAD_REQUEST)
  val NOT_FOUND   = HttpResp(HttpResponseStatus.NOT_FOUND)


  def badRequest(msg : String) = HttpResp(BAD_REQUEST, Some(Content(msg)))
  def notFound(msg : String) = HttpResp(HttpResponseStatus.NOT_FOUND, Some(Content(msg)))


  def ignoreCache( status      : HttpResponseStatus = OK,
                   body        : Option[ByteBuf] = None,
                   headers     : MultiStringMap = Map.empty) = {
    HttpResp(status, body, headers ++ DISABLE_CACHE)
  }

  def withCache( status      : HttpResponseStatus = OK,
                   body      : Option[ByteBuf] = None,
                   headers   : MultiStringMap = Map.empty) = {
    HttpResp(status, body, headers ++ ENABLE_CACHE)
  }

  def pixel(headers : MultiStringMap = Map.empty) = HttpResp(
    headers = headers + ContentType(_.GIF),
    body    = Some(Content(EMPTY_GIF))
  )

  def redirect(url: String, headers : MultiStringMap = Map.empty) = {
    require (url != null, "null url found")
    HttpResp(
      status = FOUND,
      headers = headers :+ (LOCATION -> url)
    )
  }

} 

//设置package包名称以及导入依赖的类
package strd.net

import io.netty.buffer.ByteBuf

import scala.util.matching.Regex


package object http extends MultiStringMapUtils
                            with HttpMethodAliases
                            with HeaderUtils {


  type MultiStringMap = Map[String, Seq[String]]
  type MultiString =  (String, Seq[String])

  implicit def stringToContent(str:String) : Option[ByteBuf] = Some(Content(str))

  implicit class RegexContext(sc: StringContext) {
    def r = new Regex(sc.parts.mkString, sc.parts.tail.map(_ => "x"): _*)
  }

  val EMPTY_GIF = {
    val stream = Thread.currentThread().getContextClassLoader.getResourceAsStream("empty.gif")
    if (stream == null ) {
      throw new RuntimeException("Can not find resource :empty.gif")
    }
    val ba = new Array[Byte](stream.available())
    stream.read(ba)
    ba
  }
} 

//设置package包名称以及导入依赖的类
package com.scxmpp.netty

import java.util

import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.ByteToMessageDecoder

import com.fasterxml.aalto.AsyncXMLStreamReader
import com.fasterxml.aalto.evt.EventAllocatorImpl
import com.fasterxml.aalto.stax.InputFactoryImpl

class XmlFrameDecoder extends ByteToMessageDecoder
  with ResettableChannelInboundHandler {

  val factory = new InputFactoryImpl()
  var reader = factory.createAsyncForByteBuffer
  val allocator = EventAllocatorImpl.getDefaultInstance

  def reset() {
    reader = factory.createAsyncForByteBuffer
  }

  @throws[Exception]
  override def decode(ctx: ChannelHandlerContext, buffer: ByteBuf, out: util.List[Object]) {
    reader.getInputFeeder.feedInput(buffer.nioBuffer)
    buffer.skipBytes(buffer.readableBytes)

    while (reader.hasNext && reader.next != AsyncXMLStreamReader.EVENT_INCOMPLETE) {
      out.add(allocator.allocate(reader))
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.bwsw.sj.examples.pingstation.module.input

import com.bwsw.common.{JsonSerializer, ObjectSerializer}
import com.bwsw.sj.engine.core.entities.{UnreachableResponse, EchoResponse, InputEnvelope}
import com.bwsw.sj.engine.core.environment.InputEnvironmentManager
import com.bwsw.sj.engine.core.input.{InputStreamingExecutor, Interval}
import io.netty.buffer.ByteBuf


class Executor(manager: InputEnvironmentManager) extends InputStreamingExecutor(manager) {

  val objectSerializer = new ObjectSerializer()
  val jsonSerializer = new JsonSerializer()
  val echoResponseStreamName = "echo-response"
  val unreachableResponseStreamName = "unreachable-response"
  val partition = 0

  
  override def parse(buffer: ByteBuf, interval: Interval): Option[InputEnvelope] = {

    val ts = System.currentTimeMillis()
    val rawData = buffer.slice(interval.initialValue, interval.finalValue)

    val data = new Array[Byte](rawData.capacity())
    rawData.getBytes(0, data)

    val fpingResponse = new String(data)

    val parsedResponse = fpingResponse.split("\\s+")
    parsedResponse.head match {
      case "ICMP" =>

        val response = new UnreachableResponse(ts, parsedResponse.last)
        println(response + ";") //todo for testing
        val serializedResponse = objectSerializer.serialize(jsonSerializer.serialize(response))

        Some(new InputEnvelope(
          fpingResponse,
          Array((unreachableResponseStreamName, partition)),
          false,
          serializedResponse
        ))
      case ip =>

        val response = new EchoResponse(ts, ip, parsedResponse(5).toDouble)
        println(response + ";") //todo for testing
        val serializedResponse = objectSerializer.serialize(jsonSerializer.serialize(response))

        Some(new InputEnvelope(
          fpingResponse,
          Array((echoResponseStreamName, partition)),
          false,
          serializedResponse
        ))
    }
  }
}