SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

XD-3242 created spring-cloud-streams-codec

Browse Source 
Fixes gh-3
This commit is contained in:
David Turanski
2015-07-09 10:47:09 -04:00
committed by Dave Syer
parent e6962c514f
commit 0c82a37360
29 changed files with 1442 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -5,6 +5,7 @@ asciidoctor.css
.#*
*#
target/
build/
bin/
_site/
.classpath

25
pom.xml
View File

@@ -21,11 +21,14 @@
<java.version>1.7</java.version>
<spring-boot.version>1.3.0.BUILD-SNAPSHOT</spring-boot.version>
<spring-cloud.version>Brixton.BUILD-SNAPSHOT</spring-cloud.version>
<spring-xd.version>2.0.0.BUILD-SNAPSHOT</spring-xd.version>
<spring-xd.version>1.2.1.BUILD-SNAPSHOT</spring-xd.version>
<spring-framework.version>4.2.0.RC2</spring-framework.version>
<spring-integration.version>4.2.0.M2</spring-integration.version>
</properties>
<modules>
<module>spring-cloud-streams</module>
<module>spring-cloud-streams-codec</module>
<module>spring-cloud-streams-common</module>
<module>spring-xd-runner</module>
<module>spring-xd-samples</module>
</modules>
@@ -48,17 +51,22 @@
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-xd-runner</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-codec</artifactId>
<version>${spring-xd.version}</version>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-codec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
@@ -85,6 +93,11 @@
<artifactId>spring-messaging</artifactId>
<version>${spring-framework.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>${spring-integration.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
<repositories>

53
spring-cloud-streams-codec/pom.xml Normal file
View File

@@ -0,0 +1,53 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-codec</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>spring-cloud-streams-codec</name>
<description>Serialization library used by transport</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-common</artifactId>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo-shaded</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

42
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/AbstractCodec.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* 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.ByteArrayInputStream;
import java.io.IOException;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
/**
* Support class providing convenience methods for codecs.
*
* @author David Turanski
*/
public abstract class AbstractCodec<T> implements Serializer<T>, Deserializer<T> {
/**
* Deserialize a byte array.
*
* @param bytes
* @throws IOException
*/
public T deserialize(byte[] bytes) throws IOException {
return deserialize(new ByteArrayInputStream(bytes));
}
}

89
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/CompositeCodec.java Normal file
View File

@@ -0,0 +1,89 @@
/*
* 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.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Map;
import org.springframework.integration.util.ClassUtils;
import org.springframework.util.Assert;
/**
* A codec that can delegate to one out of many codecs, depending on the type of the object to serialize/deserialize.
*
* @author David Turanski
*/
public class CompositeCodec<P> implements MultiTypeCodec<Object> {
private final MultiTypeCodec<P> defaultCodec;
private final Map<Class<?>, AbstractCodec<P>> delegates;
public CompositeCodec(Map<Class<?>, AbstractCodec<P>> delegates, MultiTypeCodec<P> defaultCodec)
{
Assert.notNull(defaultCodec, "'defaultCodec' cannot be null");
this.defaultCodec = defaultCodec;
this.delegates = delegates;
}
public CompositeCodec(MultiTypeCodec<P> 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<P> 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<P> codec = findDelegate(type);
if (codec != null) {
return codec.deserialize(inputStream);
}
else {
return defaultCodec.deserialize(inputStream, (Class<P>) type);
}
}
@Override
public Object deserialize(byte[] bytes, Class<?> type) throws IOException {
return deserialize(new ByteArrayInputStream(bytes), type);
}
private AbstractCodec<P> findDelegate(Class<?> type) {
if (delegates == null) {
return null;
}
Class<?> clazz = ClassUtils.findClosestMatch(type, delegates.keySet(), false);
return delegates.get(clazz);
}
}

48
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/MultiTypeCodec.java Normal file
View File

@@ -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<T> extends Serializer<T> {
/**
* 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<? extends T> 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<? extends T> type) throws IOException;
}

38
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/SerializationException.java Normal file
View File

@@ -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);
}
}

110
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java Normal file
View File

@@ -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<T> extends AbstractCodec<T> {
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<Object>() {
@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<T>() {
@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) {
}
}

107
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoMultiTypeCodec.java Normal file
View File

@@ -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<T> extends AbstractKryoCodec<T> implements MultiTypeCodec<T> {
/**
* 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<? extends T> 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<? extends T> 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<? extends T> type) throws IOException {
return pool.run(new KryoCallback<T>() {
@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<T> type = (Class<T>) (
(ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
return doDeserialize(kryo, input, type);
}
protected abstract T doDeserialize(Kryo kryo, Input input, Class<? extends T> type);
}

58
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoRegistrar.java Normal file
View File

@@ -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<Registration> 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);
}
}

80
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeKryoRegistrar.java Normal file
View File

@@ -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<KryoRegistrar> delegates;
public CompositeKryoRegistrar(List<KryoRegistrar> delegates) {
super();
this.delegates = delegates;
if (!CollectionUtils.isEmpty(this.delegates)) {
validateRegistrations();
}
}
@Override
public List<Registration> getRegistrations() {
List<Registration> registrations = new ArrayList<>();
for (KryoRegistrar registrar : delegates) {
registrations.addAll(registrar.getRegistrations());
}
return registrations;
}
private void validateRegistrations() {
List<Integer> ids = new ArrayList<>();
List<Class<?>> 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());
}
}
}

38
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileKryoRegistrar.java Normal file
View File

@@ -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<Registration> getRegistrations() {
return Collections.singletonList(new Registration(java.io.File.class, fileSerializer, FILE_REGISTRATION_ID));
}
}

41
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/FileSerializer.java Normal file
View File

@@ -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<File> {
@Override
public void write(Kryo kryo, Output output, File file) {
output.writeString(file.getPath());
}
@Override
public File read(Kryo kryo, Input input, Class<File> type) {
String path = input.readString();
return new File(path);
}
}

72
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassListRegistrar.java Normal file
View File

@@ -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<Class> registeredClasses;
private int initialValue = 50;
/**
* @param classes the list of classes to register
*/
public KryoClassListRegistrar(List<Class> 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<Registration> getRegistrations() {
List<Registration> 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;
}
}

52
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoClassMapRegistrar.java Normal file
View File

@@ -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<Integer, Class<?>> registeredClasses;
public KryoClassMapRegistrar(Map<Integer, Class<?>> kryoRegisteredClasses) {
this.registeredClasses = kryoRegisteredClasses;
}
@Override
public List<Registration> getRegistrations() {
List<Registration> registrations = new ArrayList<>();
if (!CollectionUtils.isEmpty(registeredClasses)) {
for (Map.Entry<Integer, Class<?>> entry : registeredClasses.entrySet()) {
registrations.add(new Registration(entry.getValue(), kryo.getSerializer(entry.getValue()), entry.getKey()));
}
}
return registrations;
}
}

66
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrar.java Normal file
View File

@@ -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<Integer,Class<?>> or a List<Class<?>>. 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<Registration> getRegistrations();
}

39
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoRegistrationRegistrar.java Normal file
View File

@@ -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<Registration> registrations;
public KryoRegistrationRegistrar(List<Registration> registrations) {
this.registrations = registrations != null ? registrations : new ArrayList<Registration>();
}
@Override
public List<Registration> getRegistrations() {
return registrations;
}
}

78
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java Normal file
View File

@@ -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<Object> {
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<KryoRegistrar> 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<? extends Object> type) {
return kryo.readObject(input, type);
}
@Override
protected void configureKryoInstance(Kryo kryo) {
super.configureKryoInstance(kryo);
if (kryoRegistrar != null) {
kryoRegistrar.registerTypes(kryo);
}
}
}

5
spring-cloud-streams-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/package-info.java Normal file
View File

@@ -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;

90
spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/CompositeCodecTests.java Normal file
View File

@@ -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<Object> codec;
@SuppressWarnings({ "unchecked", "rawtypes" })
@Before
public void setup() {
Map<Class<?>, 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;
}
}
}

173
spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoCodecTests.java Normal file
View File

@@ -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<String, Integer> map = new HashMap<String, Integer>();
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<Object, Object> map;
public Foo() {
map = new HashMap<Object, Object>();
}
public void put(Object key, Object value) {
map.put(key, value);
}
public Object get(Object key) {
return map.get(key);
}
}
}

43
spring-cloud-streams-codec/src/test/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/KryoFileCodecTests.java Normal file
View File

@@ -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);
}
}

28
spring-cloud-streams-common/pom.xml Normal file
View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-common</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>spring-cloud-streams-common</name>
<description>Spring Cloud Streams common components</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
</dependencies>
</project>

31
spring-cloud-streams-common/src/main/java/org/springframework/cloud/streams/exception/CloudStreamsRuntimeException.java Normal file
View File

@@ -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);
}
}

4
spring-cloud-streams/pom.xml
View File

@@ -43,8 +43,8 @@
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-codec</artifactId>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-codec</artifactId>
</dependency>
<dependency>

46
spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java
View File

@@ -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<String, KryoRegistrar> kryoRegistrarMap = applicationContext.getBeansOfType(KryoRegistrar
.class);
return new PojoCodec(new ArrayList<>(kryoRegistrarMap.values()));
}
@Bean
public KryoRegistrar fileRegistrar() {
return new FileKryoRegistrar();
}
}
}

26
spring-cloud-streams/src/main/resources/META-INF/spring-cloud-streams/codec.xml
View File

@@ -1,26 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="codec" class="org.springframework.xd.dirt.integration.bus.serializer.CompositeCodec">
<constructor-arg name="delegates">
<map>
<!-- pulls in spring-xd-tuple -->
<!--<entry key="org.springframework.xd.tuple.Tuple">-->
<!--<bean class="org.springframework.xd.tuple.serializer.kryo.TupleCodec"/>-->
<!--</entry>-->
<entry key="java.io.File">
<bean class="org.springframework.xd.dirt.integration.bus.serializer.kryo.FileCodec"/>
</entry>
</map>
</constructor-arg>
<constructor-arg name="defaultCodec" ref="defaultCodec"/>
</bean>
<bean id="defaultCodec" class="org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec">
<constructor-arg ref="kryoRegistrar"/>
</bean>
<bean id="kryoRegistrar" class="org.springframework.xd.dirt.integration.bus.serializer.kryo.KryoNullRegistrar"/>
</beans>

4
spring-xd-runner/pom.xml
View File

@@ -32,8 +32,8 @@
<artifactId>spring-cloud-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-codec</artifactId>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-streams-codec</artifactId>
</dependency>
<dependency>

1
spring-xd-runner/src/main/java/org/springframework/cloud/streams/xd/ModuleOptionsPropertySourceInitializer.java
View File

@@ -117,7 +117,6 @@ ApplicationContextInitializer<ConfigurableApplicationContext> {
// TODO: allow override of this
public DefaultModuleOptionsMetadataResolver defaultResolver() {
DefaultModuleOptionsMetadataResolver defaultResolver = new DefaultModuleOptionsMetadataResolver();
defaultResolver.setShouldCreateModuleClassLoader(false);
return defaultResolver;
}