, AbstractCodec> delegates, MultiTypeCodec
defaultCodec)
+ {
+ Assert.notNull(defaultCodec, "'defaultCodec' cannot be null");
+ this.defaultCodec = defaultCodec;
+ this.delegates = delegates;
+ }
+
+ public CompositeCodec(MultiTypeCodec
defaultCodec) {
+ this(null, defaultCodec);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void serialize(Object object, OutputStream outputStream) throws IOException {
+ Assert.notNull(object, "cannot serialize a null object");
+ AbstractCodec
codec = findDelegate(object.getClass());
+ if (codec != null) {
+ codec.serialize((P) object, outputStream);
+ }
+ else {
+ defaultCodec.serialize((P) object, outputStream);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Object deserialize(InputStream inputStream, Class type) throws IOException {
+ AbstractCodec
codec = findDelegate(type);
+ if (codec != null) {
+ return codec.deserialize(inputStream);
+ }
+ else {
+ return defaultCodec.deserialize(inputStream, (Class
) type);
+ }
+ }
+
+ @Override
+ public Object deserialize(byte[] bytes, Class type) throws IOException {
+ return deserialize(new ByteArrayInputStream(bytes), type);
+ }
+
+ private AbstractCodec
findDelegate(Class type) {
+ if (delegates == null) {
+ return null;
+ }
+
+ Class clazz = ClassUtils.findClosestMatch(type, delegates.keySet(), false);
+ return delegates.get(clazz);
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/MultiTypeCodec.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/MultiTypeCodec.java
new file mode 100644
index 000000000..a4d8028e2
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/MultiTypeCodec.java
@@ -0,0 +1,48 @@
+/*
+ * 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.xd.dirt.integration.bus.serializer;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.springframework.core.serializer.Serializer;
+
+
+/**
+ * Interface for classes that perform both serialization and deserialization.
+ * @author David Turanski
+ */
+public interface MultiTypeCodec extends Serializer {
+
+ /**
+ * Deserialize an object of a given type
+ * @param inputStream the input stream containing the serialized object
+ * @param type the object's class
+ * @return the object
+ * @throws IOException
+ */
+ public abstract T deserialize(InputStream inputStream, Class type) throws IOException;
+
+ /**
+ * Deserialize an object of a given type
+ * @param bytes the byte array containing the serialized object
+ * @param type the object's class
+ * @return the object
+ * @throws IOException
+ */
+ public abstract T deserialize(byte[] bytes, Class type) throws IOException;
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/SerializationException.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/SerializationException.java
new file mode 100644
index 000000000..c9b653e3d
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/SerializationException.java
@@ -0,0 +1,38 @@
+/*
+ * 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.xd.dirt.integration.bus.serializer;
+
+
+import org.springframework.cloud.streams.exception.CloudStreamsRuntimeException;
+
+/**
+ * Thrown when something goes wrong with inter-module communication.
+ *
+ * @author David Turanski
+ */
+@SuppressWarnings("serial")
+public class SerializationException extends CloudStreamsRuntimeException {
+
+ public SerializationException(String message) {
+ super(message);
+ }
+
+ public SerializationException(String message, Throwable t) {
+ super(message, t);
+ }
+
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java
new file mode 100644
index 000000000..12249e157
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java
@@ -0,0 +1,110 @@
+/*
+ * 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.pool.KryoCallback;
+import com.esotericsoftware.kryo.pool.KryoFactory;
+import com.esotericsoftware.kryo.pool.KryoPool;
+
+import org.springframework.util.Assert;
+import org.springframework.xd.dirt.integration.bus.serializer.AbstractCodec;
+
+/**
+ * Base class for Codecs using {@link com.esotericsoftware.kryo.Kryo}
+ *
+ * @author David Turanski
+ */
+public abstract class AbstractKryoCodec extends AbstractCodec {
+
+ private final KryoFactory factory;
+
+ protected final KryoPool pool;
+
+ protected AbstractKryoCodec() {
+ factory = new KryoFactory() {
+ public Kryo create() {
+ Kryo kryo = new Kryo();
+ // configure kryo instance, customize settings
+ configureKryoInstance(kryo);
+ return kryo;
+ }
+ };
+ // Build pool with SoftReferences enabled (optional)
+ pool = new KryoPool.Builder(factory).softReferences().build();
+ }
+
+ /**
+ * Serialize an object using an existing output stream
+ *
+ * @param object the object to be serialized
+ * @param outputStream the output stream, e.g. a FileOutputStream
+ * @throws IOException
+ */
+ @Override
+ public void serialize(final T object, OutputStream outputStream) throws IOException {
+ Assert.notNull(outputStream, "'outputSteam' cannot be null");
+ final Output output = new Output(outputStream);
+ try {
+ pool.run(new KryoCallback() {
+ @Override
+ public Object execute(Kryo kryo) {
+ doSerialize(kryo, object, output);
+ return Void.class;
+ }
+ });
+ } finally {
+ output.close();
+ }
+ }
+
+ /**
+ * Deserialize an object when the type is known
+ *
+ * @param inputStream the input stream containing the serialized object
+ * @return the object
+ * @throws IOException
+ */
+ @Override
+ public T deserialize(InputStream inputStream) throws IOException {
+ final Input input = new Input(inputStream);
+ try {
+ T result = pool.run(new KryoCallback() {
+ @Override
+ public T execute(Kryo kryo) {
+ return doDeserialize(kryo, input);
+ }
+ });
+ return result;
+ } finally {
+ input.close();
+ }
+ }
+
+ protected abstract void doSerialize(Kryo kryo, T object, Output output);
+
+ protected abstract T doDeserialize(Kryo kryo, Input input);
+
+ protected void configureKryoInstance(Kryo kryo) {
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoMultiTypeCodec.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoMultiTypeCodec.java
new file mode 100644
index 000000000..89d9bf3bf
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoMultiTypeCodec.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2013-2015 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.ParameterizedType;
+
+import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.pool.KryoCallback;
+
+/**
+ * Base class for Codecs using {@link com.esotericsoftware.kryo.Kryo} to serialize arbitrary types
+ *
+ * @author David Turanski
+ * @since 1.0
+ */
+abstract class AbstractKryoMultiTypeCodec extends AbstractKryoCodec implements MultiTypeCodec {
+
+ /**
+ * Deserialize an object of a given type given a byte array
+ *
+ * @param bytes the byte array containing the serialized object
+ * @param type the object's class
+ * @return the object
+ * @throws IOException
+ */
+ @Override
+ public T deserialize(byte[] bytes, Class type) throws IOException {
+ final Input input = new Input(bytes);
+ try {
+ return deserialize(input, type);
+ } finally {
+ input.close();
+ }
+ }
+
+ /**
+ * Deserialize an object of a given type given an InputStream
+ *
+ * @param inputStream the input stream containing the serialized object
+ * @param type the object's class
+ * @return the object
+ * @throws IOException
+ */
+ @Override
+ public T deserialize(InputStream inputStream, final Class type) throws IOException {
+ final Input input = new Input(inputStream);
+ try {
+ return deserialize(input, type);
+ } finally {
+ input.close();
+ }
+ }
+
+ /**
+ * Deserialize an object of a given type given an Input.
+ *
+ * @param input the Kryo input stream containing the serialized object
+ * @param type the object's class
+ * @return the object
+ * @throws IOException
+ */
+ protected T deserialize(final Input input, final Class type) throws IOException {
+ return pool.run(new KryoCallback() {
+
+ @Override
+ public T execute(Kryo kryo) {
+ return doDeserialize(kryo, input, type);
+ }
+ });
+ }
+
+ /**
+ * Infers the type from this class's generic type argument
+ * @param kryo the Kryo
+ * @param input the input
+ * @return the object
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ protected T doDeserialize(Kryo kryo, Input input) {
+ Class type = (Class) (
+ (ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
+ return doDeserialize(kryo, input, type);
+ }
+
+ protected abstract T doDeserialize(Kryo kryo, Input input, Class type);
+
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoRegistrar.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoRegistrar.java
new file mode 100644
index 000000000..4535a02b6
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoRegistrar.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2015 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.util.List;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Registration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.xd.dirt.integration.bus.serializer.SerializationException;
+
+/**
+ * @author David Turanski
+ */
+public abstract class AbstractKryoRegistrar implements KryoRegistrar {
+ protected final Logger log = LoggerFactory.getLogger(this.getClass());
+
+ protected final static Kryo kryo = new Kryo();
+
+ @Override
+ public void registerTypes(Kryo kryo) {
+ for (Registration registration : getRegistrations()) {
+ register(kryo, registration);
+ }
+ }
+
+ public abstract List getRegistrations();
+
+ protected void register(Kryo kryo, Registration registration) {
+ int id = registration.getId();
+
+ Registration existing = kryo.getRegistration(id);
+
+ if (existing != null) {
+ throw new SerializationException(String.format("registration already exists %s", existing));
+ }
+
+ log.info("registering {} with serializer {}", registration, registration.getSerializer().getClass()
+ .getName());
+
+ kryo.register(registration);
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeKryoRegistrar.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeKryoRegistrar.java
new file mode 100644
index 000000000..5f7d4ca8b
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeKryoRegistrar.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright 2015 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.esotericsoftware.kryo.Registration;
+
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.xd.dirt.integration.bus.serializer.SerializationException;
+
+/**
+ * A {@link KryoRegistrar} that delegates and validates
+ * registrations across all components.
+ * @author David Turanski
+ * @since 1.2
+ */
+public class CompositeKryoRegistrar extends AbstractKryoRegistrar {
+
+ private final List delegates;
+
+ public CompositeKryoRegistrar(List delegates) {
+ super();
+ this.delegates = delegates;
+
+ if (!CollectionUtils.isEmpty(this.delegates)) {
+ validateRegistrations();
+ }
+ }
+
+ @Override
+ public List getRegistrations() {
+ List registrations = new ArrayList<>();
+ for (KryoRegistrar registrar : delegates) {
+ registrations.addAll(registrar.getRegistrations());
+ }
+ return registrations;
+ }
+
+ private void validateRegistrations() {
+ List ids = new ArrayList<>();
+ List> types = new ArrayList<>();
+
+ for (Registration registration : getRegistrations()) {
+ Assert.isTrue(registration.getId() >= MIN_REGISTRATION_VALUE, "registration ID must be >= " +
+ MIN_REGISTRATION_VALUE);
+ if (ids.contains(registration.getId())) {
+ throw new SerializationException(String.format("Duplicate registration ID found: %d",
+ registration.getId()));
+ }
+ ids.add(registration.getId());
+
+ if (types.contains(registration.getType())) {
+ throw new SerializationException(String.format("Duplicate registration found for type: %s",
+ registration.getType()));
+ }
+ types.add(registration.getType());
+
+ log.info("configured Kryo registration {} with serializer {}", registration,
+ registration.getSerializer().getClass().getName());
+
+ }
+
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileKryoRegistrar.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileKryoRegistrar.java
new file mode 100644
index 000000000..1f483540b
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileKryoRegistrar.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2015 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.util.Collections;
+import java.util.List;
+
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * A {@link KryoRegistrar} used to register a File serializer.
+ * @author David Turanski
+ * @since 1.2
+ */
+public class FileKryoRegistrar extends AbstractKryoRegistrar {
+
+ private final static int FILE_REGISTRATION_ID = 40;
+
+ private final FileSerializer fileSerializer = new FileSerializer();
+
+ @Override
+ public List getRegistrations() {
+ return Collections.singletonList(new Registration(java.io.File.class, fileSerializer, FILE_REGISTRATION_ID));
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileSerializer.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileSerializer.java
new file mode 100644
index 000000000..607af6444
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileSerializer.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2015 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.io.File;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+/**
+ * @author David Turanski
+ * @since 1.2
+ */
+public class FileSerializer extends Serializer {
+
+ @Override
+ public void write(Kryo kryo, Output output, File file) {
+ output.writeString(file.getPath());
+ }
+
+ @Override
+ public File read(Kryo kryo, Input input, Class type) {
+ String path = input.readString();
+ return new File(path);
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassListRegistrar.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassListRegistrar.java
new file mode 100644
index 000000000..a90ddd1c6
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassListRegistrar.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright 2015 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.esotericsoftware.kryo.Registration;
+
+
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * A {@link KryoRegistrar} used to register a
+ * list of Java classes. This assigns a sequential registration ID starting with an initial value (50 by default), but
+ * may be configured. This is easiest to set up but requires that every server node be configured with the identical
+ * list in the same order.
+ * @author David Turanski
+ * @since 1.1
+ */
+public class KryoClassListRegistrar extends AbstractKryoRegistrar {
+
+ private final List registeredClasses;
+
+ private int initialValue = 50;
+
+ /**
+ * @param classes the list of classes to register
+ */
+ public KryoClassListRegistrar(List classes) {
+ this.registeredClasses = classes;
+ }
+
+ /**
+ * Set the inital ID value. Classes in the list will be sequentially assigned an ID starting with this value
+ * (default is 50).
+ * @param initialValue the initial value
+ */
+ public void setInitialValue(int initialValue) {
+ Assert.isTrue(initialValue >= MIN_REGISTRATION_VALUE, "'initialValue' must be >= " +
+ MIN_REGISTRATION_VALUE);
+ this.initialValue = initialValue;
+ }
+
+
+ @Override
+ public List getRegistrations() {
+ List registrations = new ArrayList<>();
+ if (!CollectionUtils.isEmpty(registeredClasses)) {
+ for (int i = 0; i < registeredClasses.size(); i++) {
+ registrations.add(new Registration(registeredClasses.get(i), kryo.getSerializer(registeredClasses.get
+ (i)), i + initialValue));
+ }
+ }
+ return registrations;
+ }
+
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassMapRegistrar.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassMapRegistrar.java
new file mode 100644
index 000000000..a26e24e64
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassMapRegistrar.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2014 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import com.esotericsoftware.kryo.Registration;
+
+import org.springframework.util.CollectionUtils;
+
+
+/**
+ * A {@link KryoRegistrar} implementation backed by a Map
+ * used to explicitly set the registration ID for each class.
+ * @author David Turanski
+ * @since 1.1
+ */
+public class KryoClassMapRegistrar extends AbstractKryoRegistrar {
+
+ final private Map> registeredClasses;
+
+ public KryoClassMapRegistrar(Map> kryoRegisteredClasses) {
+ this.registeredClasses = kryoRegisteredClasses;
+ }
+
+
+ @Override
+ public List getRegistrations() {
+ List registrations = new ArrayList<>();
+ if (!CollectionUtils.isEmpty(registeredClasses)) {
+ for (Map.Entry> entry : registeredClasses.entrySet()) {
+ registrations.add(new Registration(entry.getValue(), kryo.getSerializer(entry.getValue()), entry.getKey()));
+ }
+ }
+ return registrations;
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrar.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrar.java
new file mode 100644
index 000000000..15af67a48
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrar.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2015 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.util.List;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * Strategy interface used by {@link PojoCodec} to register
+ * classes consistently across {@link Kryo} instances. An XD user may register an instance of this type in the Spring XD
+ * Application Context to enable kryo class registration which results in efficiency gains if you know the types your
+ * application needs in advance. Note that Kryo serialization only applies to types used as message payloads in XD
+ * streams.
+ * By default, user defined types are not registered to Kryo. Registration allows a unique ID (small positive integer is
+ * ideal) to represent the type in the byte stream. In a distributed environment, all Kryo instances must maintain the
+ * same registration state in order to properly take advantage of this feature.
+ * This is can result in better performance in demanding situations, but requires some care to maintain. Only use this
+ * if you really need it. Otherwise, it is a great example of premature optimization.
+ * This interface applies a strategy to register a statically configured, one-to-one mapping of a Java type to an
+ * integer. Basic implementations are provided backed by a Map> or a List>. These are simple
+ * and require the user to manually configure a bean in each XD server and ensure that the configuration is always
+ * consistent.*
+ * The container looks in classpath*:META-INF/spring-xd/xd/bus/ext/*.xml for an instance of this type named
+ * "kryoRegistrar". The KryoRegistrar provides the registration mapping and the strategy to apply the mapping to every
+ * Kryo instance. Note that statically declared Java types must also be present in the XD class path (xd/lib) else the
+ * container will fail to initialize. Only one instance may be registered and identically configured across all
+ * containers.
+ *
+ * @author David Turanski
+ * @since 1.1
+ */
+public interface KryoRegistrar {
+
+ static final int MIN_REGISTRATION_VALUE = 10;
+
+ /**
+ * This method is invoked by the {@link PojoCodec} and
+ * applied to the {@link Kryo} instance whenever one is provided. This is currently done using an object pool so it
+ * is inevitable that this method will be invoked repeatedly on the same instance. Kryo registration is idempotent,
+ * but this could become inefficient if registering a large amount of types.
+ *
+ * @param kryo the provided instance
+ */
+ void registerTypes(Kryo kryo);
+
+ /**
+ *
+ * @return the list of {@link com.esotericsoftware.kryo.Registration} provided
+ */
+ List getRegistrations();
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrationRegistrar.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrationRegistrar.java
new file mode 100644
index 000000000..9e69f3e3c
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrationRegistrar.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2015 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * A {@link KryoRegistrar } implementation backed by a List of {@link com.esotericsoftware.kryo.Registration}.
+ * @author David Turanski
+ * @since 1.2
+ */
+public class KryoRegistrationRegistrar extends AbstractKryoRegistrar {
+ private final List registrations;
+
+ public KryoRegistrationRegistrar(List registrations) {
+ this.registrations = registrations != null ? registrations : new ArrayList();
+ }
+
+ @Override
+ public List getRegistrations() {
+ return registrations;
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java
new file mode 100644
index 000000000..71800c78a
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java
@@ -0,0 +1,78 @@
+/*
+ * 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.xd.dirt.integration.bus.serializer.kryo;
+
+
+import java.util.Collections;
+import java.util.List;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Kryo Codec that can serialize and deserialize arbitrary types. Classes and associated
+ * {@link com.esotericsoftware.kryo.Serializer}s may be registered via
+ * {@link KryoRegistrar}s.
+ * @author David Turanski
+ * @since 1.0
+ */
+public class PojoCodec extends AbstractKryoMultiTypeCodec {
+ private final CompositeKryoRegistrar kryoRegistrar;
+
+ public PojoCodec() {
+ this.kryoRegistrar = null;
+ }
+
+ /**
+ * Create an instance with a single KryoRegistrar.
+ * @param kryoRegistrar
+ */
+ public PojoCodec(KryoRegistrar kryoRegistrar) {
+ this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null);
+ }
+
+ /**
+ * Create an instance with zero to many KryoRegistrars.
+ * @param kryoRegistrars
+ */
+ public PojoCodec(List kryoRegistrars) {
+ kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null :
+ new CompositeKryoRegistrar(kryoRegistrars);
+ }
+
+ @Override
+ protected void doSerialize(Kryo kryo, Object object, Output output) {
+ kryo.writeObject(output, object);
+ }
+
+
+ @Override
+ protected Object doDeserialize(Kryo kryo, Input input, Class type) {
+ return kryo.readObject(input, type);
+ }
+
+ @Override
+ protected void configureKryoInstance(Kryo kryo) {
+ super.configureKryoInstance(kryo);
+ if (kryoRegistrar != null) {
+ kryoRegistrar.registerTypes(kryo);
+ }
+ }
+}
diff --git a/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/package-info.java b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/package-info.java
new file mode 100644
index 000000000..a80e77575
--- /dev/null
+++ b/spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/package-info.java
@@ -0,0 +1,5 @@
+/**
+ * Contains classes that provide kryo serialization support to/from {@link org.springframework.xd.dirt.integration.bus.MessageBus}.
+ */
+
+package org.springframework.xd.dirt.integration.bus.serializer.kryo;
diff --git a/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeCodecTests.java b/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeCodecTests.java
new file mode 100644
index 000000000..789ad1729
--- /dev/null
+++ b/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeCodecTests.java
@@ -0,0 +1,90 @@
+/*
+ * 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.xd.dirt.integration.bus.serializer.kryo;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.xd.dirt.integration.bus.serializer.AbstractCodec;
+import org.springframework.xd.dirt.integration.bus.serializer.CompositeCodec;
+import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
+
+/**
+ * @author David Turanski
+ */
+public class CompositeCodecTests {
+
+ private MultiTypeCodec codec;
+
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Before
+ public void setup() {
+ Map, AbstractCodec> codecs = new HashMap<>();
+ codec = new CompositeCodec(codecs, new PojoCodec());
+ }
+
+ @Test
+ public void testPojoSerialization() throws IOException {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("hello", 123);
+ codec.serialize(foo, bos);
+ SomeClassWithNoDefaultConstructors foo2 = (SomeClassWithNoDefaultConstructors) codec.deserialize(
+ bos.toByteArray(),
+ SomeClassWithNoDefaultConstructors.class);
+ assertEquals(foo, foo2);
+ }
+
+ static class SomeClassWithNoDefaultConstructors {
+
+ private String val1;
+
+ private int val2;
+
+ public SomeClassWithNoDefaultConstructors(String val1) {
+ this.val1 = val1;
+ }
+
+ public SomeClassWithNoDefaultConstructors(String val1, int val2) {
+ this.val1 = val1;
+ this.val2 = val2;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof SomeClassWithNoDefaultConstructors)) {
+ return false;
+ }
+ SomeClassWithNoDefaultConstructors that = (SomeClassWithNoDefaultConstructors) other;
+ return (this.val1.equals(that.val1) && val2 == that.val2);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = this.val1.hashCode();
+ result = 31 * result + val2;
+ return result;
+ }
+ }
+}
diff --git a/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoCodecTests.java b/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoCodecTests.java
new file mode 100644
index 000000000..94601713d
--- /dev/null
+++ b/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoCodecTests.java
@@ -0,0 +1,173 @@
+/*
+ * 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.xd.dirt.integration.bus.serializer.kryo;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+/**
+ * @author David Turanski
+ * @since 1.0
+ */
+public class KryoCodecTests {
+
+ @Test
+ public void testStringSerialization() throws IOException {
+ String str = "hello";
+ PojoCodec serializer = new PojoCodec();
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+
+ serializer.serialize(str, bos);
+
+ String s2 = (String)serializer.deserialize(bos.toByteArray(), String.class);
+ assertEquals(str, s2);
+ }
+
+ @Test
+ public void testSerializationWithStreams() throws IOException {
+ String str = "hello";
+ File file = new File("test.ser");
+ PojoCodec serializer = new PojoCodec();
+ FileOutputStream fos = new FileOutputStream(file);
+ serializer.serialize(str, fos);
+ fos.close();
+
+ FileInputStream fis = new FileInputStream(file);
+ String s2 = (String) serializer.deserialize(fis, String.class);
+ file.delete();
+ assertEquals(str, s2);
+ }
+
+ @Test
+ public void testPojoSerialization() throws IOException {
+ PojoCodec serializer = new PojoCodec();
+ SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("foo", 123);
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ serializer.serialize(foo, bos);
+ Object foo2 = serializer.deserialize(bos.toByteArray(), SomeClassWithNoDefaultConstructors.class);
+ assertEquals(foo, foo2);
+ }
+
+ static class SomeClassWithNoDefaultConstructors {
+
+ private String val1;
+
+ private int val2;
+
+ public SomeClassWithNoDefaultConstructors(String val1, int val2) {
+ this.val1 = val1;
+ this.val2 = val2;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof SomeClassWithNoDefaultConstructors)) {
+ return false;
+ }
+ SomeClassWithNoDefaultConstructors that = (SomeClassWithNoDefaultConstructors) other;
+ return (this.val1.equals(that.val1) && val2 == that.val2);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = val1.hashCode();
+ result = 31 * result + val2;
+ return result;
+ }
+ }
+
+ @Test
+ public void testPrimitiveSerialization() throws IOException {
+ PojoCodec serializer = new PojoCodec();
+
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ serializer.serialize(true, bos);
+ boolean b = (Boolean) serializer.deserialize(bos.toByteArray(), Boolean.class);
+ assertEquals(true, b);
+ b = (Boolean) serializer.deserialize(bos.toByteArray(), boolean.class);
+ assertEquals(true, b);
+
+ bos = new ByteArrayOutputStream();
+ serializer.serialize(3.14159, bos);
+
+ double d = (Double) serializer.deserialize(bos.toByteArray(), double.class);
+ assertEquals(3.14159, d, 0.00001);
+
+ bos = new ByteArrayOutputStream();
+ serializer.serialize(new Double(3.14159), bos);
+
+ d = (Double) serializer.deserialize(bos.toByteArray(), Double.class);
+ assertEquals(3.14159, d, 0.00001);
+
+ }
+
+ @Test
+ public void testMapSerialization() throws IOException {
+ PojoCodec serializer = new PojoCodec();
+ Map map = new HashMap();
+ map.put("one", 1);
+ map.put("two", 2);
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ serializer.serialize(map, bos);
+ Map m2 = (Map) serializer.deserialize(bos.toByteArray(), HashMap.class);
+ assertEquals(2, m2.size());
+ assertEquals(1, m2.get("one"));
+ assertEquals(2, m2.get("two"));
+ }
+
+ @Test
+ public void testComplexObjectSerialization() throws IOException {
+ PojoCodec serializer = new PojoCodec();
+ Foo foo = new Foo();
+ foo.put("one", 1);
+ foo.put("two", 2);
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ serializer.serialize(foo, bos);
+
+ Foo foo2 = (Foo) serializer.deserialize(bos.toByteArray(), Foo.class);
+ assertEquals(1, foo2.get("one"));
+ assertEquals(2, foo2.get("two"));
+ }
+
+
+
+ static class Foo {
+
+ private Map map;
+
+ public Foo() {
+ map = new HashMap();
+ }
+
+ public void put(Object key, Object value) {
+ map.put(key, value);
+ }
+
+ public Object get(Object key) {
+ return map.get(key);
+ }
+ }
+}
diff --git a/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoFileCodecTests.java b/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoFileCodecTests.java
new file mode 100644
index 000000000..6bd38eb88
--- /dev/null
+++ b/spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoFileCodecTests.java
@@ -0,0 +1,43 @@
+/*
+ * 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.xd.dirt.integration.bus.serializer.kryo;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+
+/**
+ * @author David Turanski
+ */
+public class KryoFileCodecTests {
+
+ @Test
+ public void test() throws IOException {
+
+ PojoCodec pc = new PojoCodec(new FileKryoRegistrar());
+ File file = new File("/foo/bar");
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ pc.serialize(file, bos);
+ File file2 = (File) pc.deserialize(bos.toByteArray(), File.class);
+ assertEquals(file, file2);
+ }
+}
diff --git a/spring-cloud-streams-common/pom.xml b/spring-cloud-streams-common/pom.xml
new file mode 100644
index 000000000..d8084132a
--- /dev/null
+++ b/spring-cloud-streams-common/pom.xml
@@ -0,0 +1,28 @@
+
+
+ 4.0.0
+
+ org.springframework.cloud
+ spring-cloud-streams-common
+ 1.0.0.BUILD-SNAPSHOT
+ jar
+
+ spring-cloud-streams-common
+ Spring Cloud Streams common components
+
+
+ org.springframework.cloud
+ spring-cloud-streams-parent
+ 1.0.0.BUILD-SNAPSHOT
+
+
+
+ UTF-8
+ 1.8
+
+
+
+
+
+
diff --git a/spring-cloud-streams-common/src/main/java/org/springframework/cloud/streams/exception/CloudStreamsRuntimeException.java b/spring-cloud-streams-common/src/main/java/org/springframework/cloud/streams/exception/CloudStreamsRuntimeException.java
new file mode 100644
index 000000000..fac3dd2be
--- /dev/null
+++ b/spring-cloud-streams-common/src/main/java/org/springframework/cloud/streams/exception/CloudStreamsRuntimeException.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2015 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.cloud.streams.exception;
+
+/**
+ * @author David Turanski
+ */
+public class CloudStreamsRuntimeException extends RuntimeException {
+
+ public CloudStreamsRuntimeException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public CloudStreamsRuntimeException(String message) {
+ super(message);
+ }
+
+}
diff --git a/spring-cloud-streams/pom.xml b/spring-cloud-streams/pom.xml
index a24391b03..dd1d95264 100644
--- a/spring-cloud-streams/pom.xml
+++ b/spring-cloud-streams/pom.xml
@@ -43,8 +43,8 @@
- org.springframework.xd
- spring-xd-codec
+ org.springframework.cloud
+ spring-cloud-streams-codec
diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java
index cb2ff2c7f..3f4713606 100644
--- a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java
@@ -15,14 +15,17 @@
*/
package org.springframework.cloud.streams.config;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedHashSet;
+import java.util.Map;
import java.util.Properties;
import java.util.Set;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
+
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.target.LazyInitTargetSource;
import org.springframework.beans.factory.BeanFactoryUtils;
@@ -35,21 +38,26 @@ import org.springframework.cloud.streams.adapter.Input;
import org.springframework.cloud.streams.adapter.InputChannelBinding;
import org.springframework.cloud.streams.adapter.Output;
import org.springframework.cloud.streams.adapter.OutputChannelBinding;
+import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
+import org.springframework.xd.dirt.integration.bus.serializer.kryo.FileKryoRegistrar;
+import org.springframework.xd.dirt.integration.bus.serializer.kryo.KryoRegistrar;
+import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
import org.springframework.cloud.streams.endpoint.ChannelsEndpoint;
+import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.ImportResource;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import org.springframework.xd.dirt.integration.bus.MessageBus;
import org.springframework.xd.dirt.integration.bus.MessageBusAwareRouterBeanPostProcessor;
+
/**
* @author Dave Syer
- *
+ * @author David Turanski
*/
@Configuration
-@ImportResource("classpath*:/META-INF/spring-cloud-streams/codec.xml")
+
public class ChannelBindingAdapterConfiguration {
@Autowired
@@ -58,11 +66,11 @@ public class ChannelBindingAdapterConfiguration {
@Autowired
private ListableBeanFactory beanFactory;
- @Autowired(required=false)
+ @Autowired(required = false)
@Input
private ChannelLocator inputChannelLocator;
- @Autowired(required=false)
+ @Autowired(required = false)
@Output
private ChannelLocator outputChannelLocator;
@@ -74,10 +82,10 @@ public class ChannelBindingAdapterConfiguration {
ChannelBindingAdapter adapter = new ChannelBindingAdapter(this.module, this.messageBus);
adapter.setOutputChannels(getOutputChannels());
adapter.setInputChannels(getInputChannels());
- if (this.inputChannelLocator!=null) {
+ if (this.inputChannelLocator != null) {
adapter.setInputChannelLocator(this.inputChannelLocator);
}
- if (this.outputChannelLocator!=null) {
+ if (this.outputChannelLocator != null) {
adapter.setOutputChannelLocator(this.outputChannelLocator);
}
return adapter;
@@ -140,7 +148,7 @@ public class ChannelBindingAdapterConfiguration {
source.setBeanFactory(beanFactory);
factory.setTargetSource(source);
factory.addAdvice(new PassthruAdvice());
- factory.setInterfaces(new Class[] { type });
+ factory.setInterfaces(new Class[] {type});
@SuppressWarnings("unchecked")
T proxy = (T) factory.getProxy();
return proxy;
@@ -168,9 +176,29 @@ public class ChannelBindingAdapterConfiguration {
@Configuration
@ConditionalOnMissingBean(ChannelBindingProperties.class)
protected static class ModulePropertiesConfiguration {
- @Bean(name="spring.cloud.channels.CONFIGURATION_PROPERTIES")
+ @Bean(name = "spring.cloud.channels.CONFIGURATION_PROPERTIES")
public ChannelBindingProperties moduleProperties() {
return new ChannelBindingProperties();
}
}
+
+
+ protected static class CodecConfiguration {
+ @Autowired
+ ApplicationContext applicationContext;
+
+ @Bean
+ @ConditionalOnMissingBean(name = "codec")
+ public MultiTypeCodec codec() {
+ Map kryoRegistrarMap = applicationContext.getBeansOfType(KryoRegistrar
+ .class);
+ return new PojoCodec(new ArrayList<>(kryoRegistrarMap.values()));
+ }
+
+ @Bean
+ public KryoRegistrar fileRegistrar() {
+ return new FileKryoRegistrar();
+ }
+ }
+
}
diff --git a/spring-cloud-streams/src/main/resources/META-INF/spring-cloud-streams/codec.xml b/spring-cloud-streams/src/main/resources/META-INF/spring-cloud-streams/codec.xml
deleted file mode 100644
index f88e4255d..000000000
--- a/spring-cloud-streams/src/main/resources/META-INF/spring-cloud-streams/codec.xml
+++ /dev/null
@@ -1,26 +0,0 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
diff --git a/spring-xd-runner/pom.xml b/spring-xd-runner/pom.xml
index 61afd2558..0b7285565 100644
--- a/spring-xd-runner/pom.xml
+++ b/spring-xd-runner/pom.xml
@@ -32,8 +32,8 @@
spring-cloud-streams
- org.springframework.xd
- spring-xd-codec
+ org.springframework.cloud
+ spring-cloud-streams-codec
diff --git a/spring-xd-runner/src/main/java/org/springframework/cloud/streams/xd/ModuleOptionsPropertySourceInitializer.java b/spring-xd-runner/src/main/java/org/springframework/cloud/streams/xd/ModuleOptionsPropertySourceInitializer.java
index 81b5a9b17..93414a8bf 100644
--- a/spring-xd-runner/src/main/java/org/springframework/cloud/streams/xd/ModuleOptionsPropertySourceInitializer.java
+++ b/spring-xd-runner/src/main/java/org/springframework/cloud/streams/xd/ModuleOptionsPropertySourceInitializer.java
@@ -117,7 +117,6 @@ ApplicationContextInitializer {
// TODO: allow override of this
public DefaultModuleOptionsMetadataResolver defaultResolver() {
DefaultModuleOptionsMetadataResolver defaultResolver = new DefaultModuleOptionsMetadataResolver();
- defaultResolver.setShouldCreateModuleClassLoader(false);
return defaultResolver;
}