//设置package包名称以及导入依赖的类
import com.datastax.spark.connector._
import play.api.libs.json.Json
import java.io.{ByteArrayOutputStream, ByteArrayInputStream}
import java.util.zip.{GZIPOutputStream, GZIPInputStream}
import PreProcessingConfig._

case class VT_results_by_service_name_class(service_name: String, sha256: String)
case class VT_results_by_sha256_class(sha256: String, service_name: String, results: Array[Byte] )
case class VT_join_results_class(sha256: String, service_name: String, results: String)
case class VT_sample_signatures_initial_seq_rdd_class(sha256: String, seq_results: Seq[String])
case class VT_sample_signatures_final_array_rdd_class(sha256:String, array_results:Array[Double])

def unzip(x: Array[Byte]) : String = {		
    val inputStream = new GZIPInputStream(new ByteArrayInputStream(x))
    val output = scala.io.Source.fromInputStream(inputStream).mkString
    return output
}
def deleteNumberInSampleSignatures(x: String): Boolean = {
    val regex = "[0-9]".r
    return regex.findFirstIn(x).isEmpty
}

val VT_results_by_service_name_meta = sc.cassandraTable[VT_results_by_service_name_class](keyspace,service_name_table).where("service_name=?","virustotal")
val VT_results_by_service_name_rdd = VT_results_by_service_name_meta.keyBy(x=> (x.sha256,x.service_name))
val VT_results_by_sha256_meta = sc.cassandraTable[VT_results_by_sha256_class](keyspace,sha256_table)
val VT_results_by_sha256_rdd = VT_results_by_sha256_meta.keyBy(x => (x.sha256,x.service_name))
val VT_join_results = VT_results_by_service_name_rdd.join(VT_results_by_sha256_rdd).map(x => (new VT_join_results_class(x._1._1,x._1._2, unzip(x._2._2.results)))).distinct().cache()
val sample_signatures_rdd = VT_join_results.flatMap(x=>Json.parse(x.results) \ "scans" \\ "result").map(x=>Json.stringify(x)).filter( x=> !(x == "null"))
val sample_signatures_split_rdd = sample_signatures_rdd.flatMap(x=>x.replaceAll("""["]""","").replaceAll("""\![a-zA-Z0-9\s\+]+""","").replaceAll("""@[a-zA-Z0-9\s\+]+""","").replaceAll("""~[a-zA-Z0-9\s\+]+""","").replaceAll("""[\(|\[|{][a-zA-Z0-9\s\+]*[\)|\]|}]""","").replaceAll("""(\.|\!|\:|\_|\-|\\|/|\[|\])"""," ").split(" ")).filter(x=>(x.size>3)).filter(x=>deleteNumberInSampleSignatures(x)).map(x=>x.toLowerCase())
val signatures_prefix_rdd = sc.textFile(VT_signatures_prefix_suffix_file).map(x=>x.toLowerCase())
val family_signatures_subtract_rdd = sample_signatures_split_rdd.subtract(signatures_prefix_rdd)
val family_signatures_sorted_rdd = sc.parallelize(family_signatures_subtract_rdd.countByValue().toSeq).filter(x=>(x._2>50)).sortBy(x=>x._2,false)
val family_signatures_list = family_signatures_sorted_rdd.keys.collect().toList
val VT_sample_signatures_rdd = VT_join_results.map(x=>(x.sha256,(Json.parse(x.results) \ "scans" \\ "result").map(_.toString).filter( s => !(s== "null")).flatMap(x=>x.replaceAll("""["]""","").replaceAll("""\![a-zA-Z0-9\s\+]+""","").replaceAll("""@[a-zA-Z0-9\s\+]+""","").replaceAll("""~[a-zA-Z0-9\s\+]+""","").replaceAll("""[\(|\[|{][a-zA-Z0-9\s\+]*[\)|\]|}]""","").replaceAll("""(\.|\!|\:|\_|\-|\\|/|\[|\])"""," ").split(" ")).filter(x=>(x.size>3)).filter(x=>deleteNumberInSampleSignatures(x)).map(x=>x.toLowerCase())))
val  VT_sample_signatures_initial_seq_rdd = VT_sample_signatures_rdd.map(x=>new VT_sample_signatures_initial_seq_rdd_class(x._1, x._2))

implicit def bool2int(b:Boolean) = if (b) 1 else 0
def findAllInFamilySignatures(sample_signatures_seq : Seq[String]) : Array[Double] ={
    val forlist = for (family <- family_signatures_list) yield {
        (sample_signatures_seq.contains(family):Int).toDouble
    }
    return forlist.toArray
}

val VT_sample_signatures_final_array_rdd = VT_sample_signatures_initial_seq_rdd.map(x=>new VT_sample_signatures_final_array_rdd_class(x.sha256,findAllInFamilySignatures(x.seq_results)))
VT_sample_signatures_final_array_rdd.toDF().write.format("parquet").save(VT_sample_signatures_final_array_file) 

//设置package包名称以及导入依赖的类
package at.hazm.quebic

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream, OutputStream}
import java.util.zip.{GZIPInputStream, GZIPOutputStream}

import scala.annotation.tailrec

sealed abstract class Codec(val id:Byte, val name:String) extends Type {
  def encode(buffer:Array[Byte]):Array[Byte]

  def decode(buffer:Array[Byte]):Array[Byte]
}

object Codec {
  val values:Seq[Codec] = Seq(PLAIN, GZIP)
  private[this] val valuesMap = values.groupBy(_.id).mapValues(_.head)

  def valueOf(id:Byte):Codec = valuesMap(id)

  case object PLAIN extends Codec(0, "plain") {
    def encode(buffer:Array[Byte]):Array[Byte] = buffer

    def decode(buffer:Array[Byte]):Array[Byte] = buffer
  }

  case object GZIP extends Codec(1, "gzip") {
    def encode(buffer:Array[Byte]):Array[Byte] = {
      val baos = new ByteArrayOutputStream()
      val out = new GZIPOutputStream(baos)
      out.write(buffer)
      out.finish()
      out.finish()
      baos.toByteArray
    }

    def decode(buffer:Array[Byte]):Array[Byte] = {
      val in = new GZIPInputStream(new ByteArrayInputStream(buffer))
      val out = new ByteArrayOutputStream()
      _copy(in, out, new Array[Byte](2014))
      out.close()
      out.toByteArray
    }
  }

  @tailrec
  private[Codec] def _copy(in:InputStream, out:OutputStream, buffer:Array[Byte]):Unit = {
    val len = in.read(buffer)
    if(len > 0) {
      out.write(buffer, 0, len)
      _copy(in, out, buffer)
    }
  }

} 

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

import java.io.{BufferedInputStream, InputStream}
import java.util.zip.GZIPInputStream

import scala.collection.mutable.{Map => MutableMap}
import scala.io.Source



  private lazy val testDataIdMapped: Vector[SearchRecord] = {
    testDataTyped.map { record =>
      val userCity = cityIds.getOrElse(record.userCity, 0)
      val dest = destIds.getOrElse(record.dest, 0)
      SearchRecord(userCity, record.distance, dest)
    }.toVector
  }

  private lazy val trainDataTyped: Iterator[SearchRecord] = readData(trainDataFile)

  private lazy val testDataTyped: Iterator[SearchRecord] = readData(testDataFile)

  private lazy val trainDataFile = getFileStream("train-data.csv.gz")
  private lazy val trainLabelsFile = getFileStream("train-labels.csv.gz")
  private lazy val testDataFile = getFileStream("test-data.csv.gz")
  private lazy val testLabelsFile = getFileStream("test-labels.csv.gz")

  private def getFileStream(name: String): InputStream = {
    new BufferedInputStream(new GZIPInputStream(
      this.getClass.getClassLoader.getResourceAsStream(name)))
  }

  private def readLabels(stream: InputStream): Iterator[Int] =
    Source.fromInputStream(stream, "UTF8").getLines.map(_.toInt)

  private def readData(stream: InputStream): Iterator[SearchRecord] = {
    Source.fromInputStream(stream, "UTF8").getLines.map(SearchRecord.fromString)
  }
}

case class SearchRecord(userCity: Int, distance: Double, dest: Int)
object SearchRecord {
  def fromString(s: String) = {
    val fields = s.split(',')
    val userCity = fields(0).toInt
    val distance = fields(1).toDouble
    val dest = fields(2).toInt
    SearchRecord(userCity, distance, dest)
  }
} 

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

import java.io.{ InputStream, File, FileInputStream }
import java.util.zip.GZIPInputStream
import models.place.{ GazetteerRecord, PlaceService }
import play.api.Logger
import scala.concurrent.{ Await, ExecutionContext }
import scala.concurrent.duration._

class DumpImporter {
  
  private def getStream(file: File, filename: String) =
    if (filename.endsWith(".gz"))
      new GZIPInputStream(new FileInputStream(file))
    else
      new FileInputStream(file)
  
  def importDump(file: File, filename: String, crosswalk: InputStream => Seq[GazetteerRecord])(implicit places: PlaceService, ctx: ExecutionContext) = {
    val records = crosswalk(getStream(file, filename))
    Logger.info("Importing " + records.size + " records")
    Await.result(places.importRecords(records), 60.minute)   
  }
  
} 

//设置package包名称以及导入依赖的类
package uk.pkerrigan.dmarcparser

import java.io.ByteArrayInputStream
import java.nio.charset.CodingErrorAction
import java.util.Properties
import java.util.zip.{GZIPInputStream, ZipInputStream}
import javax.activation.DataSource
import javax.mail.Session
import javax.mail.internet.MimeMessage
import scala.collection.JavaConverters._

import org.apache.commons.mail.util.MimeMessageParser
import uk.pkerrigan.dmarcparser.report.Feedback

import scala.io._

class EmailParser(parser: ParserTrait = new Parser()) extends EmailParserTrait{

  implicit val codec = Codec("UTF-8")
  codec.onMalformedInput(CodingErrorAction.REPLACE)
  codec.onUnmappableCharacter(CodingErrorAction.REPLACE)

  def parseEmail(email: String): Option[Feedback] = {
    val s = Session.getDefaultInstance(new Properties())
    val is = new ByteArrayInputStream(email.getBytes)
    val message = new MimeMessage(s, is)
    val messageParser = new MimeMessageParser(message).parse()

    messageParser.getAttachmentList.asScala.headOption.flatMap(extract)
  }

  private def extract(a: DataSource): Option[Feedback] = a match {
    case `a` if a.getContentType.equals("application/gzip") => extractGzip(a)
    case `a` if a.getContentType.equals("application/x-gzip") => extractGzip(a)
    case `a` if a.getContentType.equals("application/zip") => extractZip(a)
    case `a` if a.getContentType.equals("application/x-zip-compressed") => extractZip(a)
    case _ => None
  }

  private def extractZip(a: DataSource): Option[Feedback] = {
    val zip = new ZipInputStream(a.getInputStream)
    zip.getNextEntry
    val rawXml = Source.fromInputStream(zip).mkString
    if (rawXml == "") None else Some(parser.parse(rawXml))
  }
  private def extractGzip(a: DataSource): Option[Feedback] = {
    val zip = new GZIPInputStream(a.getInputStream)
    val rawXml = Source.fromInputStream(zip).mkString
    if (rawXml == "") None else Some(parser.parse(rawXml))
  }
} 

//设置package包名称以及导入依赖的类
import com.datastax.spark.connector._
import play.api.libs.json.Json
import play.api.libs.json._
import java.io.{ByteArrayOutputStream, ByteArrayInputStream}
import java.util.zip.{GZIPOutputStream, GZIPInputStream}
import PreProcessingConfig._

case class objdump_results_by_service_name_class(service_name: String, sha256: String)
case class objdump_results_by_sha256_class(sha256: String, service_name: String, results: Array[Byte])
case class objdump_join_results_class(sha256: String, service_name: String, results: String)
case class objdump_binaray_final_array_rdd_class(sha256: String, array_results: Array[Double])
 
val objdump_main_list = sc.textFile(objdump_x86Opcodes_file).collect.toList
def unzip(x: Array[Byte]) : String = {		
    val inputStream = new GZIPInputStream(new ByteArrayInputStream(x))
    val output = scala.io.Source.fromInputStream(inputStream).mkString
    return output
}
def combineAllObjdumpInOne( malwarelist :Seq[play.api.libs.json.JsValue]) : List[String] ={
    if (malwarelist(0).toString() == "null") return List("null")
    var begin = malwarelist(0).as[List[String]]
    for (i <- 1 to (malwarelist.size-1)){
        if (malwarelist(i).toString() == "null") begin = begin
        else begin = begin ::: malwarelist(i).as[List[String]]
    }
    return  begin
}
def convertToList( malwarelist :Seq[play.api.libs.json.JsValue]) : List[String] = {
    if (malwarelist(0).toString() == "null") return List("null")
    else {
        return malwarelist(0).as[List[String]]
    } 
    
}
def findAllBininobjdump_main_list(malware :List[String]) : Array[Double] ={
    if (malware == List("null")) return (List.fill(10000)(0.0)).toArray
    else {
        val forlist = for ( one  <- malware ) yield {
            objdump_main_list.indexOf(one) + 1.0
        }
        if (forlist.size < 10000){
            return  (List.concat(forlist,List.fill(10000-forlist.size)(0.0))).toArray
        }
        else return forlist.toArray
    }
}

val objdump_results_by_service_name_meta = sc.cassandraTable[objdump_results_by_service_name_class](keyspace,service_name_table).where("service_name=?","objdump")
val objdump_results_by_service_name_rdd = objdump_results_by_service_name_meta.keyBy(x=> (x.sha256,x.service_name))
val objdump_results_by_sha256_meta = sc.cassandraTable[objdump_results_by_sha256_class](keyspace,sha256_table)
val objdump_results_by_sha256_rdd = objdump_results_by_sha256_meta.keyBy(x => (x.sha256,x.service_name))
val objdump_join_results = objdump_results_by_service_name_rdd.join(objdump_results_by_sha256_rdd).map(x=> (new objdump_join_results_class(x._1._1,x._1._2, unzip(x._2._2.results)))).distinct()
val objdump_binaray_final_array_rdd = objdump_join_results.map(x=>(x.sha256,(Json.parse(x.results) \\ "opcodes"))).filter(x=> (x._2.size > 0)).map(x=>(x._1,if ( x._2.size == 1 ) convertToList(x._2) else combineAllObjdumpInOne(x._2))).map(x=>(x._1,findAllBininobjdump_main_list(x._2)))
objdump_binaray_final_array_rdd.toDF().write.format("parquet").save(objdump_binaray_final_array_file) 

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

import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
import java.util.zip.{GZIPInputStream, GZIPOutputStream}

import com.google.common.io.ByteStreams


object Zip {

  def compress(bytes: Array[Byte]): Array[Byte] = {
    val bos = new ByteArrayOutputStream()
    val zs = new GZIPOutputStream(bos)
    zs.write(bytes)
    zs.close()
    bos.toByteArray
  }

  def decompress(bytes: Array[Byte]): Array[Byte] = {
    val bis = new ByteArrayInputStream(bytes)
    val zs = new GZIPInputStream(bis)
    ByteStreams.toByteArray(zs)
  }

} 

//设置package包名称以及导入依赖的类
package com.flipkart.connekt.commons.utils

import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
import java.util.Base64
import java.util.zip.{GZIPInputStream, GZIPOutputStream}

import com.flipkart.connekt.commons.core.Wrappers._
import org.apache.commons.io.IOUtils

import scala.util.Try
import scala.util.control.NoStackTrace

object CompressionUtils {

  def inflate(deflatedTxt: String): Try[String] = Try_ {
    val bytes = Base64.getUrlDecoder.decode(deflatedTxt)
    try {
      val zipInputStream = new GZIPInputStream(new ByteArrayInputStream(bytes))
      IOUtils.toString(zipInputStream)
    } catch {
      case ex:java.util.zip.ZipException =>
        throw new Exception(ex.getMessage) with NoStackTrace
    }
  }

  def deflate(txt: String): Try[String] = Try_ {
    val arrOutputStream = new ByteArrayOutputStream()
    val zipOutputStream = new GZIPOutputStream(arrOutputStream)
    zipOutputStream.write(txt.getBytes)
    zipOutputStream.close()
    Base64.getUrlEncoder.encodeToString(arrOutputStream.toByteArray)
  }

  implicit class StringCompress(val s: String) {
    def compress: Try[String] = deflate(s)
    def decompress: Try[String] = inflate(s)
  }

} 

//设置package包名称以及导入依赖的类
import java.nio.file.{Files, Path}
import java.time.format.DateTimeFormatter
import java.time.{LocalDate, LocalDateTime, ZoneId}
import java.util.Date
import java.util.zip.GZIPInputStream

import spray.json.{JsString, JsValue, RootJsonFormat}

import scala.io.Source


package object vijar {

  implicit def localDateTimeJsonFormat = new RootJsonFormat[LocalDateTime] {
    val dateFormatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME
    override def read(json: JsValue): LocalDateTime = LocalDateTime.parse(json.asInstanceOf[JsString].value, dateFormatter)

    override def write(obj: LocalDateTime): JsValue = JsString(obj.format(dateFormatter))
  }

  implicit class RichStringsOps(s: String) {
    def fromLongToLocalDateTime = {
      val t = new Date(s.toLong).toInstant
      LocalDateTime.ofInstant(t, ZoneId.systemDefault())
    }
  }

  case class GZIPReader[+T](path: Path)(f: Array[String] => T) extends Iterable[T]  {
    override def iterator: Iterator[T] = {
      Source.fromInputStream(new GZIPInputStream(Files.newInputStream(path))).getLines().map(l => f(l.split(" ")))
    }
  }
} 

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

import java.io.{BufferedReader, FileInputStream, InputStreamReader}
import java.util.zip.GZIPInputStream



class BufferedReaderIterator(reader: BufferedReader) extends Iterator[String] {
  override def hasNext() = reader.ready
  override def next() = reader.readLine()
}
object GzFileIterator {
  def apply(file: java.io.File, encoding: String) = {
    new BufferedReaderIterator(
      new BufferedReader(
        new InputStreamReader(
          new GZIPInputStream(
            new FileInputStream(file)), encoding)))
  }
} 

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

import java.io.{File, FileInputStream, FileOutputStream, ObjectInputStream, ObjectOutputStream}
import java.util.zip.{GZIPInputStream, GZIPOutputStream}

import com.outr.jefe.repo._

import scala.collection.mutable.ListBuffer

case class Configuration(dependency: VersionedDependency,
                         mainClass: String,
                         args: Array[String] = Array.empty,
                         vmArgs: Array[String] = Array.empty,
                         workingDirectory: File = new File("."),
                         showDialogIfPossible: Boolean = true,
                         repositories: Repositories = Repositories(),
                         newProcess: Boolean = false)

object Configuration {
  def load(file: File): Configuration = {
    val ois = new ObjectInputStream(new GZIPInputStream(new FileInputStream(file)))
    try {
      ois.readObject().asInstanceOf[Configuration]
    } finally {
      ois.close()
    }
  }

  def save(configuration: Configuration, file: File): Unit = {
    Option(file.getParentFile).foreach(_.mkdirs())
    val oos = new ObjectOutputStream(new GZIPOutputStream(new FileOutputStream(file)))
    try {
      oos.writeObject(configuration)
    } finally {
      oos.flush()
      oos.close()
    }
  }
}

case class Repositories(list: List[Repository] = List(Ivy2.Local, Maven.Repo1, Sonatype.Releases, Sonatype.Snapshots)) {
  def withMaven(name: String, url: String): Repositories = copy(list ::: List(MavenRepository(name, url)))
} 

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

import java.io.InputStream
import java.util.Date
import java.util.zip.GZIPInputStream

import com.google.common.base.Function
import com.treasuredata.client.{ExponentialBackOff, TDClient, TDClientHttpNotFoundException}
import com.treasuredata.client.model._
import org.msgpack.core.MessagePack

object JobName {
  def main(args: Array[String]): Unit = {
    val client: TDClient = TDClient.newClient()
    try {
      val date: Date = new Date()
      val jobId: String = client.startSavedQuery("saved_query", date) // Specify SavedQuery Name

      val backOff: ExponentialBackOff = new ExponentialBackOff()
      var job: TDJobSummary = client.jobStatus(jobId)
      while (!job.getStatus().isFinished()) {
        Thread.sleep(backOff.nextWaitTimeMillis())
        job = client.jobStatus(jobId)
      }

      val jobInfo: TDJob = client.jobInfo(jobId)
      println(jobInfo.getCmdOut())
      println(jobInfo.getStdErr())

      client.jobResult(jobId, TDResultFormat.MESSAGE_PACK_GZ, new Function[InputStream, Int] {
        def apply(input: InputStream): Int = {
          var count = 0
          try {
            val unpacker = MessagePack.newDefaultUnpacker(new GZIPInputStream(input))
            while (unpacker.hasNext()) {
              val array = unpacker.unpackValue().asArrayValue()
              println(array)
              count += 1
            }
          }
          count
        }
      })
    } catch  {
      case e: TDClientHttpNotFoundException => println(e)
    } finally {
      // Never forget to close the TDClient.
      // Program won't stop if client.close doesn't exist.
      client.close()
    }
  }
} 

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

import java.io.InputStream
import java.util.zip.GZIPInputStream

import com.google.common.base.Function
import com.treasuredata.client.{TDClient, TDClientHttpNotFoundException}
import com.treasuredata.client.model._
import org.msgpack.core.MessagePack

object JobId {
  def main(args: Array[String]): Unit = {
    val client: TDClient = TDClient.newClient()
    try {
      val jobId: String = "123456789" // Specify Job Id

      val jobInfo: TDJob = client.jobInfo(jobId)
      println(jobInfo.getCmdOut())
      println(jobInfo.getStdErr())

      client.jobResult(jobId, TDResultFormat.MESSAGE_PACK_GZ, new Function[InputStream, Int] {
        def apply(input: InputStream): Int = {
          var count = 0
          try {
            val unpacker = MessagePack.newDefaultUnpacker(new GZIPInputStream(input))
            while (unpacker.hasNext()) {
              val array = unpacker.unpackValue().asArrayValue()
              println(array)
              count += 1
            }
          }
          count
        }
      })
    } catch  {
      case e: TDClientHttpNotFoundException => println(e)
    } finally {
      // Never forget to close the TDClient.
      // Program won't stop if client.close doesn't exist.
      client.close()
    }
  }
} 

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

import com.google.common.base.Function
import java.io.InputStream
import java.util.zip.GZIPInputStream

import com.treasuredata.client.{ExponentialBackOff, TDClient, TDClientHttpNotFoundException}
import com.treasuredata.client.model._
import org.msgpack.core.MessagePack

import scala.collection.JavaConversions._

object Hello {
  def main(args: Array[String]): Unit = {
    val client: TDClient = TDClient.newClient()
    try {
      val list: java.util.List[TDDatabase] = client.listDatabases()
      for (databases <- list) {
        println("database: " + databases.getName)
        for (tables <- client.listTables(databases.getName)) {
          println("table: " + tables)
        }
      }

      val jobId: String = client.submit(TDJobRequest.newHiveQuery("sample_db", "SELECT v['code'] AS code, COUNT(1) AS cnt FROM www_access GROUP BY v['code']"))
      println(jobId)

      val backOff: ExponentialBackOff = new ExponentialBackOff()
      var job: TDJobSummary = client.jobStatus(jobId)
      while (!job.getStatus().isFinished()) {
        Thread.sleep(backOff.nextWaitTimeMillis())
        job = client.jobStatus(jobId)
      }

      val jobInfo: TDJob = client.jobInfo(jobId)
      println(jobInfo.getCmdOut())
      println(jobInfo.getStdErr())

      client.jobResult(jobId, TDResultFormat.MESSAGE_PACK_GZ, new Function[InputStream, Int] {
        def apply(input: InputStream): Int = {
          var count = 0
          try {
            val unpacker = MessagePack.newDefaultUnpacker(new GZIPInputStream(input))
            while (unpacker.hasNext()) {
              val array = unpacker.unpackValue().asArrayValue()
              println(array)
              count += 1
            }
          }
          count
        }
      })
    } catch {
      case e: TDClientHttpNotFoundException => println(e)
    } finally {
      // Never forget to close the TDClient.
      // Program won't stop if client.close doesn't exist.
      client.close()
    }
  }
} 

//设置package包名称以及导入依赖的类
package uk.pkerrigan.darwin

import java.io.ByteArrayInputStream
import java.net.InetSocketAddress
import java.util.zip.GZIPInputStream

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.util.ByteString
import io.tvc.stomp.{Credentials, QueueName, StompSource}

import scala.io.Source

object Main extends App {

  implicit val as = ActorSystem()
  implicit val mat = ActorMaterializer()
  implicit val ec = mat.executionContext

  def gzDecode(b: ByteString): String =
    Source.fromInputStream(new GZIPInputStream(new ByteArrayInputStream(b.toArray))).mkString

  val stompSource = (for {
    queue <- sys.env.get("NATIONAL_RAIL_QUEUE_NAME").toRight("No queue name")
    username <- sys.env.get("NATIONAL_RAIL_USERNAME").toRight("No username")
    password <- sys.env.get("NATIONAL_RAIL_PASSWORD").toRight("No password")
  } yield StompSource(
    queue = QueueName(queue),
    host = InetSocketAddress.createUnresolved("datafeeds.nationalrail.co.uk", 61613),
    credentials = Some(Credentials(login = username, passcode = password))
  )).fold(e => throw new Exception(e), identity)

  stompSource
    .mapConcat(b => b.body.map(gzDecode).toList)
    .runForeach(println(_))
} 

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

import java.io.ByteArrayInputStream
import java.util.zip.GZIPInputStream

import com.fustigatedcat.heystk.common.normalization.Normalization
import com.fustigatedcat.heystk.engine.queue.RabbitQueue
import org.apache.commons.codec.binary.Base64
import org.apache.commons.codec.digest.DigestUtils
import org.apache.commons.io.IOUtils

import org.json4s.native.JsonMethods.parse
import org.slf4j.LoggerFactory

object NormalizationHandler {

  val logger = LoggerFactory.getLogger(this.getClass)

  implicit val formats = org.json4s.DefaultFormats

  def checksum(string : String) : String = {
    DigestUtils.sha256Hex(string)
  }

  def unzip(string : String) : String = {
    val arr = Base64.decodeBase64(string)
    val gis = new GZIPInputStream(new ByteArrayInputStream(arr))
    IOUtils.toString(gis, "UTF-8")
  }

  def handle(string : String, chksm : String) = {
    val body = unzip(string)
    if(checksum(body) == chksm) {
      // write normalization to temp DB and post to Queue for needs processing
      parse(unzip(string)).extract[List[Normalization]].par.foreach(RabbitQueue.postToProcess)
    } else {
      logger.error(s"Invalid checksum ${checksum(body)} != $chksm")
    }
  }

} 

//设置package包名称以及导入依赖的类
package org.zalando.znap.utils

import java.io.{BufferedReader, ByteArrayInputStream, ByteArrayOutputStream, InputStreamReader}
import java.util.Base64
import java.util.zip.{GZIPInputStream, GZIPOutputStream}

import akka.http.impl.model.parser.Base64Parsing
import org.apache.commons.compress.utils.IOUtils

object Compressor {
  private val encoding = "UTF-8"

  private val base64Encoder = Base64.getEncoder
  private val base64Decoder = Base64.getDecoder

  def compress(string: String): Array[Byte] = {
    val baos = new ByteArrayOutputStream()
    val gos = new GZIPOutputStream(baos)
    gos.write(string.getBytes(encoding))
    gos.flush()
    baos.flush()
    IOUtils.closeQuietly(gos)
    IOUtils.closeQuietly(baos)
    baos.toByteArray
  }

  def compressBase64(string: String): String = {
    base64Encoder.encodeToString(compress(string))
  }

  def decompress(array: Array[Byte]): String = {
    val bais = new ByteArrayInputStream(array)
    val gis = new GZIPInputStream(bais)
    val result = new String(IOUtils.toByteArray(gis), encoding)
    IOUtils.closeQuietly(bais)
    IOUtils.closeQuietly(gis)
    result
  }

  def decompressBase64(base64String: String): String = {
    decompress(base64Decoder.decode(base64String))
  }
} 

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

import java.io._
import java.util.zip.{GZIPInputStream, GZIPOutputStream}

import scala.annotation.tailrec

sealed trait CompressionType
case object NoneType extends CompressionType
case object GzipType extends CompressionType

object Compression {

  def gzip(data: Array[Byte]): Array[Byte] = {
    try {
        val bos = new ByteArrayOutputStream()
        val out = new GZIPOutputStream(bos)
        out.write(data)
        out.close()
        bos.toByteArray
    } catch {
      case e: IOException => throw new RuntimeException(e);
    }
  }

  def gunzip(data: Array[Byte]): Array[Byte] = {
    try {
      val bos = new ByteArrayOutputStream()
      val bis = new ByteArrayInputStream(data)
      val in = new GZIPInputStream(bis)
      copy(in, bos)
      in.close()
      bos.close()
      bos.toByteArray()
    } catch {
      case e: IOException => throw new RuntimeException(e);
    }
  }

  @tailrec
  private def copy(is: InputStream, os: OutputStream, bufferSize: Int = 1024) {
    val buffer = new Array[Byte](bufferSize)
    is.read(buffer, 0, buffer.length) match {
      case -1 => ()
      case n =>
        os.write(buffer, 0, n)
        copy(is, os, bufferSize)
    }
  }

} 

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

import java.io.BufferedInputStream
import java.net.URI
import java.util.zip.GZIPInputStream

import org.tukaani.xz.XZInputStream

import scala.io.Source

object Main {

  lazy val log = org.slf4j.LoggerFactory.getLogger(getClass.getName)

  def main(args: Array[String]): Unit = {
    val conf = new Conf(args)
    val publisher = new KafkaPublisher(conf.brokers())
    val topic = conf.topic()
    val sleep = 1000L / conf.eventsPerSecond()
    val uri = new URI(conf.uri())
    val inputStream = new BufferedInputStream(uri.toURL.openStream())

    val wrappedStream = if (conf.uri().endsWith(".gz")) {
      new GZIPInputStream(inputStream)
    }
    else if (conf.uri().endsWith(".xz")) {
      new XZInputStream(inputStream)
    }
    else {
      inputStream
    }
    val source = Source.fromInputStream(wrappedStream)

    var done = 0

    log.info(s"Reading crime from ${conf.uri()} and publishing to ${conf.brokers()} every ${sleep}ms")

    source.getLines().foreach(line => {
      publisher.publishKafka(topic, line.getBytes)
      done += 1

      if (done % 1000 == 0) {
        log.info(s"$done lines done")
      }

      Thread.sleep(sleep)
    })

    log.info(s"$done lines done")
  }
} 

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

import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream}
import java.util.zip.{GZIPInputStream, GZIPOutputStream}


object Zip {

  def compress(bytes: Array[Byte]): Array[Byte] = {
    val bos = new ByteArrayOutputStream()
    val zs = new GZIPOutputStream(bos)
    zs.write(bytes)
    zs.close()
    bos.toByteArray
  }

  def decompress(bytes: Array[Byte]): Array[Byte] = {
    val bis = new ByteArrayInputStream(bytes)
    val zs = new GZIPInputStream(bis)
    toByteArray(zs)
  }

  private def toByteArray(in: InputStream): Array[Byte] = {
    val arr = new Array[Byte](in.available)
    in.read(arr)
    arr
  }
}