Add byte array serialization to InstantiationUtil

flink-core/src/main/java/org/apache/flink/util/InstantiationUtil.java

@@ -21,6 +21,8 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.ObjectInputStream;
@@ -237,6 +239,32 @@ public static void writeObjectToConfig(Object o, Configuration config, String ke
  byte[] bytes = serializeObject(o);
  config.setBytes(key, bytes);
  }
+
+ public static <T> byte[] serializeToByteArray(TypeSerializer<T> serializer, T record) throws IOException {
+ if (record == null) {
+ throw new NullPointerException("Record to serialize to byte array must not be null.");
+ }
+
+ ByteArrayOutputStream bos = new ByteArrayOutputStream(64);
+ OutputViewDataOutputWrapper outputViewWrapper = new OutputViewDataOutputWrapper();
+ outputViewWrapper.setDelegate(new DataOutputStream(bos));
+
+ serializer.serialize(record, outputViewWrapper);
+
+ return bos.toByteArray();
+ }
+
+ public static <T> T deserializeFromByteArray(TypeSerializer<T> serializer, byte[] buf) throws IOException {
+ if (buf == null) {
+ throw new NullPointerException("Byte array to deserialize from must not be null.");
+ }
+
+ InputViewDataInputWrapper inputViewWrapper = new InputViewDataInputWrapper();
+ inputViewWrapper.setDelegate(new DataInputStream(new ByteArrayInputStream(buf)));
+
+ T record = serializer.createInstance();
+ return serializer.deserialize(record, inputViewWrapper);
+ }
 
  public static Object deserializeObject(byte[] bytes, ClassLoader cl) throws IOException, ClassNotFoundException {
  ObjectInputStream oois = null;

flink-java/src/main/java/org/apache/flink/api/java/io/CollectionInputFormat.java

@@ -44,10 +44,9 @@ public class CollectionInputFormat<T> extends GenericInputFormat<T> implements N
  private TypeSerializer<T> serializer;
 
  private transient Collection<T> dataSet; // input data as collection. transient, because it will be serialized in a custom way
- 
+
  private transient Iterator<T> iterator;
 
-
  public CollectionInputFormat(Collection<T> dataSet, TypeSerializer<T> serializer) {
  if (dataSet == null) {
  throw new NullPointerException();
@@ -58,7 +57,6 @@ public CollectionInputFormat(Collection<T> dataSet, TypeSerializer<T> serializer
  this.dataSet = dataSet;
  }
 
-
  @Override
  public boolean reachedEnd() throws IOException {
  return !this.iterator.hasNext();

stratosphere-core/src/test/java/eu/stratosphere/util/InstantiationUtilTest.java

@@ -0,0 +1,80 @@
+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2013 by the Stratosphere project (http:https://stratosphere.eu)
+ *
+ * Licensed 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:https://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 eu.stratosphere.util;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import eu.stratosphere.api.common.typeutils.base.DoubleValueSerializer;
+import eu.stratosphere.types.DoubleValue;
+import eu.stratosphere.types.TypeInformation;
+import junit.framework.Assert;
+import org.junit.Test;
+
+import eu.stratosphere.types.StringValue;
+import eu.stratosphere.types.Value;
+
+import java.io.IOException;
+import java.util.Random;
+
+public class InstantiationUtilTest {
+
+ @Test
+ public void testInstatiationOfStringValue() {
+ StringValue stringValue = InstantiationUtil.instantiate(StringValue.class, null);
+ assertNotNull(stringValue);
+ }
+
+ @Test
+ public void testInstatiationOfStringValueAndCastToValue() {
+ StringValue stringValue = InstantiationUtil.instantiate(StringValue.class, Value.class);
+ assertNotNull(stringValue);
+ }
+
+ @Test
+ public void testHasNullaryConstructor() {
+ assertTrue(InstantiationUtil.hasPublicNullaryConstructor(StringValue.class));
+ }
+
+ @Test
+ public void testClassIsProper() {
+ assertTrue(InstantiationUtil.isProperClass(StringValue.class));
+ }
+
+ @Test
+ public void testClassIsNotProper() {
+ assertFalse(InstantiationUtil.isProperClass(Value.class));
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void testCheckForInstantiationOfPrivateClass() {
+ InstantiationUtil.checkForInstantiation(TestClass.class);
+ }
+
+ @Test
+ public void testSerializationToByteArray() throws IOException {
+ final DoubleValue toSerialize = new DoubleValue(Math.random());
+ final DoubleValueSerializer serializer = new DoubleValueSerializer();
+
+ byte[] serialized = InstantiationUtil.serializeToByteArray(serializer, toSerialize);
+
+ DoubleValue deserialized = InstantiationUtil.deserializeFromByteArray(serializer, serialized);
+
+ Assert.assertEquals("Serialized record is not equal after serialization.", toSerialize, deserialized);
+ }
+
+ private class TestClass {}
+}