//设置package包名称以及导入依赖的类
package org.dsa.iot.scala.netty

import java.util.Map

import collection.JavaConverters.mapAsScalaMapConverter
import util.control.NonFatal

import org.dsa.iot.dslink.provider.HttpProvider
import org.dsa.iot.dslink.util.URLInfo
import org.dsa.iot.dslink.util.http.HttpResp
import org.dsa.iot.shared.SharedObjects

import io.netty.bootstrap.Bootstrap
import io.netty.channel._
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioSocketChannel
import io.netty.handler.codec.http._
import io.netty.handler.ssl.SslContextBuilder
import io.netty.handler.ssl.util.InsecureTrustManagerFactory
import io.netty.util.CharsetUtil


private[netty] class HttpHandler extends SimpleChannelInboundHandler[Object] {
  private val content = new StringBuffer
  private var status: HttpResponseStatus = null
  private var t: Throwable = null

  protected def channelRead0(ctx: ChannelHandlerContext, msg: Object) = {
    if (msg.isInstanceOf[HttpResponse])
      status = msg.asInstanceOf[HttpResponse].getStatus
    if (msg.isInstanceOf[HttpContent]) {
      val buf = msg.asInstanceOf[HttpContent].content
      content.append(buf.toString(CharsetUtil.UTF_8))
    }
    if (msg.isInstanceOf[LastHttpContent])
      ctx.close
  }

  override def exceptionCaught(ctx: ChannelHandlerContext, t: Throwable) = {
    this.t = t
    ctx.close
  }

  def getThrowable = t

  def getStatus = status

  def getContent = content.toString
} 

//设置package包名称以及导入依赖的类
package com.microsoft.azure.iot.kafka.connect.sink

import java.util.Map

import com.microsoft.azure.sdk.iot.service.DeliveryAcknowledgement
import org.apache.kafka.common.config.ConfigDef.{Importance, Type, Width}
import org.apache.kafka.common.config.{AbstractConfig, ConfigDef}

object IotHubSinkConfig {

  val IotHubConnectionString    = "IotHub.ConnectionString"
  val IotHubMessageDeliveryAcknowledgement = "IotHub.MessageDeliveryAcknowledgement"

  private val IotHubConnectionStringDoc =
    """IoT Hub ConnectionString. (see "IoT Hub" >> your hub >> "Shared access policies" >> "service" >> """ +
      """"Connection string")"""
  private val IotHubMessageDeliveryAcknowledgementDoc = "The type of delivery acknowledgement for a C2D message. " +
    "Valid values are None, Full, NegativeOnly, PositiveOnly"
  private val iotConfigGroup = "Azure IoT Hub"
  private val validDeliveryAcknowledgementString = ConfigDef.ValidString.in(
    DeliveryAcknowledgement.None.toString,
    DeliveryAcknowledgement.Full.toString,
    DeliveryAcknowledgement.PositiveOnly.toString,
    DeliveryAcknowledgement.NegativeOnly.toString)

  lazy val configDef = new ConfigDef()
    .define(IotHubConnectionString, Type.STRING, Importance.HIGH, IotHubConnectionStringDoc, iotConfigGroup, 1,
      Width.MEDIUM, "IoT Hub Connection String")
    .define(IotHubMessageDeliveryAcknowledgement, Type.STRING, DeliveryAcknowledgement.None.toString,
      validDeliveryAcknowledgementString, Importance.HIGH, IotHubMessageDeliveryAcknowledgementDoc, iotConfigGroup, 1,
      Width.MEDIUM, "Delivery acknowledgement")

  def getConfig(configValues: Map[String, String]): IotHubSinkConfig = {
    new IotHubSinkConfig(configDef, configValues)
  }
}

class IotHubSinkConfig(configDef: ConfigDef, configValues: Map[String, String])
  extends AbstractConfig(configDef, configValues) 

//设置package包名称以及导入依赖的类
// Copyright (c) Microsoft. All rights reserved.

package com.microsoft.azure.iot.kafka.connect.source

import java.util.{Collections, Map}

import com.typesafe.scalalogging.LazyLogging
import org.apache.kafka.connect.data.Struct
import org.apache.kafka.connect.errors.ConnectException
import org.apache.kafka.connect.source.SourceRecord

import scala.collection.mutable.ListBuffer
import scala.util.control.NonFatal

class IotHubPartitionSource(val dataReceiver: DataReceiver,
    val partition: String,
    val topic: String,
    val batchSize: Int,
    val sourcePartition: Map[String, String])
  extends LazyLogging
    with JsonSerialization {

  def getRecords: List[SourceRecord] = {

    logger.debug(s"Polling for data from Partition $partition")
    val list = ListBuffer.empty[SourceRecord]
    try {
      val messages: Iterable[IotMessage] = this.dataReceiver.receiveData(batchSize)

      if (messages.isEmpty) {
        logger.debug(s"Finished processing all messages from partition ${this.partition}")
      } else {
        logger.debug(s"Received ${messages.size} messages from partition ${this.partition} " +
          s"(requested $batchSize batch)")

        for (msg: IotMessage <- messages) {

          val kafkaMessage: Struct = IotMessageConverter.getIotMessageStruct(msg)
          val sourceOffset = Collections.singletonMap("EventHubOffset",
            kafkaMessage.getString(IotMessageConverter.offsetKey))
          val sourceRecord = new SourceRecord(sourcePartition, sourceOffset, this.topic, kafkaMessage.schema(),
            kafkaMessage)
          list += sourceRecord
        }
      }
    } catch {
      case NonFatal(e) =>
        val errorMsg = s"Error while getting SourceRecords for partition ${this.partition}. " +
          s"Exception - ${e.toString} Stack trace - ${e.printStackTrace()}"
        logger.error(errorMsg)
        throw new ConnectException(errorMsg, e)
    }
    logger.debug(s"Obtained ${list.length} SourceRecords from IotHub")
    list.toList
  }
} 

//设置package包名称以及导入依赖的类
package cores.traceable.internal

import java.util.Map

import org.slf4j.MDC

import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
import scala.util.control.Exception.{Catch, ultimately}


private[internal] final class TraceableExecutionContext(mdcContext: Map[String, String], delegate: ExecutionContext) extends ExecutionContextExecutor {
  def execute(runnable: Runnable): Unit = delegate.execute(new Runnable {
    def run() {
      val oldMDCContext = MDC.getCopyOfContextMap
      setContextMap(mdcContext)
      withResetContext(oldMDCContext) {
        runnable.run()
      }
    }
  })

  private[this] def setContextMap(context: Map[String, String]) {
    Option(context) match {
      case Some(c) => MDC.setContextMap(c)
      case None => MDC.clear()
    }
  }

  private def withResetContext(context: Map[String, String]): Catch[Unit] = {
    ultimately[Unit] {
      setContextMap(context)
    }
  }

  def reportFailure(t: Throwable): Unit = delegate.reportFailure(t)
} 

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

import java.io.PrintWriter
import java.net.ServerSocket
import java.util
import java.util.Map

import com.fasterxml.jackson.core.`type`._
import com.fasterxml.jackson.databind.ObjectMapper

import scala.io.Source

object JsonToSocketSimulator extends App {

  //val incomingFile = "samples/dronestream_withcontrols.jsons"
  //val incomingFile = "samples/flight_at_home.jsons"
  val incomingFile = "samples/parque_berlin_5.jsons"

  val server = new ServerSocket(7891)
  println(s"Server Address: ${server.getLocalSocketAddress}")
  val connection  = server.accept
  println(s"Connection from: ${connection.getRemoteSocketAddress}")
  val out = new PrintWriter(connection.getOutputStream)

  init()

  close()

  def init() = {
    var jsonMap = new util.HashMap[String, Object]()
    val mapper = new ObjectMapper
    var prevTimestamp = Long.MaxValue
    var currentTimestamp = Long.MaxValue

    for (line <- Source.fromFile(incomingFile).getLines.zipWithIndex) {
      jsonMap = mapper.readValue(line._1, new TypeReference[Map[String, Object]]() {})
      println(s"LINE ${line._2} = gcs_timestamp_ms: ${jsonMap.get("gcs_timestamp_ms")}")
      currentTimestamp = jsonMap.get("gcs_timestamp_ms").asInstanceOf[Long]
      if (prevTimestamp != Long.MaxValue) {
        //println(s"Waiting ${currentTimestamp - prevTimestamp} ms")
        Thread.sleep(currentTimestamp - prevTimestamp)
      }
      prevTimestamp = currentTimestamp
      write(line._1)
    }
  }

  def write(line: String) = {
    out.write(s"$line${System.lineSeparator}")
    out.flush
  }

  def close() = {
    out.flush
    server.close
  }

} 

//设置package包名称以及导入依赖的类
package ch.loewenfels.jira.plugin.customfields

import java.util.Map
import com.atlassian.jira.issue.Issue
import com.atlassian.jira.issue.fields.CustomField
import com.atlassian.jira.issue.customfields.impl.CascadingSelectCFType
import com.atlassian.jira.issue.customfields.persistence.CustomFieldValuePersister
import com.atlassian.jira.issue.customfields.manager.GenericConfigManager
import com.atlassian.jira.issue.fields.rest.json.beans.JiraBaseUrls
import com.atlassian.jira.issue.customfields.manager.OptionsManager
import com.atlassian.jira.issue.customfields.option.{ Option => JiraOption }
import com.atlassian.jira.issue.CustomFieldManager

class VertecProjectCascadeFieldType(optionsManager: OptionsManager, customFieldValuePersister: CustomFieldValuePersister, genericConfigManager: GenericConfigManager, jiraBaseUrls: JiraBaseUrls, customFieldProvider: CustomFieldProvider)
  extends CascadingSelectCFType(optionsManager: OptionsManager, customFieldValuePersister: CustomFieldValuePersister, genericConfigManager: GenericConfigManager, jiraBaseUrls: JiraBaseUrls) {

  override def getValueFromIssue(field: CustomField, issue: Issue): Map[String, JiraOption] = {
    Option(super.getValueFromIssue(field, issue))
      .orElse(getValueFromParentIssue(field, issue))
      .orElse(getValueFromEpos(field, issue))
      .orNull
  }

  private lazy val epicLinkCustomField = customFieldProvider.find(CustomFieldProvider.AgileEpicLinkKey)

  private def getValueFromEpos(field: CustomField, issue: Issue): Option[Map[String, JiraOption]] = {
    for {
      c <- epicLinkCustomField
      eposIssue <- Option(issue.getCustomFieldValue(c)) if eposIssue.isInstanceOf[Issue]
      v <- Option(getValueFromIssue(field, eposIssue.asInstanceOf[Issue]))
    } yield v
  }

  private def getValueFromParentIssue(field: CustomField, issue: Issue): Option[Map[String, JiraOption]] = {
    Option(issue.getParentObject) match {
      case Some(i) => Option(getValueFromIssue(field, i))
      case _ => None
    }
  }

} 

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

import java.util.concurrent.ConcurrentHashMap
import java.util.Map
import java.lang.Integer

class Struct {
  
  var v = List[List[Int]]()
  val map: Map[Long, Integer] = new ConcurrentHashMap()
  
  def add(p:List[Int]){
    v = v ++ List(p)
  }
  def size() = {
    len() / 4
  }
  private def len() = {
    v.collect({case x: List[Int] => x.length}).sum
  }
  def pixel(x: Long, y:Long) : Int = {
    val lim = Math.floor(Math.sqrt(size()))
    require(x<lim, y<lim)
    require(x>=0, y>=0)
    //println(s"((${y})*${lim.toLong-1})+(${x})")
    get(((y)*lim.toLong)+(x))
  }
  def set(x:Long, y:Long, v:Int) = {
    val lim = Math.floor(Math.sqrt(size()))
    map.put(((y)*lim.toLong)+(x), v)
  }
  def get(i:Long):Int={
    // 4 - 32 bits
    require(i<size())
    require(i>=0)
    var e = map.get(i);
    if(e == null){
      
      val b = {
        var fin = List[Int]()
        val start = i*4
        var st = start
        while(st<start+4){
          fin = fin ++ List( v( (st / Int.MaxValue).toInt )((st % Int.MaxValue).toInt) )
          st = st + 1
          //println(st)
        }
        fin
      }
      e = Bits.getInt(b)
      map.put(i, e)
    }
    e
  }
} 

//设置package包名称以及导入依赖的类
package com.svds.kafka.connect.opentsdb

import java.util.{List, Map}

import com.typesafe.scalalogging.LazyLogging
import org.apache.kafka.common.config.ConfigDef
import org.apache.kafka.connect.errors.ConnectException
import org.apache.kafka.connect.sink.SinkConnector

import scala.collection.JavaConversions._
import scala.collection.mutable

class OpenTsdbSinkConnector extends SinkConnector with LazyLogging {

  private val configProps: Map[String, String] = mutable.Map[String, String]()

  override def config: ConfigDef = {
    logger.debug("config")
    OpenTsdbConnectorConfig.config
  }

  @throws(classOf[ConnectException])
  override def start(props: Map[String, String]) = {
    logger.debug(s"start(props: ${props})")
    this.configProps.putAll(props)
  }

  override def taskClass: Class[OpenTsdbSinkTask] = {
    classOf[OpenTsdbSinkTask]
  }

  def taskConfigs(maxTasks: Int): List[Map[String, String]] = {
    logger.debug(s"taskConfigs(), this.configProps: ${this.configProps}")
    val range = (1 to maxTasks).toBuffer
    val configs = range map { _ : Int =>
      this.configProps
    }
    logger.debug(s"taskConfigs(), configs: ${configs}")
    configs
  }

  override def stop = {}

  override def version = "0.0.1"
}