//设置package包名称以及导入依赖的类
package serialisation

import java.util

import model.Event
import org.apache.kafka.common.serialization.{Serializer, StringSerializer}
import org.json4s.native.Serialization
import org.json4s.native.Serialization.write
import org.json4s.{Formats, NoTypeHints}

class EventSerialiser extends Serializer[Event] {

  implicit val formats: Formats = Serialization.formats(NoTypeHints)

  val stringSerialiser = new StringSerializer

  override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = {
    stringSerialiser.configure(configs, isKey)
  }

  override def serialize(topic: String, data: Event): Array[Byte] = {
    val stringValue = write(data)
    stringSerialiser.serialize(topic, stringValue)
  }

  override def close(): Unit = {
    stringSerialiser.close()
  }
} 

//设置package包名称以及导入依赖的类
package transformation

import model.Event
import org.apache.kafka.streams.KeyValue
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import org.json4s.{Formats, NoTypeHints}

class StringEventCollector(storeName: String)
  extends CollectTransformer[String, String, String, String, String](storeName) {

  implicit val formats: Formats = Serialization.formats(NoTypeHints)

  override def appendToStore(storeValue: String, appendValue: String): String = {
    appendValue
  }

  override def collectComplete(storeValue: String, appendValue: String): Boolean = {
    // just merges two events
    storeValue.nonEmpty && appendValue.nonEmpty
  }

  override def collectOutput(inputKey: String, storeValue: String, mergeValue: String): KeyValue[String, String] = {
    new KeyValue(inputKey, collect(storeValue, mergeValue))
  }

  def collect(storeValue: String, mergeValue: String): String = {
    log.info(s"Collecting ...")
    log.info(s"   with store : $storeValue")
    log.info(s"   with event : $mergeValue")
    val firstEvent = read[Event](storeValue)
    val secondEvent = read[Event](mergeValue)
    write(Event(firstEvent.eventType + secondEvent.eventType,
      firstEvent.count + secondEvent.count,
      firstEvent.name + secondEvent.name))
  }
} 

//设置package包名称以及导入依赖的类
package org.argus.amandroid.serialization

import java.io.{FileReader, FileWriter}

import org.argus.amandroid.alir.componentSummary.ApkYard
import org.argus.amandroid.core.decompile.{ConverterUtil, DecompileLayout, DecompileStrategy, DecompilerSettings}
import org.argus.amandroid.core.model.ApkModel
import org.argus.jawa.core.DefaultReporter
import org.json4s.NoTypeHints
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import org.scalatest.{FlatSpec, Matchers}
import org.argus.jawa.core.util.FileUtil


class SerializationTest extends FlatSpec with Matchers {

  "ApkModel" should "successfully serialized and deserialized" in {
    val apkFile = getClass.getResource("/icc-bench/IccHandling/icc_explicit_src_sink.apk").getPath
    val apkUri = FileUtil.toUri(apkFile)
    val outputUri = FileUtil.toUri(apkFile.substring(0, apkFile.length - 4))
    val reporter = new DefaultReporter
    val yard = new ApkYard(reporter)
    val layout = DecompileLayout(outputUri)
    val strategy = DecompileStrategy(layout)
    val settings = DecompilerSettings(debugMode = false, forceDelete = true, strategy, reporter)
    val apk = yard.loadApk(apkUri, settings, collectInfo = true)
    val model = apk.model
    implicit val formats = Serialization.formats(NoTypeHints) + ApkModelSerializer
    val apkRes = FileUtil.toFile(FileUtil.appendFileName(outputUri, "apk.json"))
    val oapk = new FileWriter(apkRes)
    try {
      write(model, oapk)
    } catch {
      case e: Exception =>
        e.printStackTrace()
    } finally {
      oapk.flush()
      oapk.close()
    }
    val iapk = new FileReader(apkRes)
    var newApkModel: ApkModel = null
    try {
      newApkModel = read[ApkModel](iapk)
    } catch {
      case e: Exception =>
        e.printStackTrace()
    } finally {
      iapk.close()
      ConverterUtil.cleanDir(outputUri)
    }
    require(
      model.getAppName == newApkModel.getAppName &&
      model.getComponents == newApkModel.getComponents &&
      model.getLayoutControls == newApkModel.getLayoutControls &&
      model.getCallbackMethods == newApkModel.getCallbackMethods &&
      model.getComponentInfos == newApkModel.getComponentInfos &&
      model.getEnvMap == newApkModel.getEnvMap)
  }
} 

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

import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}

import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.FSM
import akka.actor.FSM._
import akka.actor.Props
import scala.collection.immutable.Queue

case class ItemData(val erpData: ERPData, val specData: SpecData, val prodData: Queue[_])

class Item extends Actor {
    val fsm: ActorRef = context.actorOf(Props[ItemFSM])
    var erpData: Option[ERPData] = None
    var specData: Option[SpecData] = None
    var prodData: Option[Queue[_]] = None

    implicit val formats = Serialization.formats(NoTypeHints)

    def receive = {
        case (x: ProdEvent, y: ProdData) => fsm ! (x, y)
        case x: ERPData => erpData = Some(x)
        case x: SpecData => {
            specData = Some(x)
            trySubmit()
        }
        case x: Queue[_] => prodData = Some(x)
    }

    def itemData: ItemData = {
        val item = for {
            e <- erpData
            s <- specData
            p <- prodData
        } yield ItemData(e, s, p)
        item.get
    }

    def serialize(item: ItemData): String = {
        write(item)
    }

    def trySubmit() {
        erpData.isDefined && specData.isDefined && prodData.isDefined match {
            case true => context.parent ! serialize(itemData)
            case _ => ()
        }
    }
} 

//设置package包名称以及导入依赖的类
package isWaitInstruction

import akka.actor._
import core.Domain._
import core.Helpers._
import core.ServiceBase
import org.json4s._
import org.json4s.native.Serialization.write



class IsWaitFiller extends ServiceBase {
  // Type aliases
  type Instruction = String

  // Functions
  def handleAmqMessage(json: JValue) = {
    if (json.has("programPointerPosition") && json.has("instruction") && !json.has("isWaiting")) {
      val event: PointerWithInstruction = json.extract[PointerWithInstruction]
      fill(event)
    } else {
      // do nothing... OR log.info("Received message of unmanageable type property.")
    }
  }

  def fill(event: PointerWithInstruction) = {
    val instruction: Instruction = event.instruction
    var isWaiting: Boolean = false
    if (instruction.startsWith("Wait") || instruction.startsWith("ExecEngine"))
      isWaiting = true
    val filledEvent = PointerWithIsWaiting(event.robotId, event.workCellId,event.address, instruction, isWaiting,
      event.programPointerPosition)
    val json: String = write(filledEvent)
    log.info("From isWaiting: " + json)
    sendToBus(json)
  }
}

object IsWaitFiller {
  def props = Props[IsWaitFiller]
} 

//设置package包名称以及导入依赖的类
package waitChange

import akka.actor._
import core.Domain._
import core.Helpers._
import core.ServiceBase
import java.util.UUID
import org.json4s._
import org.json4s.native.Serialization.write



class WaitChange extends ServiceBase {
  // Type aliases
  type RobotId = String
  type ActivityId = UUID
  type WaitInstruction = String

  // State
  var isWaiting: Map[RobotId, Option[(ActivityId, WaitInstruction)]] = Map.empty

  // Functions
  def handleAmqMessage(json: JValue) = {
    if (json.has("isWaiting"))
      checkIfWaitChange(json)
  }

  def checkIfWaitChange(json: JValue) = {
    val event: PointerWithIsWaiting = json.extract[PointerWithIsWaiting]

    if (!isWaiting.contains(event.robotId)) {
      isWaiting += (event.robotId -> None)
    }

    if (isWaiting(event.robotId).isDefined != event.isWaiting) {
      val (activityId, waitInstruction): (ActivityId, WaitInstruction) = if (event.isWaiting) {
        val id = UUID.randomUUID()
        isWaiting += (event.robotId -> Some((id, event.instruction)))
        (id, event.instruction)
      } else {
        val (id, instruction) = isWaiting(event.robotId).get
        isWaiting += (event.robotId -> None)
        (id, instruction)
      }
      val activityEvent = ActivityEvent(activityId.toString, event.isWaiting, waitInstruction, event.robotId,
        event.programPointerPosition.time, "wait", event.workCellId)
      log.info("From waitChange: " + activityEvent)
      sendToBus(write(activityEvent))
    }

  }
}

object WaitChange {
  def props = Props[WaitChange]
} 

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

import java.nio.charset.{Charset, StandardCharsets}
import java.nio.file.{Files, Path, Paths}

import com.ponkotuy.queries.{Exif, LoginEmail}
import org.json4s.DefaultFormats
import org.json4s.ext.JodaTimeSerializers
import org.json4s.native.Serialization.write
import skinny.http.{HTTP, Request}
import com.ponkotuy.utils.EitherUtil.eitherToRightProjection

import scala.compat.java8.StreamConverters._

object Main {
  implicit val formats = DefaultFormats ++ JodaTimeSerializers.all

  val config = new Config

  def main(args: Array[String]): Unit = {
    val extractor = new com.ponkotuy.Extractor
    login().fold(println("Auth failed")){ session =>
      args.foreach{ raw =>
        val exifs = files(raw).filter(_.toString.toLowerCase.endsWith(".jpg")).flatMap{ file =>
          val map = extractor.read(file).tagMaps
          Exif.fromMap(file.getFileName.toString, map).left.map { e =>
            println(s"${file.toString} parse error: ${e} not found.")
          }.toOption
        }
        exifs.foreach{ exif =>
          println(exif.fileName)
          val req = Request(s"${config.server}/api/exif")
              .header("Cookie", session.toString)
              .body(write(exif).getBytes(Charset.forName("UTF-8")), "application/json")
          val res = HTTP.post(req)
          if(res.status / 100 != 2) println(new String(res.body))
        }
      }
    }
  }

  private def login(): Option[Cookie] = {
    val auth = LoginEmail(config.email, config.password)
    val json = write(auth).getBytes(StandardCharsets.UTF_8)
    val req = Request(s"${config.server}/api/session").body(json, "application/json")
    val res = HTTP.post(req)
    res.header("Set-Cookie").map(Cookie.fromStr)
  }

  private def files(raw: String): List[Path] = {
    val path = Paths.get(raw)
    if(Files.isDirectory(path)) Files.list(path).toScala[List]
    else path :: Nil
  }
} 

//设置package包名称以及导入依赖的类
package com.fustigatedcat.heystk.engine.queue

import com.fustigatedcat.heystk.engine.EngineAPI
import com.fustigatedcat.heystk.common.normalization.Normalization
import com.rabbitmq.client.{MessageProperties, BuiltinExchangeType, ConnectionFactory}
import org.json4s.native.Serialization.write
import org.slf4j.LoggerFactory

object RabbitQueue {

  implicit val formats = org.json4s.DefaultFormats

  val logger = LoggerFactory.getLogger(this.getClass)

  val config = EngineAPI.config.getConfig("engine.queue.amqp")

  val apiExchangeName = config.getString("api.exchange-name")

  val toProcessRoutingKey = config.getString("api.routing-key")

  val factory = {
    val f = new ConnectionFactory()
    f.setHost(config.getString("host"))
    f.setPort(config.getInt("port"))
    f.setVirtualHost(config.getString("vhost"))
    f.setUsername(config.getString("user"))
    f.setPassword(config.getString("password"))
    f
  }

  val connection = factory.newConnection()

  val channel = connection.createChannel()

  def postToProcess(normalization : Normalization) = {
    val norm = write(normalization)
    logger.debug("Publishing normalization {}", norm)
    channel.basicPublish(
      apiExchangeName,
      toProcessRoutingKey,
      MessageProperties.PERSISTENT_TEXT_PLAIN,
      norm.getBytes
    )
  }

} 

//设置package包名称以及导入依赖的类
package com.mariotti.urlshortener.dao.persistence

import com.mariotti.urlshortener.model.Stats
import java.nio.file.{ Paths, Files }
import scala.io.Source
import java.nio.charset.StandardCharsets
import org.json4s._
import org.json4s.native.JsonMethods._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{ read, write }

object Persistence extends PersistenceWrapper {
  val MAPPING_PREFIX = "MAPPING_"
  val STATS_PREFIX = "STATS_"

  
  private def getStatsFileName(id: Int): String = {
    STATS_PREFIX + String.valueOf(id)
  }

  def get(id: Int): String = {
    val filename = getMappingFileName(id);
    if (Files.exists(Paths.get(filename))) {
      Source.fromFile(filename).getLines.mkString
    } else
      null;
  }
  def put(id: Int, value: String) {
    val filename = getMappingFileName(id);
    if (!Files.exists(Paths.get(filename))) {
      Files.write(Paths.get(filename), value.getBytes(StandardCharsets.UTF_8))
    } else
      throw new Exception("key collision detected when writing file")
  }
  def getStats(id: Int): Stats = {
    val filename = getStatsFileName(id);
    implicit val formats = Serialization.formats(NoTypeHints)
    val ser = Source.fromFile(filename).getLines.mkString
    read[Stats](ser)
  }
  def putStats(id: Int, stats: Stats) {
    if (stats != null) {
      val filename = getStatsFileName(id);
      implicit val formats = Serialization.formats(NoTypeHints)
      val ser = write(stats)
      Files.write(Paths.get(filename), ser.getBytes(StandardCharsets.UTF_8))
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.protegra_ati.agentservices.protocols

import java.net.URI
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}

class URISerializer extends Serializer[URI] {
  private val URIClass = classOf[URI]

  override def deserialize(implicit format: Formats): PartialFunction[(TypeInfo, JValue), URI] = {
    case (TypeInfo(URIClass, _), json) => json match {
      case JString(s) => new URI(s)
      case x => throw new MappingException("Can't convert " + x + " to java.net.URI")
    }
  }

  override def serialize(implicit format: Formats): PartialFunction[Any, JValue] = {
    case x: URI => JString(x.toString)
  }
}

object Serializer {
  implicit val formats = Serialization.formats(NoTypeHints) + new URISerializer

  def serialize(o: AnyRef): String = {
    write(o)
  }

  def deserialize[T: Manifest](s: String): T = {
    read[T](s)
  }
} 

//设置package包名称以及导入依赖的类
package com.biosimilarity.evaluator.spray

import java.net.URI
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}

class URISerializer extends Serializer[URI] {
  private val URIClass = classOf[URI]

  override def deserialize(implicit format: Formats): PartialFunction[(TypeInfo, JValue), URI] = {
    case (TypeInfo(URIClass, _), json) => json match {
      case JString(s) => new URI(s)
      case x => throw new MappingException("Can't convert " + x + " to java.net.URI")
    }
  }

  override def serialize(implicit format: Formats): PartialFunction[Any, JValue] = {
    case x: URI => JString(x.toString)
  }
}

object Serializer {
  implicit val formats = Serialization.formats(NoTypeHints) + new URISerializer

  def serialize(o: AnyRef): String = {
    write(o)
  }

  def deserialize[T: Manifest](s: String): T = {
    read[T](s)
  }
} 

//设置package包名称以及导入依赖的类
package renesca.json.protocols

import org.json4s.native.Serialization
import renesca.Query
import renesca.json.{Request, Statement}
import org.json4s._
import org.json4s.JsonDSL._
import native.JsonMethods._
import org.json4s.native.Serialization._
import org.json4s.native.Serialization.{read, write}
import renesca.parameter._


object RequestJson {

  implicit val formats = Serialization.formats(NoTypeHints) + new PropertyKeySerializer() + new ParameterMapSerializer()

  def jsonOf(obj: Request): String = {
    write(obj)
  }

  def jsonOf(obj: Statement): String = {
    write(obj)
  }

  def jsonOfStatement(query: Query, resultDataContents: List[String]): String = {
    val a = Statement.apply(query, resultDataContents)
    jsonOf(a)
  }

} 

//设置package包名称以及导入依赖的类
package com.packt.masteringakka.bookstore.server

import io.netty.channel.ChannelHandler.Sharable
import unfiltered.netty.ServerErrorResponse
import unfiltered.request._
import java.util.Date
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import unfiltered.response._


@Sharable
object PretentCreditCardService extends unfiltered.netty.async.Plan with ServerErrorResponse{
  case class ChargeRequest(cardHolder:String, cardType:String, cardNumber:String, expiration:Date, amount:Double)
  case class ChargeResponse(confirmationCode:String)
  implicit val formats = Serialization.formats(NoTypeHints)
  
  def intent = {
    case req @ POST(Path(Seg("credit" :: "charge" :: Nil))) =>
      val body = Body.string(req)
      val request = read[ChargeRequest](body)
      val resp = write(ChargeResponse(java.util.UUID.randomUUID().toString))
      req.respond(JsonContent ~> ResponseString(resp))
  }
} 

//设置package包名称以及导入依赖的类
package com.packt.masteringakka.bookstore.common

import io.netty.channel.ChannelHandler.Sharable
import unfiltered.netty.ServerErrorResponse
import unfiltered.request._
import java.util.Date
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import unfiltered.response._


@Sharable
object PretentCreditCardService extends unfiltered.netty.async.Plan with ServerErrorResponse{
  case class ChargeRequest(cardHolder:String, cardType:String, cardNumber:String, expiration:Date, amount:Double)
  case class ChargeResponse(confirmationCode:String)
  implicit val formats = Serialization.formats(NoTypeHints)
  
  def intent = {
    case req @ POST(Path(Seg("credit" :: "charge" :: Nil))) =>
      val body = Body.string(req)
      val request = read[ChargeRequest](body)
      val resp = write(ChargeResponse(java.util.UUID.randomUUID().toString))
      req.respond(JsonContent ~> ResponseString(resp))
  }
} 

//设置package包名称以及导入依赖的类
package com.packt.masteringakka.bookstore.common

import akka.serialization.SerializerWithStringManifest
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import org.json4s.ext.EnumNameSerializer



class JsonSerializer extends SerializerWithStringManifest{
  implicit val formats = Serialization.formats(NoTypeHints)
  
  def toBinary(o:AnyRef):Array[Byte] = {
    val json = write(o)
    json.getBytes()
  }
  
  def fromBinary(bytes:Array[Byte], manifest:String):AnyRef = {
    val m = Manifest.classType[AnyRef](Class.forName(manifest))
    val json = new String(bytes, "utf8")
    read[AnyRef](json)(formats, m)
  }  
  
  def identifier:Int = 999
  def manifest(o:AnyRef):String = o.getClass.getName
} 

//设置package包名称以及导入依赖的类
package com.packt.masteringakka.bookstore.credit

import akka.actor.Props
import com.packt.masteringakka.bookstore.common.EntityActor
import java.util.Date
import dispatch._
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import com.packt.masteringakka.bookstore.common.EntityFieldsObject

object CreditTransactionStatus extends Enumeration{
  val Approved, Rejected = Value
}
case class CreditCardInfo(cardHolder:String, cardType:String, cardNumber:String, expiration:Date)
case class CreditCardTransactionFO(id:Int, cardInfo:CreditCardInfo, amount:Double, 
  status:CreditTransactionStatus.Value, confirmationCode:Option[String], createTs:Date, modifyTs:Date, deleted:Boolean = false) extends EntityFieldsObject[Int, CreditCardTransactionFO]{
  def assignId(id:Int) = {
    this.copy(id = id)
  }
  def markDeleted = this.copy(deleted = true)
}


  def chargeCard(info:CreditCardInfo, amount:Double) = { 
    val jsonBody = write(ChargeRequest(info.cardHolder, info.cardType, info.cardNumber, info.expiration, amount))
    val request = url(settings.creditChargeUrl) << jsonBody
    Http(request OK as.String).map(read[ChargeResponse])
  } 
} 

//设置package包名称以及导入依赖的类
package com.packt.masteringakka.bookstore.credit

import akka.actor.Props
import com.packt.masteringakka.bookstore.common.EntityActor
import java.util.Date

import com.packt.masteringakka.bookstore.common.EntityActor.FinishCreate
import dispatch._
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import com.packt.masteringakka.bookstore.common.EntityFieldsObject

object CreditTransactionStatus extends Enumeration{
  val Approved, Rejected = Value
}
case class CreditCardInfo(cardHolder:String, cardType:String, cardNumber:String, expiration:Date)
case class CreditCardTransactionFO(id:Int, cardInfo:CreditCardInfo, amount:Double, 
  status:CreditTransactionStatus.Value, confirmationCode:Option[String], createTs:Date, modifyTs:Date, deleted:Boolean = false) extends EntityFieldsObject[CreditCardTransactionFO]{
  def assignId(id:Int) = {
    this.copy(id = id)
  }
  def markDeleted = this.copy(deleted = true)
}


  def chargeCard(info:CreditCardInfo, amount:Double) = { 
    val jsonBody = write(ChargeRequest(info.cardHolder, info.cardType, info.cardNumber, info.expiration, amount))
    val request = url(settings.creditChargeUrl) << jsonBody
    Http(request OK as.String).map(read[ChargeResponse])
  } 
} 

//设置package包名称以及导入依赖的类
package etherchainml.core.util


import etherchainml.core.{JSONWrapper, Event, EtherChainTrade, Common}
import org.json4s
import org.json4s._
import org.json4s.native.JsonMethods._
import org.json4s.native.Serialization.{ read, write, writePretty }

import scala.concurrent.Future
import scalaj.http.Http
import scala.concurrent.ExecutionContext.Implicits.global


//TODO: Generalize functionality to EventPull
class HttpGetEtherChain(parseResponse: (json4s.JValue) => List[Event]) extends EventPull {

  def requestBatch(start: Long, end: Long):Future[List[Event]]={
    //s"https://etherchain.org/api/txs/0/100"
    val url = s"https://etherchain.org/api/txs/${start}/${end}"

    val baseRequest = Http(url)
    val response = Future {
      baseRequest.asString.body
    }
    val parsedResponse = response.map { r => parse(r) }

    parsedResponse.map{
      parseResponse

    }
  }
}


object HttpGetEtherChain {
  implicit val formats = DefaultFormats

  def apply(parseResponse: (json4s.JValue) => List[Event] = parseResponseJsonWrapper): HttpGetEtherChain =
    new HttpGetEtherChain(parseResponse)

  def parseResponseEtherChainTrade: (JValue) => List[EtherChainTrade] = {
    jsonResponse =>
      val data: JValue = jsonResponse \ "data"
      val txs: List[EtherChainTrade] = data.extract[List[EtherChainTrade]]
      txs
  }

  def parseResponseJsonWrapper: (json4s.JValue) => List[JSONWrapper] = {
    jsonResponse =>
      val data: JValue = jsonResponse \ "data"
      val txs: List[JSONWrapper] = data.extract[List[EtherChainTrade]].map(trade => JSONWrapper(
        Common.EtherChainTradeJsonSchema,
        trade.hash, Extraction.decompose(trade)
        //write(trade)
      ))
      txs
  }
} 

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

import com.example.models.SystemMessage
import com.rabbitmq.client.{Channel, Connection, ConnectionFactory}
import org.json4s.NoTypeHints
import org.json4s.native.Serialization
import org.json4s.native.Serialization.write

class Messenger {
  private val QUEUE_NAME: String = "places"
  implicit val formats = Serialization.formats(NoTypeHints)


  def send(message: String) : Unit = {
    val factory: ConnectionFactory = new ConnectionFactory
    factory.setHost("localhost")
    val connection: Connection = factory.newConnection
    val channel: Channel = connection.createChannel


    channel.queueDeclare(QUEUE_NAME, false, false, false, null)
    channel.basicPublish("", QUEUE_NAME, null, message.getBytes())


    channel.close()
    connection.close()
  }

  def send(message: SystemMessage): Unit = {
    send(write(message))
  }

}
object Messenger {
  def apply() = new Messenger()
} 

//设置package包名称以及导入依赖的类
package com.ovoenergy.kafka.serialization.json4s

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStreamReader, OutputStreamWriter}
import java.nio.charset.StandardCharsets

import com.ovoenergy.kafka.serialization.core._
import org.apache.kafka.common.serialization.{Deserializer => KafkaDeserializer, Serializer => KafkaSerializer}
import org.json4s.Formats
import org.json4s.native.Serialization.{read, write}

import scala.reflect.ClassTag
import scala.reflect.runtime.universe._

trait Json4sSerialization {

  def json4sSerializer[T <: AnyRef](implicit jsonFormats: Formats): KafkaSerializer[T] = serializer { (_, data) =>
    val bout = new ByteArrayOutputStream()
    val writer = new OutputStreamWriter(bout, StandardCharsets.UTF_8)

    // TODO Use scala-arm
    try {
      write(data, writer)
      writer.flush()
    } finally {
      writer.close()
    }
    bout.toByteArray
  }

  def json4sDeserializer[T: TypeTag](implicit jsonFormats: Formats): KafkaDeserializer[T] = deserializer { (_, data) =>
    val tt = implicitly[TypeTag[T]]
    implicit val cl = ClassTag[T](tt.mirror.runtimeClass(tt.tpe))
    read[T](new InputStreamReader(new ByteArrayInputStream(data), StandardCharsets.UTF_8))
  }

}