XD-1024 Added KryoMessageSerializer and configured in redis adapters

Author: David Turanski

spring-xd-dirt/src/main/java/org/springframework/integration/x/redis/KryoMessageHeadersSerializer.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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://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.springframework.integration.x.redis;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageHeaderAccessor;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+
+/**
+ * Implementation of Kryo Serializer to handle {@link MessageHeaders} which is immutable.
+ * 
+ * @author David Turanski
+ * @since 1.0
+ */
+public class KryoMessageHeadersSerializer extends Serializer<MessageHeaders> {
+
+	@Override
+	public void write(Kryo kryo, Output output, MessageHeaders headers) {
+		MessageHeaderAccessor mha = new MessageHeaderAccessor();
+		mha.copyHeaders(headers);
+		kryo.writeObject(output, mha.toMap());
+	}
+
+	@Override
+	public MessageHeaders read(Kryo kryo, Input input, Class<MessageHeaders> type) {
+		@SuppressWarnings("unchecked")
+		Map<String, Object> map = kryo.readObject(input, HashMap.class);
+		return new MessageHeaders(map);
+	}
+}

spring-xd-dirt/src/main/java/org/springframework/integration/x/redis/KryoMessageSerializer.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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://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.springframework.integration.x.redis;
+
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.SerializationException;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.GenericMessage;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.shaded.org.objenesis.strategy.StdInstantiatorStrategy;
+
+
+/**
+ * A Redis Serializer that uses Kryo serialization
+ * 
+ * @author David Turanski
+ * @since 1.0
+ */
+public class KryoMessageSerializer implements RedisSerializer<Message<?>> {
+
+	@Override
+	public byte[] serialize(Message<?> message) throws SerializationException {
+		Kryo kryo = kryoInstance();
+		Output output = new Output(2048, -1);
+		kryo.writeObjectOrNull(output, message, GenericMessage.class);
+		return output.getBuffer();
+	}
+
+	@Override
+	public Message<?> deserialize(byte[] bytes) throws SerializationException {
+		if (bytes == null) {
+			return null;
+		}
+		Kryo kryo = kryoInstance();
+		Input input = new Input(bytes);
+		return kryo.readObjectOrNull(input, GenericMessage.class);
+	}
+
+	private Kryo kryoInstance() {
+		Kryo kryo = new Kryo();
+		kryo.register(MessageHeaders.class, new KryoMessageHeadersSerializer());
+		kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+		return kryo;
+	}
+
+}

spring-xd-dirt/src/main/resources/META-INF/spring-xd/transports/redis-admin.xml

@@ -13,7 +13,9 @@
 		channel="deployChannel"
 		connection-factory="redisConnectionFactory"
 		queue="queue.deployer"
-		extract-payload="false" />
+		extract-payload="false" 
+		serializer="redisSerializer"
+		/>
 
 	<int-redis:outbound-channel-adapter channel="undeployChannel" topic="topic.undeployer" connection-factory="redisConnectionFactory" />
 

spring-xd-dirt/src/main/resources/META-INF/spring-xd/transports/redis-common.xml

@@ -6,7 +6,7 @@
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
 
-	<import resource="classpath:/META-INF/spring-xd/internal/xd-common.xml"/>
+	<import resource="classpath:/META-INF/spring-xd/internal/xd-common.xml" />
 
 	<bean id="redisConnectionFactory"
 		class="org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory">
@@ -18,10 +18,12 @@
 	<bean id="humanFriendlyRedisTemplate" class="org.springframework.data.redis.core.StringRedisTemplate">
 		<property name="connectionFactory" ref="redisConnectionFactory" />
 	</bean>
-	
+
 	<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
 		<property name="connectionFactory" ref="redisConnectionFactory" />
 	</bean>
-	
+
+	<bean id="redisSerializer"
+		class="org.springframework.integration.x.redis.KryoMessageSerializer" />
 
 </beans>

spring-xd-dirt/src/main/resources/META-INF/spring-xd/transports/redis-container.xml

@@ -19,7 +19,9 @@
 		channel="containerControlChannel"
 		connection-factory="redisConnectionFactory"
 		queue="queue.deployer"
-		expect-message="true" />
+		expect-message="true"
+		serializer="redisSerializer"
+		/>
 
 	<int-redis:inbound-channel-adapter topics="topic.undeployer" channel="containerControlChannel" />
 

spring-xd-dirt/src/test/java/org/springframework/integration/x/redis/KryoMessageSerializerTests.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * 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://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.springframework.integration.x.redis;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+
+/**
+ * 
+ * @author David Turanski
+ */
+public class KryoMessageSerializerTests {
+
+	private KryoMessageSerializer serializer = new KryoMessageSerializer();
+
+	@Test
+	public void testSerializeAndDeserialize() {
+		Map<String, Object> headers = new HashMap<String, Object>();
+		headers.put("header1", "foo");
+		headers.put("header2", "bar");
+		Foo foo = new Foo(123, "hello");
+		headers.put("header3", foo);
+		GenericMessage<String> message = new GenericMessage<String>("Hello", headers);
+		byte[] rawMessage = serializer.serialize(message);
+		@SuppressWarnings("unchecked")
+		Message<String> convertedMsg = (Message<String>) serializer.deserialize(rawMessage);
+		assertEquals("Hello", convertedMsg.getPayload());
+		assertEquals("foo", convertedMsg.getHeaders().get("header1"));
+		assertEquals("bar", convertedMsg.getHeaders().get("header2"));
+		assertNotNull(convertedMsg.getHeaders().get("id"));
+		assertNotNull(convertedMsg.getHeaders().get("timestamp"));
+		assertEquals(foo, convertedMsg.getHeaders().get("header3"));
+	}
+
+	@Test
+	public void testDeserializeNull() {
+		assertNull(serializer.deserialize(null));
+	}
+
+	public static class Foo {
+
+		private final int i;
+
+		private final String s;
+
+		public Foo(int i, String s) {
+			this.i = i;
+			this.s = s;
+		}
+
+		public int getI() {
+			return i;
+		}
+
+		public String getS() {
+			return s;
+		}
+
+		@Override
+		public boolean equals(Object other) {
+			if (!(other instanceof Foo)) {
+				return false;
+			}
+			Foo foo = (Foo) other;
+
+			return i == foo.i && (s == null && foo.s == null || s.equals(foo.s));
+		}
+
+	}
+
+}