//设置package包名称以及导入依赖的类
package org.hammerlab.hadoop.kryo

import java.io.{ DataInputStream, DataOutputStream }

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer }
import org.apache.hadoop.io.Writable

class WritableSerializer[T <: Writable](ctorArgs: Any*) extends Serializer[T] {
  override def read(kryo: Kryo, input: Input, clz: Class[T]): T = {
    val t = clz.newInstance()
    t.readFields(new DataInputStream(input))
    t
  }

  override def write(kryo: Kryo, output: Output, t: T): Unit = {
    t.write(new DataOutputStream(output))
  }
} 

//设置package包名称以及导入依赖的类
package org.hammerlab.hadoop.kryo

import java.io.{ ObjectInputStream, ObjectOutputStream }

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer }

case class SerializableSerializer[T <: Serializable]()
  extends Serializer[T] {
  override def read(kryo: Kryo, input: Input, `type`: Class[T]): T =
    new ObjectInputStream(input)
      .readObject()
      .asInstanceOf[T]

  override def write(kryo: Kryo, output: Output, t: T): Unit =
    new ObjectOutputStream(output)
      .writeObject(t)
} 

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

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer }

package object kryo {
  def serializeAs[T, U](implicit to: T ? U, from: U ? T): Serializer[T] =
    new Serializer[T] {
      override def read(kryo: Kryo, input: Input, `type`: Class[T]): T =
        from(
          kryo
            .readClassAndObject(input)
            .asInstanceOf[U]
        )

      override def write(kryo: Kryo, output: Output, t: T): Unit =
        kryo.writeClassAndObject(output, to(t))
    }
} 

//设置package包名称以及导入依赖的类
package org.hammerlab.hadoop.kryo

import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream }

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{ Input, Output }
import org.hammerlab.test.Suite

class SerializableSerializerTest
  extends Suite {
  test("serde") {
    val kryo = new Kryo()
    kryo.setRegistrationRequired(true)
    val baos = new ByteArrayOutputStream()
    val output = new Output(baos)

    val foo = new Foo
    foo.n = 123
    foo.s = "abc"

    intercept[IllegalArgumentException] {
      kryo.writeClassAndObject(output, foo)
    }
    .getMessage should startWith("Class is not registered: org.hammerlab.hadoop.kryo.Foo")

    kryo.register(classOf[Foo], SerializableSerializer[Foo]())

    kryo.writeClassAndObject(output, foo)

    output.close()

    val bytes = baos.toByteArray
    bytes.length should be(93)

    val bais = new ByteArrayInputStream(bytes)

    val input = new Input(bais)
    val after = kryo.readClassAndObject(input).asInstanceOf[Foo]

    after.n should be(foo.n)
    after.s should be(foo.s)
  }
}

class Foo
  extends Serializable {

  var n = 0
  var s = ""

  private def writeObject(out: ObjectOutputStream): Unit = {
    out.writeInt(n)
    out.writeUTF(s)
  }

  private def readObject(in: ObjectInputStream): Unit = {
    n = in.readInt()
    s = in.readUTF()
  }
} 

//设置package包名称以及导入依赖的类
package org.janzhou.nvmr.storage

import org.janzhou.nvmr.config
import org.janzhou.nvmr.console
import java.io._
import com.esotericsoftware.kryo._
import com.esotericsoftware.kryo.io.Input
import com.esotericsoftware.kryo.io.Output

object Setup {
  val kryo = new Kryo()

  def driver_setup(dev: String) {
    val driver = new Driver(dev)
    if( 0 != driver.trim(0L, driver.size) ) {
      console.debug("trim not supported")
    }
    driver.writeObject(0L, 83L)
  }

  def device_setup(_dev: String) {
    val dev = new RandomAccessFile(_dev, "rw")
    val out = new DeviceOutputStream(dev, 0L)
    new ObjectOutputStream(out).writeObject(82L)
  }

  def kryo_device_setup(_dev: String) {
    val dev = new RandomAccessFile(_dev, "rw")
    val out = new DeviceOutputStream(dev, 0L)
    val output = new Output(out)
    kryo.writeObject(output, 82L);
    output.close()
  }

  def main (args: Array[String]) {
    for ( dev <- config.getStringList("storage.devices") ) {
      config.getString("storage.engine") match {
        case "kryo" => kryo_device_setup(dev)
        case "device" => device_setup(dev)
        case "driver" => driver_setup(dev)
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package org.hammerlab.genomics.bases

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer }

class BasesSerializer
  extends Serializer[Bases] {

  override def write(kryo: Kryo, output: Output, bases: Bases): Unit = {
    output.writeInt(bases.bytes.length, true)
    bases.foreach(base ? output.writeByte(base.byte))
  }

  override def read(kryo: Kryo, input: Input, cls: Class[Bases]): Bases = {
    val count: Int = input.readInt(true)
    val bases = Bases.newBuilder
    (0 until count).foreach(_ ? bases += input.readByte())
    bases.result()
  }
} 

//设置package包名称以及导入依赖的类
package se.kth.climate.fast.netcdf

import org.apache.spark.serializer.KryoRegistrator
import com.esotericsoftware.kryo.{ Kryo, KryoException, Serializer }
import com.esotericsoftware.kryo.io.Output
import com.esotericsoftware.kryo.io.Input
import ucar.nc2.NetcdfFile
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import ucar.nc2.iosp.netcdf3.N3outputStreamWriter;
import java.util.UUID;

class NCKryoRegistrator extends KryoRegistrator {
  override def registerClasses(kryo: Kryo) {
    println("Registering custom NetCDF serializers");
    com.esotericsoftware.minlog.Log.TRACE();
    kryo.register(classOf[NetcdfFile], new NetcdfFileSerializer());
    kryo.register(classOf[Array[String]], new com.esotericsoftware.kryo.serializers.DefaultArraySerializers.StringArraySerializer())
    kryo.register(classOf[Array[Int]], new com.esotericsoftware.kryo.serializers.DefaultArraySerializers.IntArraySerializer())

    //kryo.setRegistrationRequired(true);
  }
}

class NetcdfFileSerializer extends Serializer[NetcdfFile] {
  override def write(kryo: Kryo, output: Output, ncfile: NetcdfFile) {
    val baos = new ByteArrayOutputStream();
    val out = new DataOutputStream(baos);
    val w = new N3outputStreamWriter(ncfile);
    val ulim = ncfile.getUnlimitedDimension();
    val numrec = if (ulim == null) 0 else ncfile.getUnlimitedDimension().getLength();
    w.writeHeader(out, numrec);
    w.writeDataAll(out);
    out.flush();
    out.close();
    val byteArray = baos.toByteArray();
    //    println("**********BYTES***********");
    //    println(byteArray.take(100000).mkString);
    output.writeInt(byteArray.length);
    output.write(byteArray);
    println(s"******** Wrote ncfile (size=${byteArray.length}) **********");
  }

  override def read(kryo: Kryo, input: Input, `type`: Class[NetcdfFile]): NetcdfFile = {
    val len = input.readInt();
    println(s"******** Reading ncfile (size=${len}) **********");
    val raw = new Array[Byte](len);
    var readBytes = 0;
    do {
      val res = input.read(raw, readBytes, len - readBytes);
      if (res > -1) {
        readBytes += res;
      } else {
        throw new KryoException(s"Read only $readBytes bytes when $len bytes were expected!");
      }
    } while (readBytes < len);
    println(s"******** Read ncfile (read=${readBytes}, size=${len}) **********");
    NetcdfFile.openInMemory(UUID.randomUUID().toString(), raw);
  }
} 

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

import java.io.{ByteArrayInputStream, ByteArrayOutputStream}

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{Input, Output}
import com.esotericsoftware.kryo.pool.{KryoFactory, KryoPool}
import org.objenesis.strategy.StdInstantiatorStrategy

object KryoSerializer extends Serializer {

  val factory = new KryoFactory() {
    override def create(): Kryo = {
      val kryo = new Kryo()
      // configure kryo instance, customize settings
      kryo.setInstantiatorStrategy(new StdInstantiatorStrategy())
      kryo
    }
  }

  
  val kryoPool = new KryoPool.Builder(factory).softReferences().build()


  override def serialize(obj: AnyRef): Array[Byte] = {

    val stream = new ByteArrayOutputStream()
    val output = new Output(stream)
    val kryoInstance = kryoPool.borrow()
    kryoInstance.writeClassAndObject(output, obj)
    output.close()
    kryoPool.release(kryoInstance)
    stream.toByteArray
  }

  override def deserialize[T](bytes: Array[Byte])(implicit cTag: reflect.ClassTag[T]): T = {
    val stream = new ByteArrayInputStream(bytes)
    val input = new Input(stream)
    val kryoInstance = kryoPool.borrow()
    val obj = kryoInstance.readClassAndObject(input)
    kryoPool.release(kryoInstance)
    obj.asInstanceOf[T]
  }
} 

//设置package包名称以及导入依赖的类
package ru.ispras.modis.tm.chinesetm

import java.io.{FileInputStream, FileOutputStream}

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{Input, Output}
import org.objenesis.strategy.StdInstantiatorStrategy
import ru.ispras.modis.tm.attribute.{DefaultAttributeType, AttributeType}
import ru.ispras.modis.tm.matrix.{AttributedPhi, Background, Theta}
import ru.ispras.modis.tm.plsa.TrainedModel


class NTMTrainedModel(phi: Map[AttributeType, AttributedPhi],
                      theta: Theta,
                      val noise: Map[AttributeType, Background],
                      val background: Map[AttributeType, Background],
                      perplexity: Double) extends TrainedModel(phi, theta, perplexity)  {
    def getNoise() = {
        require(noise.contains(DefaultAttributeType), "there is no default attribute in collection")
        require(noise.keys.size == 1, "Do not use this method in case of multiattribute collection")
        noise(DefaultAttributeType)
    }

    def getBackground() = {
        require(noise.contains(DefaultAttributeType), "there is no default attribute in collection")
        require(noise.keys.size == 1, "Do not use this method in case of multiattribute collection")
        noise(DefaultAttributeType)
    }
}

object NTMTrainedModel {
    def save(model: NTMTrainedModel, path: String) {
        val kryo = new Kryo
        kryo.setInstantiatorStrategy(new StdInstantiatorStrategy)
        val output = new Output(new FileOutputStream(path))
        kryo.writeObject(output, model)
        output.close()
    }

    def load(path: String) = {
        val kryo = new Kryo
        kryo.setInstantiatorStrategy(new StdInstantiatorStrategy)
        val input = new Input(new FileInputStream(path))
        val trainedModel = kryo.readObject(input, classOf[NTMTrainedModel])
        input.close()
        trainedModel
    }
} 

//设置package包名称以及导入依赖的类
package org.hammerlab.genomics.loci.set

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer ? KryoSerializer }
import com.google.common.collect.{ TreeRangeSet, Range ? JRange }
import JRange.closedOpen
import org.hammerlab.genomics.reference.{ ContigName, Interval, Locus }

// We serialize a LociSet simply by writing out its constituent Contigs.
class ContigSerializer extends KryoSerializer[Contig] {

  def write(kryo: Kryo, output: Output, obj: Contig) = {
    kryo.writeObject(output, obj.name)
    output.writeInt(obj.numRanges)
    for {
      Interval(start, end) ? obj.ranges
    } {
      output.writeLong(start.locus)
      output.writeLong(end.locus)
    }
  }

  def read(kryo: Kryo, input: Input, klass: Class[Contig]): Contig = {
    val name = kryo.readObject(input, classOf[ContigName])
    val numRanges = input.readInt()
    val treeRangeSet = TreeRangeSet.create[Locus]()
    val ranges =
      (0 until numRanges).foreach { _ ?
        treeRangeSet.add(
          closedOpen(
            Locus(input.readLong()),
            Locus(input.readLong())
          )
        )
      }

    Contig(name, treeRangeSet)
  }
} 

//设置package包名称以及导入依赖的类
package org.hammerlab.genomics.loci.set

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer ? KryoSerializer }

// We just serialize the underlying contigs, which contain their names which are the string keys of LociSet.map.
class Serializer extends KryoSerializer[LociSet] {
  def write(kryo: Kryo, output: Output, obj: LociSet) = {
    kryo.writeObject(output, obj.contigs)
  }

  def read(kryo: Kryo, input: Input, klass: Class[LociSet]): LociSet = {
    val contigs = kryo.readObject(input, classOf[Array[Contig]])
    LociSet.fromContigs(contigs)
  }
} 

//设置package包名称以及导入依赖的类
package org.hammerlab.genomics.loci.map

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer ? KryoSerializer }
import org.hammerlab.genomics.reference.{ ContigName, Locus }


class ContigSerializer[T] extends KryoSerializer[Contig[T]] {
  def write(kryo: Kryo, output: Output, obj: Contig[T]) = {
    kryo.writeObject(output, obj.name)
    output.writeLong(obj.asMap.size)
    obj.asMap.foreach {
      case (range, value) =>
        output.writeLong(range.start.locus)
        output.writeLong(range.end.locus)
        kryo.writeClassAndObject(output, value)
    }
  }

  def read(kryo: Kryo, input: Input, klass: Class[Contig[T]]): Contig[T] = {
    val builder = LociMap.newBuilder[T]
    val contig = kryo.readObject(input, classOf[ContigName])
    val count = input.readLong()
    (0L until count).foreach { _ =>
      val start = input.readLong()
      val end = input.readLong()
      val value: T = kryo.readClassAndObject(input).asInstanceOf[T]
      builder.put(contig, Locus(start), Locus(end), value)
    }
    builder.result(contig)
  }
} 

//设置package包名称以及导入依赖的类
package org.hammerlab.genomics.loci.map

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer ? KryoSerializer }


class Serializer[T] extends KryoSerializer[LociMap[T]] {
  def write(kryo: Kryo, output: Output, obj: LociMap[T]) = {
    output.writeLong(obj.contigs.size)
    obj.contigs.foreach(contig =>
      kryo.writeObject(output, contig)
    )
  }

  def read(kryo: Kryo, input: Input, klass: Class[LociMap[T]]): LociMap[T] = {
    val count: Long = input.readLong()
    val contigs = (0L until count).map(i =>
      kryo.readObject(input, classOf[Contig[T]])
    )
    LociMap.fromContigs(contigs)
  }
} 

//设置package包名称以及导入依赖的类
package com.truecar.mleap.spark.benchmark.util

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{Input, Output}
import com.twitter.chill.ScalaKryoInstantiator


object SparkSerializer {
  def apply(): SparkSerializer = {
    val kryoInstantiator = new ScalaKryoInstantiator()
    kryoInstantiator.setRegistrationRequired(false)
    val kryo = kryoInstantiator.newKryo()
    kryo.setClassLoader(Thread.currentThread.getContextClassLoader)

    SparkSerializer(kryo)
  }
}

case class SparkSerializer(kryo: Kryo) {
  def write[T](obj: T, output: Output) = {
    kryo.writeClassAndObject(output, obj)
  }

  def read[T](input: Input): T = {
    kryo.readClassAndObject(input).asInstanceOf[T]
  }
}