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

import javax.inject.{Inject, Provider, Singleton}

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import play.api.inject.{ApplicationLifecycle, Module}
import play.api.{Configuration, Environment}
import play.libs.Json

import scala.concurrent.Future

class CustomObjectMapperModule extends Module {

  def bindings(environment: Environment, configuration: Configuration) = Seq(
    bind[ObjectMapper].toProvider[ObjectMapperProvider].eagerly()
  )
}

@Singleton
class ObjectMapperProvider @Inject() (lifecycle: ApplicationLifecycle) extends Provider[ObjectMapper] {

  lazy val get : ObjectMapper = {
    val objectMapper = Json.newDefaultMapper()
    objectMapper.registerModule(DefaultScalaModule)

    Json.setObjectMapper(objectMapper)
    lifecycle.addStopHook { () =>
      Future.successful(Json.setObjectMapper(null))
    }
    objectMapper
  }
} 

//设置package包名称以及导入依赖的类
import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule



object JsonUtil {
  val mapper = new ObjectMapper() with ScalaObjectMapper
  mapper.registerModule(DefaultScalaModule)
  mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)

  def toJson(value: Map[Symbol, Any]): String = {
    toJson(value map { case (k,v) => k.name -> v})
  }

  def toJson(value: Any): String = {
    mapper.writeValueAsString(value)
  }

  def toMap[V](json:String)(implicit m: Manifest[V]) = fromJson[Map[String,V]](json)

  def fromJson[T](json: String)(implicit m : Manifest[T]): T = {
    mapper.readValue[T](json)
  }
}

object MarshallableImplicits {

  implicit class Unmarshallable(unMarshallMe: String) {
    def toMap: Map[String,Any] = JsonUtil.toMap(unMarshallMe)
    //def toMapOf[V]()(implicit m: Manifest[V]): Map[String,V] = JsonUtil.toMapOf[V](unMarshallMe)
    def fromJson[T]()(implicit m: Manifest[T]): T =  JsonUtil.fromJson[T](unMarshallMe)
  }

  implicit class Marshallable[T](marshallMe: T) {
    def toJson: String = JsonUtil.toJson(marshallMe)
  }
} 

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

import java.util.{Date, UUID}

import com.fasterxml.jackson.databind.{SerializationFeature, ObjectMapper}
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import tools._
import common._
import Tool._

class BaseBDBEntity[+Self <: BaseBDBEntity[Self]](tableName: String) extends BDBEntity(tableName) {
  def toJson: String = {
    BaseBDBEntity.map.writeValueAsString(this)
  }

  def fromJson(json: String): Self = {
    BaseBDBEntity.map.readValue(json, this.getClass).asInstanceOf[Self]
  }

  //????????????
  def changeUpdateBean(): Self = {
    fromJson(toJson)
  }

  override def queryById(id: String, fields: String*): Option[Self] = {
    super.queryById(id,fields:_*) map (_.asInstanceOf[Self])
  }
  override def queryByIds(idName: String, ids: List[Long], fields: String*): List[Self] = {
    super.queryByIds(idName,ids,fields:_*) map (_.asInstanceOf[Self])
  }

  //????????????
  override def queryPage(where: String, pageNum: Int, pageSize: Int, fields: String*):List[Self] = {
    val  list = super.queryPage(where, pageNum, pageSize,fields: _*)
     list map (_.asInstanceOf[Self])
  }

}

object BaseBDBEntity {
  private val map = new ObjectMapper() with ScalaObjectMapper
  map.registerModule(DefaultScalaModule)
  map.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false)
}

//??????
class LoanData(val id:String=UUID.randomUUID().toString,val Title:String="",val text:String="",val createTime:Date=new Date())extends BaseBDBEntity[LoanData]("LoanData") 

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

import com.fasterxml.jackson.databind.ObjectMapper

class GoogleChartWrapperSpec extends UnitSpec {

  "A chart row " should "contain number values" in {
    val row = ChartRow(Array(ChartValueNumber(1), ChartValueNumber(2)))
    val mapper = new ObjectMapper()
    mapper.writeValueAsString(row) shouldEqual """{"c":[{"v":1},{"v":2}]}"""
  }

  it should "also contain string values" in {
    val row = ChartRow(Array(ChartValueName("A"), ChartValueName("B")))
    val mapper = new ObjectMapper()
    mapper.writeValueAsString(row) shouldEqual """{"c":[{"v":"A"},{"v":"B"}]}"""
  }

  it should "also contain mixed values" in {
    val row = ChartRow(Array(ChartValueName("A"), ChartValueNumber(1)))
    val mapper = new ObjectMapper()
    mapper.writeValueAsString(row) shouldEqual """{"c":[{"v":"A"},{"v":1}]}"""
  }

  "A chart column " should "contain string labels" in {
    val col = ChartColumn("ID", "LABEL")
    val mapper = new ObjectMapper()
    mapper.writeValueAsString(col) shouldEqual """{"id":"ID","label":"LABEL","type":"string"}"""
  }

  it should "also contain numeric labels" in {
    val col = ChartColumn("ID", "LABEL", "number")
    val mapper = new ObjectMapper()
    mapper.writeValueAsString(col) shouldEqual """{"id":"ID","label":"LABEL","type":"number"}"""
  }

  "Chart data " should "contain columns and rows" in {
    val cols = Array(ChartColumn("n", "NAME"), ChartColumn("q", "QTY", "number"))
    val rows = Array(ChartRow(Array(ChartValueName("A"), ChartValueNumber(1))),ChartRow(Array(ChartValueName("B"), ChartValueNumber(2))))
    val data = ChartData(cols,rows)
    val mapper = new ObjectMapper()
    mapper.writeValueAsString(data) shouldEqual """{"cols":[{"id":"n","label":"NAME","type":"string"},{"id":"q","label":"QTY","type":"number"}],"rows":[{"c":[{"v":"A"},{"v":1}]},{"c":[{"v":"B"},{"v":2}]}]}"""
  }

} 

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

import javax.inject.Singleton

import com.fasterxml.jackson.annotation.JsonInclude
import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper, SerializationFeature}
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
import de.stema.pullrequests.dto.PullRequestDTO

import scala.reflect.ClassTag
import scala.util.{Failure, Success, Try}

@Singleton
class JsonObjectMapper {
  private lazy val mapper = new ObjectMapper with ScalaObjectMapper
  mapper.registerModule(DefaultScalaModule)
  mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
  mapper.configure(SerializationFeature.INDENT_OUTPUT, true)
  mapper.setSerializationInclusion(JsonInclude.Include.NON_EMPTY)
  mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)

  def getInstance[T](json: String)(implicit ct: ClassTag[T]): T =
    Try {
      mapper.readValue(json, ct.runtimeClass).asInstanceOf[T]
    } match {
      case Success(instance) => instance
      case Failure(e) => throw new IllegalStateException(s"Error during parsing of '$json'", e)
    }

  def getInstances[T](json: String)(implicit ct: ClassTag[T]): Seq[PullRequestDTO] =
    Try {
      mapper.readValue[Seq[PullRequestDTO]](json)
    } match {
      case Success(instances) => instances
      case Failure(e) => throw new IllegalStateException(s"Error during parsing of '$json'", e)
    }
} 

//设置package包名称以及导入依赖的类
// Copyright (c) 2017 Grier Forensics. All Rights Reserved.

package com.grierforensics.greatdane.connector

import java.net.URI
import java.nio.charset.StandardCharsets
import java.nio.file.{Files, Paths}

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.apache.commons.io.IOUtils
import org.apache.http.HttpHeaders
import org.apache.http.client.methods.HttpPost
import org.apache.http.entity.StringEntity
import org.apache.http.impl.client.HttpClients


object ProvisionUser {
  def main(args: Array[String]): Unit = {

    def die = {
      println("Usage: provision-user <email-address> [<certificate file>]")
      sys.exit(1)
    }
    val (emailAddress, certPem) = args.toList match {
      case email :: tail => tail match {
        case Nil => (email, "")
        case certFile :: Nil => (email, new String(Files.readAllBytes(Paths.get(certFile)), StandardCharsets.UTF_8))
        case _ => die
      }
      case _ => die
    }

    val client = HttpClients.createDefault()
    val uri = new URI(s"http://${Settings.Host}:${Settings.Port}/api/v1/user/$emailAddress")
    val post = new HttpPost(uri)
    post.addHeader(HttpHeaders.CONTENT_TYPE, "application/json")
    post.addHeader(HttpHeaders.AUTHORIZATION, Settings.ApiKey)
    println(post.toString)

    val req = ProvisionRequest(None, if (certPem.length > 0) Some(Seq(certPem)) else None)
    val mapper = new ObjectMapper().registerModule(DefaultScalaModule)
    val json = mapper.writeValueAsString(req)
    println(json)

    post.setEntity(new StringEntity(json))

    val resp = client.execute(post)
    try {
      val entity = resp.getEntity
      println(resp.getStatusLine.getStatusCode, resp.getStatusLine.getReasonPhrase)
      println(IOUtils.toString(entity.getContent, StandardCharsets.UTF_8))
    } finally {
      resp.close()
    }

  }
} 

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

import java.io.StringWriter

import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper

import scala.reflect.{ClassTag, Manifest}

object JsonUtil {

  val mapper = new ObjectMapper() with ScalaObjectMapper
  mapper.registerModule(DefaultScalaModule)
  mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)

  def toJson[T](data: T): String = {
    val out = new StringWriter()
    mapper.writeValue(out, data)
    out.toString
  }

  def fromJson[T: ClassTag](json: String)(implicit m: Manifest[T]): T =
    mapper.readValue[T](json)
} 

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

import java.util.concurrent.TimeUnit

import com.fasterxml.jackson.core.JsonFactory
import com.fasterxml.jackson.databind.ObjectMapper
import knot.msgpack.testValues.{JavaMedia, Media}
import org.msgpack.jackson.dataformat.MessagePackFactory
import org.openjdk.jmh.annotations._

@State(Scope.Benchmark)
@Measurement(timeUnit = TimeUnit.MILLISECONDS)
class MsgPackSerializeBench {

  val knotSer = new MsgPackSerializerFactory().get[Media]()
  val jacksonFactory = new MessagePackFactory()
  jacksonFactory.configure(JsonFactory.Feature.USE_THREAD_LOCAL_FOR_BUFFER_RECYCLING, false)
  val jacksonMapper = new ObjectMapper(jacksonFactory)

  val src = new Media().setup()
  val jsrc = JavaMedia.setup()
  var knotSize: Int = 0
  var jacksonSize: Int = 0

  @TearDown
  def teardown() = {
    println(s"knot:$knotSize")
    println(s"jackson:$jacksonSize")
  }

  @Benchmark
  def ser_knot() = {
    val ar = knotSer.serialize(src)
    knotSize = ar.length
  }

  @Benchmark
  def ser_jackson() = {
    val ar = jacksonMapper.writeValueAsBytes(jsrc)
    jacksonSize = ar.length
  }
} 

//设置package包名称以及导入依赖的类
package io.github.mijicd.prezi.modules

import com.fasterxml.jackson.databind.ObjectMapper
import com.google.inject.{Provides, Singleton}
import com.twitter.inject.TwitterModule
import io.github.mijicd.prezi.domain.Presentation

import scala.collection.JavaConverters._
import scala.io.BufferedSource

object DataSource extends TwitterModule with Managed with DateConversion {
  @Singleton
  @Provides
  def provideSource: Seq[Presentation] = {
    val resource = getClass.getResource("/prezis.json")

    using(scala.io.Source.fromURL(resource)) { source =>
      val items = loadFrom(source)
      sortSeq(items)
    }
  }

  private def loadFrom(source: BufferedSource): Seq[Presentation] = {
    val json = source.getLines().mkString
    val mapper = new ObjectMapper()
    val factory = mapper.getTypeFactory

    val items: java.util.List[Presentation] =
      mapper.readValue(json,
        factory.constructCollectionType(classOf[java.util.List[Presentation]],
          classOf[Presentation]))

    items.asScala
  }

  private def sortSeq(seq: Seq[Presentation]) = seq.sortWith(compare)

  private def compare(first: Presentation, second: Presentation): Boolean = {
    val firstDate = dateOf(first.createdAt)
    val secondDate = dateOf(second.createdAt)

    firstDate.isAfter(secondDate)
  }
} 

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

import scala.collection.JavaConverters._
import com.fasterxml.jackson.databind.{ ObjectMapper, JsonNode }
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import play.api.Play
import sys.process._

case class Truck (
  name: String,
  address: String,
  cuisine: List[String],
  url: String) {
  override def toString: String = s"""$name (${cuisine.mkString(",")}) @ $address [$url]"""
}

object Truck {
  def newObjectMapper: ObjectMapper = {
    val m = new ObjectMapper()
    m.registerModule(DefaultScalaModule)
  }

  private[Truck] lazy val mapper = newObjectMapper

  val defaultLat = Play.maybeApplication.flatMap(_.configuration.getDouble("foodtrucks.latitude")).getOrElse(0.0)
  val defaultLon = Play.maybeApplication.flatMap(_.configuration.getDouble("foodtrucks.longitude")).getOrElse(0.0)

  def apply(jsonNode: JsonNode): Truck = {
    val info = jsonNode.get("obj")

    Truck(
      info.get("name").asText,
      info.get("last_seen").get("address").asText,
      info.get("cuisine").elements.asScala.map(_.asText).toList,
      info.get("url").asText
    )
  }

  def listAll(lat: Double = defaultLat, lon: Double = defaultLon): List[Truck] = {
    val result = Seq("curl", "-XGET", s"http://nyctruckfood.com/api/trucks/search?q=$lat,$lon").!!
    mapper.readTree(result).elements.asScala.map(Truck(_)).toList
  }
} 

//设置package包名称以及导入依赖的类
package mesosphere.marathon.api

import java.io.StringWriter
import java.util.concurrent.TimeUnit
import javax.servlet.http.HttpServletRequest
import javax.ws.rs._
import javax.ws.rs.core.{ Context, MediaType, Response }

import com.codahale.metrics.{ MetricFilter, MetricRegistry }
import com.codahale.metrics.annotation.Timed
import com.codahale.metrics.json.MetricsModule
import com.fasterxml.jackson.databind.ObjectMapper
import com.google.inject.Inject
import mesosphere.marathon.MarathonConf
import mesosphere.marathon.io.IO
import mesosphere.marathon.plugin.auth.AuthorizedResource.SystemConfig
import mesosphere.marathon.plugin.auth.{ Authenticator, Authorizer, ViewResource }


@Path("")
@Consumes(Array(MediaType.APPLICATION_JSON))
@Produces(Array(MarathonMediaType.PREFERRED_APPLICATION_JSON))
class SystemResource @Inject() (metrics: MetricRegistry, val config: MarathonConf)(implicit
  val authenticator: Authenticator,
    val authorizer: Authorizer) extends RestResource with AuthResource {

  private[this] lazy val mapper = new ObjectMapper().registerModule(
    new MetricsModule(TimeUnit.SECONDS, TimeUnit.SECONDS, false, MetricFilter.ALL)
  )

  @GET
  @Path("ping")
  @Timed
  def ping(@Context req: HttpServletRequest): Response = authenticated(req) { implicit identity =>
    withAuthorization(ViewResource, SystemConfig){
      ok("pong")
    }
  }

  @GET
  @Path("metrics")
  @Timed
  def metrics(@Context req: HttpServletRequest): Response = authenticated(req) { implicit identity =>
    withAuthorization(ViewResource, SystemConfig){
      IO.using(new StringWriter()) { writer =>
        mapper.writer().writeValue(writer, metrics)
        ok(writer.toString)
      }
    }
  }
} 

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

import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper

object JsonMarshaller {
  val mapper = new ObjectMapper with ScalaObjectMapper
  mapper.registerModule(DefaultScalaModule)
  mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)

  def toJson(value: Any): String = {
    mapper.writeValueAsString(value)
  }

  def toMap[V](json:String)(implicit m: Manifest[V]) = fromJson[Map[String,V]](json)

  def fromJson[T](json: String)(implicit m : Manifest[T]): T = {
    mapper.readValue[T](json)
  }
}

object MarshallableImplicits {

  implicit class Unmarshallable(unMarshallMe: String) {
    def toMapOf[V]()(implicit m: Manifest[V]): Map[String,V] = JsonMarshaller.toMap[V](unMarshallMe)
    def fromJson[T]()(implicit m: Manifest[T]): T =  JsonMarshaller.fromJson[T](unMarshallMe)
  }

  implicit class Marshallable[T](marshallMe: T) {
    def toJson: String = JsonMarshaller.toJson(marshallMe)
  }
} 

//设置package包名称以及导入依赖的类
package org.styx.mongo

import com.fasterxml.jackson.databind.ObjectMapper
import org.bson.RawBsonDocument
import org.mongodb.scala.{Completed, MongoCollection}
import org.styx.handler.EventHandler
import org.styx.handler.EventHandler.{SuccessfullyHandled, UnsuccessfullyHandled}
import org.styx.model.Event
import org.styx.state.State
import org.styx.state.State.AggregationId

import scala.concurrent.{ExecutionContext, Future}


trait MongoDBEventHandler[S <: State] extends EventHandler[S] {
  val collection: MongoCollection[MongoDBEvent]
  val objectMapper: ObjectMapper

  override def handle(aggregationId: AggregationId, event: Event[S], actualState: S)(implicit executionContext: ExecutionContext): Future[EventHandler.HandleStatus[S]] = {
    collection
      .insertOne {
        MongoDBEvent(
          eventType = event.getClass.getSimpleName,
          aggregationId = aggregationId,
          data = RawBsonDocument.parse(objectMapper.writeValueAsString(event.data)),
          eventDate = event.eventDate,
          revision = event.revision
        )
      }.head
      .map { case Completed() => SuccessfullyHandled(event) }
      .recover { case e: Exception => UnsuccessfullyHandled(e, event) }
  }
}

object MongoDBEventHandler {
  def apply[S <: State](mongoCollection: MongoCollection[MongoDBEvent], mapper: ObjectMapper): MongoDBEventHandler[S] = new MongoDBEventHandler[S](){
    override val collection: MongoCollection[MongoDBEvent] = mongoCollection
    override val objectMapper: ObjectMapper = mapper
  }
} 

//设置package包名称以及导入依赖的类
package org.styx.mongo

import com.fasterxml.jackson.databind.ObjectMapper
import org.mongodb.scala.MongoCollection
import org.mongodb.scala.bson.Document
import org.styx.handler.EventFetcher
import org.styx.model.Event
import org.styx.state.State
import org.styx.state.State.AggregationId

import scala.collection.Seq
import scala.concurrent.{ExecutionContext, Future}


trait MongoDBEventFetcher[S <: State] extends EventFetcher[S] {
  val collection: MongoCollection[MongoDBEvent]
  val objectMapper: ObjectMapper
  val converter: (MongoDBEvent => Event[S])

  override def get(aggregationId: AggregationId)(implicit executionContext: ExecutionContext): Future[Seq[Event[S]]] = {
    val eventualEvents: Future[Seq[MongoDBEvent]] = collection
      .find(Document("aggregationId" -> aggregationId))
      .sort(Document("eventDate" -> 1))
      .collect()
      .head()

    eventualEvents.map[Seq[Event[S]]] {
      events => events.map(e => converter(e)).distinct
    }
  }
}

object MongoDBEventFetcher {
  def apply[S <: State](col: MongoCollection[MongoDBEvent],
                        mapper: ObjectMapper,
                        eventConverter: (MongoDBEvent => Event[S])): MongoDBEventFetcher[S] = new MongoDBEventFetcher[S]() {
    override val collection: MongoCollection[MongoDBEvent] = col
    override val objectMapper: ObjectMapper = mapper
    override val converter: (MongoDBEvent) => Event[S] = eventConverter
  }
} 

//设置package包名称以及导入依赖的类
package org.styx.mongo

import java.util.concurrent.Executors

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.styx.bank.example.events._
import org.styx.bank.example.state.BankAccount
import org.styx.handler.{EventFetcher, EventHandler}
import org.styx.model.Event

import scala.concurrent.ExecutionContext


object BankAccountEventHandler {
  implicit val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(30))

  val mapper = new ObjectMapper
  mapper.registerModule(DefaultScalaModule)

  def converter: MongoDBEvent => Event[BankAccount] = { mongoDBEvent =>
    val event = mongoDBEvent match {
      case MongoDBEvent("BankAccountCreated", eventDate, version, _, _) => BankAccountCreated(version, eventDate)
      case MongoDBEvent("BankAccountClosed", eventDate, version, _, _) => BankAccountClosed(version, eventDate)
      case MongoDBEvent("OwnerChanged", eventDate, version, _, _) => OwnerChanged(version, eventDate)
      case MongoDBEvent("DepositPerformed", eventDate, version, _, _) => DepositPerformed(version, eventDate)
      case MongoDBEvent("WithdrawalPerformed", eventDate, version, _, _) => WithdrawalPerformed(version, eventDate)
    }

    event.data = mapper.readValue(mongoDBEvent.data.toJson(), classOf[Map[String, Any]])
    event
  }

  implicit val eventHandler: EventHandler[BankAccount] with EventFetcher[BankAccount] = MongoDBEventHandlerFetcher(MongoD.collection, mapper, converter)
} 

//设置package包名称以及导入依赖的类
package com.groupon.dse.kafka.topic

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule


object TopicUtils {

  def extractTopicListFromJson(json: String): List[TopicAndMetadata] = {
    val mapper = new ObjectMapper()
    mapper.registerModule(DefaultScalaModule)
    //The top level "topics" key is mapped to a list of [[TopicAndMetadata]] objects
    val contentMap = mapper.readValue(json, classOf[Map[String, List[Map[String, AnyRef]]]])

    val topics = contentMap.getOrElse("topics", List.empty)

    val topicList = topics.map(topicMap => {
      val topic = topicMap.getOrElse("name", "")
      val metadata = topicMap.getOrElse("metadata", Map.empty)
      TopicAndMetadata(topic.toString, metadata.asInstanceOf[Map[String, AnyRef]])
    }).filterNot(_.name.isEmpty)
    topicList
  }
} 

//设置package包名称以及导入依赖的类
package info.matsumana.flink

import java.util.regex.Pattern
import java.util.{HashMap, Properties}

import com.fasterxml.jackson.databind.ObjectMapper
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010
import org.apache.flink.streaming.connectors.twitter.TwitterSource
import org.apache.flink.streaming.util.serialization.SimpleStringSchema

object TweetCollect {

  val DELETED_TWEET_PATTERN = Pattern.compile("""^\{"delete":\{""")

  // ????
  val TARGET_TWEET_PATTERN = Pattern.compile("^.*[\u3040-\u3096]+.*$")

  val mapper = new ObjectMapper()

  def main(args: Array[String]): Unit = {
    val params = ParameterTool.fromPropertiesFile(args(0))

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.enableCheckpointing(60000)

    // source (twitter)
    val twitterProps = new Properties()
    twitterProps.setProperty(TwitterSource.CONSUMER_KEY, params.get("consumer_key", ""))
    twitterProps.setProperty(TwitterSource.CONSUMER_SECRET, params.get("consumer_secret", ""))
    twitterProps.setProperty(TwitterSource.TOKEN, params.get("token", ""))
    twitterProps.setProperty(TwitterSource.TOKEN_SECRET, params.get("token_secret", ""))
    val sourceStream = env.addSource(new TwitterSource(twitterProps))

    // sink (kafka)
    val kafkaProps = new Properties()
    kafkaProps.setProperty("bootstrap.servers", params.get("bootstrap.servers", "localhost:9092"))
    val topic = params.get("topic", "twitter")
    val sink = new FlinkKafkaProducer010[String](
      topic,
      new SimpleStringSchema,
      kafkaProps)

    // stream processing
    sourceStream
      .filter(!DELETED_TWEET_PATTERN.matcher(_).matches())
      .map(mapper.readValue(_, classOf[HashMap[String, Object]]))
      .filter(m => {
        val text = m.get("text")
        text != null && TARGET_TWEET_PATTERN.matcher(String.valueOf(text)).matches()
      })
      .map(mapper.writeValueAsString(_))
      .addSink(sink)

    env.execute("TweetCollect")
  }
} 

//设置package包名称以及导入依赖的类
package moe.pizza.zkapi

import java.net.URI

import com.fasterxml.jackson.databind.{DeserializationConfig, ObjectMapper}
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import moe.pizza.zkapi.zkillboard.Killmail
import org.eclipse.jetty.websocket.api.{WebSocketListener, Session, WebSocketAdapter}
import org.eclipse.jetty.websocket.client.WebSocketClient
import org.slf4j.{LoggerFactory, Logger}

import scala.concurrent.duration._



object WebsocketFeed {
  def main (args: Array[String]) {
    val logger = LoggerFactory.getLogger("main")
    val ws = createClient("ws://ws.eve-kill.net/kills", {s: Killmail => logger.info(s.toString)})
    ws.start()
    Thread.sleep(10.seconds.toMillis)
  }


  def createClient(url: String, onrecv: (Killmail => Unit)): WebSocketClient = {

    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)

    val logger = LoggerFactory.getLogger("websockets")

    class EKWSClientSocket(callback: (Killmail => Unit)) extends WebSocketAdapter {

      override def onWebSocketText(message: String): Unit = {
        if (message.startsWith("{")) {
          val kill = OM.readValue(message, classOf[zkillboard.Killmail])
          callback(kill)
        }
      }
    }

    val ws = new WebSocketClient()
    ws.start()
    ws.connect(new EKWSClientSocket(onrecv), URI.create(url))
    ws
  }

} 

//设置package包名称以及导入依赖的类
package moe.pizza.zkapi

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import moe.pizza.zkapi.StatsTypes.SuperPilot
import org.scalatest.{FlatSpec, MustMatchers}

import scala.io.Source

class RedisQSpec extends FlatSpec with MustMatchers {

  "redisq responses" should "be parsed" in {
    val json = Source.fromInputStream(getClass.getResourceAsStream("/redisq.json")).getLines().mkString("\n")
    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)
    val r = OM.readValue[RedisQTypes.RedisQResponse](json, classOf[RedisQTypes.RedisQResponse])
    r.payload.isDefined must equal(true)
    r.payload.get.killID must equal(51738685)
  }

  "redisq responses 2" should "be parsed" in {
    val json = Source.fromInputStream(getClass.getResourceAsStream("/redisq2.json")).getLines().mkString("\n")
    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)
    val r = OM.readValue[RedisQTypes.RedisQResponse](json, classOf[RedisQTypes.RedisQResponse])
    r.payload.isDefined must equal(true)
    r.payload.get.killID must equal(51738559)
  }

  "redisq responses 3" should "be parsed" in {
    val json = Source.fromInputStream(getClass.getResourceAsStream("/redisq3.json")).getLines().mkString("\n")
    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)
    val r = OM.readValue[RedisQTypes.RedisQResponse](json, classOf[RedisQTypes.RedisQResponse])
    r.payload.isDefined must equal(false)
  }

  "redisq responses 4, attacker with faction" should "be parsed" in {
    val json = Source.fromInputStream(getClass.getResourceAsStream("/redisq4_attackerwithfaction.json")).getLines().mkString("\n")
    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)
    val r = OM.readValue[RedisQTypes.RedisQResponse](json, classOf[RedisQTypes.RedisQResponse])
    r.payload.isDefined must equal(true)
    val factionedattacker = r.payload.get.killmail.attackers.find(_.faction.isDefined)
    factionedattacker.get.faction.get.name must equal("Blood Raider Covenant")
  }

  "redisq responses 5, victim with faction" should "be parsed" in {
    val json = Source.fromInputStream(getClass.getResourceAsStream("/redisq5_victimwithfaction.json")).getLines().mkString("\n")
    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)
    val r = OM.readValue[RedisQTypes.RedisQResponse](json, classOf[RedisQTypes.RedisQResponse])
    r.payload.isDefined must equal(true)
    r.payload.get.killmail.victim.faction.get.name must equal("Gallente Federation")
  }



} 

//设置package包名称以及导入依赖的类
package moe.pizza.zkapi

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import moe.pizza.zkapi.StatsTypes.SuperPilot
import org.scalatest.{MustMatchers, FlatSpec}

import scala.io.Source


class StatsSpec extends FlatSpec with MustMatchers {

  "alliance stats" should "be parsed" in {
    val json = Source.fromInputStream(getClass.getResourceAsStream("/alliancestats.json")).getLines().mkString("\n")
    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)
    val r = OM.readValue[StatsTypes.AllianceInfo](json, classOf[StatsTypes.AllianceInfo])
    val superpilots = r.supers.supercarriers.data ++ r.supers.titans.data
    superpilots.size must equal(199)
    superpilots.head must equal(new SuperPilot(105,93260759,"Doctor Chalke"))
  }

  "alliance stats 2" should "be parsed" in {
    val json = Source.fromInputStream(getClass.getResourceAsStream("/alliancestats2.json")).getLines().mkString("\n")
    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)
    val r = OM.readValue[StatsTypes.AllianceInfo](json, classOf[StatsTypes.AllianceInfo])
    val superpilots = r.getSupers ++ r.getTitans
    superpilots.isEmpty must equal(true)
  }

  "alliance stats 3" should "be parsed" in {
    val json = Source.fromInputStream(getClass.getResourceAsStream("/alliancestats3.json")).getLines().mkString("\n")
    val OM = new ObjectMapper()
    OM.registerModule(DefaultScalaModule)
    val r = OM.readValue[StatsTypes.AllianceInfo](json, classOf[StatsTypes.AllianceInfo])
    val superpilots = r.getSupers ++ r.getTitans
    println(superpilots)
  }

}