2011-03-07 21:41:53 -05:00
|
|
|
package spark
|
|
|
|
|
|
|
|
import java.io._
|
2012-06-07 03:25:47 -04:00
|
|
|
import java.nio.ByteBuffer
|
2011-03-07 21:41:53 -05:00
|
|
|
|
2012-06-20 02:48:26 -04:00
|
|
|
import spark.util.ByteBufferInputStream
|
|
|
|
|
2011-03-07 21:41:53 -05:00
|
|
|
class JavaSerializationStream(out: OutputStream) extends SerializationStream {
|
|
|
|
val objOut = new ObjectOutputStream(out)
|
|
|
|
def writeObject[T](t: T) { objOut.writeObject(t) }
|
|
|
|
def flush() { objOut.flush() }
|
|
|
|
def close() { objOut.close() }
|
|
|
|
}
|
|
|
|
|
2012-06-07 03:25:47 -04:00
|
|
|
class JavaDeserializationStream(in: InputStream, loader: ClassLoader)
|
|
|
|
extends DeserializationStream {
|
2011-03-07 21:41:53 -05:00
|
|
|
val objIn = new ObjectInputStream(in) {
|
|
|
|
override def resolveClass(desc: ObjectStreamClass) =
|
2012-06-07 03:25:47 -04:00
|
|
|
Class.forName(desc.getName, false, loader)
|
2011-03-07 21:41:53 -05:00
|
|
|
}
|
|
|
|
|
|
|
|
def readObject[T](): T = objIn.readObject().asInstanceOf[T]
|
|
|
|
def close() { objIn.close() }
|
|
|
|
}
|
|
|
|
|
2011-05-17 15:41:13 -04:00
|
|
|
class JavaSerializerInstance extends SerializerInstance {
|
2012-06-07 03:25:47 -04:00
|
|
|
def serialize[T](t: T): ByteBuffer = {
|
2011-03-07 21:41:53 -05:00
|
|
|
val bos = new ByteArrayOutputStream()
|
2012-06-07 03:25:47 -04:00
|
|
|
val out = serializeStream(bos)
|
2011-03-07 21:41:53 -05:00
|
|
|
out.writeObject(t)
|
|
|
|
out.close()
|
2012-06-07 03:25:47 -04:00
|
|
|
ByteBuffer.wrap(bos.toByteArray)
|
2011-03-07 21:41:53 -05:00
|
|
|
}
|
|
|
|
|
2012-06-07 03:25:47 -04:00
|
|
|
def deserialize[T](bytes: ByteBuffer): T = {
|
2012-06-20 02:48:26 -04:00
|
|
|
val bis = new ByteBufferInputStream(bytes)
|
2012-06-07 03:25:47 -04:00
|
|
|
val in = deserializeStream(bis)
|
2011-03-07 21:41:53 -05:00
|
|
|
in.readObject().asInstanceOf[T]
|
|
|
|
}
|
|
|
|
|
2012-06-07 03:25:47 -04:00
|
|
|
def deserialize[T](bytes: ByteBuffer, loader: ClassLoader): T = {
|
2012-06-20 02:48:26 -04:00
|
|
|
val bis = new ByteBufferInputStream(bytes)
|
2012-06-07 03:25:47 -04:00
|
|
|
val in = deserializeStream(bis, loader)
|
|
|
|
in.readObject().asInstanceOf[T]
|
2012-04-10 00:59:56 -04:00
|
|
|
}
|
|
|
|
|
2012-06-07 03:25:47 -04:00
|
|
|
def serializeStream(s: OutputStream): SerializationStream = {
|
2011-03-07 21:41:53 -05:00
|
|
|
new JavaSerializationStream(s)
|
|
|
|
}
|
|
|
|
|
2012-06-07 03:25:47 -04:00
|
|
|
def deserializeStream(s: InputStream): DeserializationStream = {
|
2012-06-29 21:47:12 -04:00
|
|
|
new JavaDeserializationStream(s, Thread.currentThread.getContextClassLoader)
|
2012-06-07 03:25:47 -04:00
|
|
|
}
|
|
|
|
|
|
|
|
def deserializeStream(s: InputStream, loader: ClassLoader): DeserializationStream = {
|
|
|
|
new JavaDeserializationStream(s, loader)
|
2011-03-07 21:41:53 -05:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2011-05-17 15:41:13 -04:00
|
|
|
class JavaSerializer extends Serializer {
|
|
|
|
def newInstance(): SerializerInstance = new JavaSerializerInstance
|
2011-03-07 21:41:53 -05:00
|
|
|
}
|