You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by cr...@apache.org on 2015/02/12 23:43:45 UTC
samza git commit: SAMZA-547; add java serializable serde
Repository: samza
Updated Branches:
refs/heads/master 6743df319 -> 5e32a1bb0
SAMZA-547; add java serializable serde
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/5e32a1bb
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/5e32a1bb
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/5e32a1bb
Branch: refs/heads/master
Commit: 5e32a1bb018cb0382ab4911c31554f8889de984f
Parents: 6743df3
Author: Ruslan Khafizov <ru...@gmail.com>
Authored: Thu Feb 12 14:43:31 2015 -0800
Committer: Chris Riccomini <cr...@apache.org>
Committed: Thu Feb 12 14:43:31 2015 -0800
----------------------------------------------------------------------
.../samza/serializers/SerializableSerde.scala | 67 ++++++++++++++++++++
.../serializers/TestSerializableSerde.scala | 45 +++++++++++++
2 files changed, 112 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/5e32a1bb/samza-core/src/main/scala/org/apache/samza/serializers/SerializableSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/serializers/SerializableSerde.scala b/samza-core/src/main/scala/org/apache/samza/serializers/SerializableSerde.scala
new file mode 100644
index 0000000..c43f863
--- /dev/null
+++ b/samza-core/src/main/scala/org/apache/samza/serializers/SerializableSerde.scala
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.serializers
+
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+import java.io.ObjectInputStream
+import java.io.ObjectOutputStream
+
+import org.apache.samza.config.Config
+
+/**
+ * A serializer for Serializable
+ */
+class SerializableSerdeFactory[T <: java.io.Serializable] extends SerdeFactory[T] {
+ def getSerde(name: String, config: Config): Serde[T] =
+ new SerializableSerde[T]
+}
+
+class SerializableSerde[T <: java.io.Serializable] extends Serde[T] {
+ def toBytes(obj: T): Array[Byte] = if (obj != null) {
+ val bos = new ByteArrayOutputStream
+ val oos = new ObjectOutputStream(bos)
+
+ try {
+ oos.writeObject(obj)
+ }
+ finally {
+ oos.close()
+ }
+
+ bos.toByteArray
+ } else {
+ null
+ }
+
+ def fromBytes(bytes: Array[Byte]): T = if (bytes != null) {
+ val bis = new ByteArrayInputStream(bytes)
+ val ois = new ObjectInputStream(bis)
+
+ try {
+ ois.readObject.asInstanceOf[T]
+ }
+ finally{
+ ois.close()
+ }
+ } else {
+ null.asInstanceOf[T]
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/5e32a1bb/samza-core/src/test/scala/org/apache/samza/serializers/TestSerializableSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/serializers/TestSerializableSerde.scala b/samza-core/src/test/scala/org/apache/samza/serializers/TestSerializableSerde.scala
new file mode 100644
index 0000000..8e899d0
--- /dev/null
+++ b/samza-core/src/test/scala/org/apache/samza/serializers/TestSerializableSerde.scala
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.samza.serializers
+
+import org.junit.Assert._
+import org.junit.Test
+
+class TestSerializableSerde {
+ @Test
+ def testSerializableSerde {
+ val serde = new SerializableSerde[String]
+ assertNull(serde.toBytes(null))
+ assertNull(serde.fromBytes(null))
+
+ val obj = "String is serializable"
+
+ // Serialized string is prefix + string itself
+ val prefix = Array(0xAC, 0xED, 0x00, 0x05, 0x74, 0x00, 0x16).map(_.toByte)
+ val expected = (prefix ++ obj.getBytes("UTF-8"))
+
+ val bytes = serde.toBytes(obj)
+
+ assertArrayEquals(expected, bytes)
+
+ val objRoundTrip:String = serde.fromBytes(bytes)
+ assertEquals(obj, objRoundTrip)
+ }
+}