diff --git a/build.gradle b/build.gradle index f8bf7c44ee..4adabb170e 100644 --- a/build.gradle +++ b/build.gradle @@ -113,6 +113,7 @@ subprojects { subproject -> jsonpathVersion = '2.0.0' junitVersion = '4.11' jythonVersion = '2.5.3' + kryoShadedVersion = '3.0.0' log4jVersion = '1.2.17' mockitoVersion = '1.9.5' mysqlVersion = '5.1.34' @@ -304,6 +305,7 @@ project('spring-integration-core') { compile("com.fasterxml.jackson.core:jackson-databind:$jackson2Version", optional) compile("com.jayway.jsonpath:json-path:$jsonpathVersion", optional) compile("io.fastjson:boon:$boonVersion", optional) + compile("com.esotericsoftware:kryo-shaded:$kryoShadedVersion", optional) testCompile ("org.aspectj:aspectjweaver:$aspectjVersion") testCompile ("net.openhft:chronicle:$chronicleVersion") diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/Codec.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/Codec.java new file mode 100644 index 0000000000..9637933668 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/Codec.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.integration.codec; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +/** + * Interface for classes that perform both encode (serialize) and decode (deserialize) on multiple classes. + * + * @author David Turanski + * @since 4.2 + */ +public interface Codec { + /** + * Encode (encode) an object to an OutputStream + * @param object the object to encode + * @param outputStream the OutputStream + * @throws IOException if the operation fails + */ + void encode(Object object, OutputStream outputStream) throws IOException; + + /** + * Encode an object to a byte array + * @param object the object to encode + * @return the bytes + * @throws IOException if the operation fails + */ + byte[] encode(Object object) throws IOException; + + /** + * Decode an object of a given type + * @param inputStream the input stream containing the encoded object + * @param type the object's class + * @param the object's type + * @return the object + * @throws IOException if the operation fails + */ + T decode(InputStream inputStream, Class type) throws IOException; + + /** + * Decode an object of a given type + * @param bytes the byte array containing the encoded object + * @param type the object's class + * @param the object's type + * @return the object + * @throws IOException if the operation fails + */ + T decode(byte[] bytes, Class type) throws IOException; + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/CompositeCodec.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/CompositeCodec.java new file mode 100644 index 0000000000..dbfa070297 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/CompositeCodec.java @@ -0,0 +1,102 @@ +/* + * 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.integration.codec; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.HashMap; +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, each mapped to a class. + * @author David Turanski + * @since 4.2 + */ +public class CompositeCodec implements Codec { + + private final Codec defaultCodec; + + private final Map, Codec> delegates; + + public CompositeCodec(Map, Codec> delegates, Codec defaultCodec) { + Assert.notNull(defaultCodec, "'defaultCodec' cannot be null"); + this.defaultCodec = defaultCodec; + this.delegates = new HashMap, Codec>(delegates); + } + + public CompositeCodec(Codec defaultCodec) { + this(null, defaultCodec); + } + + @Override + public void encode(Object object, OutputStream outputStream) throws IOException { + Assert.notNull(object, "cannot encode a null object"); + Assert.notNull(outputStream, "'outputStream' cannot be null"); + Codec codec = findDelegate(object.getClass()); + if (codec != null) { + codec.encode(object, outputStream); + } + else { + this.defaultCodec.encode(object, outputStream); + } + } + + @Override + public byte[] encode(Object object) throws IOException { + Assert.notNull(object, "cannot encode a null object"); + Codec codec = findDelegate(object.getClass()); + if (codec != null) { + return codec.encode(object); + } + else { + return this.defaultCodec.encode(object); + } + } + + @Override + public T decode(InputStream inputStream, Class type) throws IOException { + Assert.notNull(inputStream, "'inputStream' cannot be null"); + Assert.notNull(type, "'type' cannot be null"); + Codec codec = findDelegate(type); + if (codec != null) { + return codec.decode(inputStream, type); + } + else { + return this.defaultCodec.decode(inputStream, type); + } + } + + @Override + public T decode(byte[] bytes, Class type) throws IOException { + return decode(new ByteArrayInputStream(bytes), type); + } + + private Codec findDelegate(Class type) { + if (this.delegates == null) { + return null; + } + + Class clazz = ClassUtils.findClosestMatch(type, this.delegates.keySet(), false); + return this.delegates.get(clazz); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoCodec.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoCodec.java new file mode 100644 index 0000000000..887d60752e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoCodec.java @@ -0,0 +1,141 @@ +/* + * 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.integration.codec.kryo; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +import org.springframework.integration.codec.Codec; +import org.springframework.util.Assert; + +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; + +/** + * Base class for {@link Codec}s using {@link Kryo}. + * Manages pooled {@link Kryo} instances. + * + * @author David Turanski + * @since 4.2 + */ +public abstract class AbstractKryoCodec implements Codec { + + protected final KryoPool pool; + + protected AbstractKryoCodec() { + KryoFactory 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(); + } + + @Override + public void encode(final Object object, OutputStream outputStream) throws IOException { + Assert.notNull(object, "cannot encode a null object"); + Assert.notNull(outputStream, "'outputSteam' cannot be null"); + final Output output = (outputStream instanceof Output ? (Output) outputStream : new Output(outputStream)); + this.pool.run(new KryoCallback() { + + public Object execute(Kryo kryo) { + doEncode(kryo, object, output); + return Void.class; + } + + }); + output.close(); + } + + @Override + public T decode(byte[] bytes, Class type) throws IOException { + Assert.notNull(bytes, "'bytes' cannot be null"); + final Input input = new Input(bytes); + try { + return decode(input, type); + } + finally { + input.close(); + } + } + + @Override + public T decode(InputStream inputStream, final Class type) throws IOException { + Assert.notNull(inputStream, "'inputStream' cannot be null"); + Assert.notNull(type, "'type' cannot be null"); + final Input input = (inputStream instanceof Input ? (Input) inputStream : new Input(inputStream)); + T result = null; + try { + result = this.pool.run(new KryoCallback() { + + public T execute(Kryo kryo) { + return doDecode(kryo, input, type); + } + + }); + } + finally { + input.close(); + } + return result; + } + + @Override + public byte[] encode(Object object) throws IOException { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + encode(object, bos); + byte[] bytes = bos.toByteArray(); + bos.close(); + return bytes; + } + + /** + * Subclasses implement this method to encode with Kryo. + * @param kryo the Kryo instance + * @param object the object to encode + * @param output the Kryo Output instance + */ + protected abstract void doEncode(Kryo kryo, Object object, Output output); + + /** + * Subclasses implement this method to decode with Kryo. + * @param kryo the Kryo instance + * @param input the Kryo Input instance + * @param type the class of the decoded object + * @param the type for decoded object + * @return the decoded object + */ + protected abstract T doDecode(Kryo kryo, Input input, Class type); + + /** + * Subclasses implement this to configure the kryo instance. This is invoked on each new Kryo instance + * when it is created. + * @param kryo the Kryo instance + */ + protected abstract void configureKryoInstance(Kryo kryo); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoRegistrar.java new file mode 100644 index 0000000000..1557a27642 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoRegistrar.java @@ -0,0 +1,68 @@ +/* + * 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.integration.codec.kryo; + +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Registration; + +/** + * Base class for {@link KryoRegistrar} implementations. + * + * @author David Turanski + * @since 4.2 + */ +public abstract class AbstractKryoRegistrar implements KryoRegistrar { + + protected final static Kryo kryo = new Kryo(); + + protected final Log log = LogFactory.getLog(this.getClass()); + + @Override + public void registerTypes(Kryo kryo) { + for (Registration registration : getRegistrations()) { + register(kryo, registration); + } + } + + /** + * Subclasses implement this to get provided registrations. + * @return a list of {@link Registration} + */ + public abstract List getRegistrations(); + + private void register(Kryo kryo, Registration registration) { + int id = registration.getId(); + + Registration existing = kryo.getRegistration(id); + + if (existing != null) { + throw new RuntimeException((String.format("registration already exists %s", existing))); + } + + if (log.isInfoEnabled()) { + log.info(String.format("registering %s with serializer %s", registration, + registration.getSerializer().getClass().getName())); + } + + kryo.register(registration); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/CompositeKryoRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/CompositeKryoRegistrar.java new file mode 100644 index 0000000000..e50f938174 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/CompositeKryoRegistrar.java @@ -0,0 +1,79 @@ +/* + * 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.integration.codec.kryo; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +import com.esotericsoftware.kryo.Registration; + +/** + * A {@link KryoRegistrar} that delegates and validates registrations across all components. + * + * @author David Turanski + * @since 4.2 + */ +public class CompositeKryoRegistrar extends AbstractKryoRegistrar { + + private final List delegates; + + public CompositeKryoRegistrar(List delegates) { + this.delegates = new ArrayList(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 RuntimeException(String.format("Duplicate registration ID found: %d", + registration.getId())); + } + ids.add(registration.getId()); + + if (types.contains(registration.getType())) { + throw new RuntimeException(String.format("Duplicate registration found for type: %s", + registration.getType())); + } + types.add(registration.getType()); + + if (log.isInfoEnabled()) { + log.info(String.format("configured Kryo registration %s with serializer %s", registration, + registration.getSerializer().getClass().getName())); + } + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileKryoRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileKryoRegistrar.java new file mode 100644 index 0000000000..6813d55620 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileKryoRegistrar.java @@ -0,0 +1,55 @@ +/* + * 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.integration.codec.kryo; + +import java.io.File; +import java.util.Collections; +import java.util.List; + +import com.esotericsoftware.kryo.Registration; + +/** + * A {@link KryoRegistrar} used to validateRegistration a File serializer. + * + * @author David Turanski + * @since 4.2 + */ +public class FileKryoRegistrar extends AbstractKryoRegistrar { + + private final static int DEFAULT_REGISTRATION_ID = 40; + + private final int registrationId; + + private final FileSerializer fileSerializer = new FileSerializer(); + + public FileKryoRegistrar() { + this.registrationId = DEFAULT_REGISTRATION_ID; + } + + /** + * + * @param registrationId overrides the default registration ID. + */ + public FileKryoRegistrar(int registrationId) { + this.registrationId = registrationId; + } + + @Override + public List getRegistrations() { + return Collections.singletonList(new Registration(File.class, this.fileSerializer, this.registrationId)); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileSerializer.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileSerializer.java new file mode 100644 index 0000000000..0b2cfa9075 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileSerializer.java @@ -0,0 +1,47 @@ +/* + * 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.integration.codec.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; + +/** + * A custom Kryo {@link Serializer} for serializing File payloads. + * It serializes the file path and creates a new File instance to preserve the original path. + * File does not preserve the absolute otherwise as prefixLength + * is declared transient. + * + * @author David Turanski + * @since 4.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-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassListRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassListRegistrar.java new file mode 100644 index 0000000000..b0d127ce5a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/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.integration.codec.kryo; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +import com.esotericsoftware.kryo.Registration; + +/** + * A {@link KryoRegistrar} used to validateRegistration 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 4.2 + */ +public class KryoClassListRegistrar extends AbstractKryoRegistrar { + + private final List> registeredClasses; + + private int initialValue = 50; + + /** + * @param classes the list of classes to validateRegistration + */ + public KryoClassListRegistrar(List> classes) { + this.registeredClasses = new ArrayList>(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(this.registeredClasses)) { + for (int i = 0; i < this.registeredClasses.size(); i++) { + registrations.add(new Registration(this.registeredClasses.get(i), + kryo.getSerializer(this.registeredClasses.get(i)), i + this.initialValue)); + } + } + return registrations; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassMapRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassMapRegistrar.java new file mode 100644 index 0000000000..0bb0017ae7 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassMapRegistrar.java @@ -0,0 +1,54 @@ +/* + * 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.integration.codec.kryo; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.springframework.util.CollectionUtils; + +import com.esotericsoftware.kryo.Registration; + +/** + * A {@link KryoRegistrar} implementation backed by a Map + * used to explicitly set the registration ID for each class. + * + * @author David Turanski + * @since 4.2 + */ +public class KryoClassMapRegistrar extends AbstractKryoRegistrar { + + private final Map> registeredClasses; + + public KryoClassMapRegistrar(Map> kryoRegisteredClasses) { + this.registeredClasses = new HashMap>(kryoRegisteredClasses); + } + + @Override + public List getRegistrations() { + List registrations = new ArrayList(); + if (!CollectionUtils.isEmpty(this.registeredClasses)) { + for (Map.Entry> entry : this.registeredClasses.entrySet()) { + registrations.add( + new Registration(entry.getValue(), kryo.getSerializer(entry.getValue()), entry.getKey())); + } + } + return registrations; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoRegistrar.java new file mode 100644 index 0000000000..f30ce5aa70 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoRegistrar.java @@ -0,0 +1,52 @@ +/* + * 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.integration.codec.kryo; + +import java.util.List; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Registration; + +/** + * Strategy interface used by {@link PojoCodec} to configure registrations + * classes consistently across {@link Kryo} instances. + * 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 a + * consistent registration configuration in order for serialization to function properly. + * Registrations can result in better performance in demanding situations, + * but requires some care to maintain. Use this feature only if you really need it. + * + * @author David Turanski + * @since 4.2 + */ +public interface KryoRegistrar { + + int MIN_REGISTRATION_VALUE = 10; + + /** + * This method is invoked by the {@link PojoCodec} and + * applied to the {@link Kryo} instance whenever a new instance is created. + * @param kryo the Kryo instance + */ + void registerTypes(Kryo kryo); + + /** + * + * @return the list of {@link Registration} provided + */ + List getRegistrations(); +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoRegistrationRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoRegistrationRegistrar.java new file mode 100644 index 0000000000..3a771f306a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoRegistrationRegistrar.java @@ -0,0 +1,44 @@ +/* + * 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.integration.codec.kryo; + +import java.util.ArrayList; +import java.util.List; + +import com.esotericsoftware.kryo.Registration; + +/** + * A {@link KryoRegistrar} implementation backed by a List of {@link Registration}. + * + * @author David Turanski + * @since 4.2 + */ +public class KryoRegistrationRegistrar extends AbstractKryoRegistrar { + + private final List registrations; + + public KryoRegistrationRegistrar(List registrations) { + this.registrations = registrations != null + ? new ArrayList(registrations) + : new ArrayList(); + } + + @Override + public List getRegistrations() { + return this.registrations; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/PojoCodec.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/PojoCodec.java new file mode 100644 index 0000000000..ed5f4f766b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/PojoCodec.java @@ -0,0 +1,105 @@ +/* + * 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.integration.codec.kryo; + +import java.util.Collections; +import java.util.List; + +import org.springframework.util.CollectionUtils; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; + +/** + * Kryo Codec that can encode and decode arbitrary types. Classes and associated + * {@link com.esotericsoftware.kryo.Serializer}s may be registered via + * {@link KryoRegistrar}s. + * + * @author David Turanski + * @since 4.2 + */ +public class PojoCodec extends AbstractKryoCodec { + + private final CompositeKryoRegistrar kryoRegistrar; + + private final boolean useReferences; + + public PojoCodec() { + this.kryoRegistrar = null; + this.useReferences = true; + } + + /** + * Create an instance with a single KryoRegistrar. + * @param kryoRegistrar the registrar. + */ + public PojoCodec(KryoRegistrar kryoRegistrar) { + this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null, true); + } + + /** + * Create an instance with zero to many KryoRegistrars. + * @param kryoRegistrars a list KryoRegistrars. + */ + public PojoCodec(List kryoRegistrars) { + this.kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null : + new CompositeKryoRegistrar(kryoRegistrars); + this.useReferences = true; + } + + /** + * Create an instance with a single KryoRegistrar. + * @param kryoRegistrar the registrar. + * @param useReferences set to false if references are not required (if the object graph is known to be acyclical). + * The default is 'true' which is less performant but more flexible. + */ + public PojoCodec(KryoRegistrar kryoRegistrar, boolean useReferences) { + this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null, useReferences); + } + + /** + * Create an instance with zero to many KryoRegistrars. + * @param kryoRegistrars a list KryoRegistrars. + * @param useReferences set to false if references are not required (if the object graph is known to be acyclical). + * The default is 'true' which is less performant but more flexible. + */ + public PojoCodec(List kryoRegistrars, boolean useReferences) { + this.kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null : + new CompositeKryoRegistrar(kryoRegistrars); + this.useReferences = useReferences; + } + + @Override + protected void doEncode(Kryo kryo, Object object, Output output) { + kryo.writeObject(output, object); + } + + @Override + protected T doDecode(Kryo kryo, Input input, Class type) { + return kryo.readObject(input, type); + } + + @Override + protected void configureKryoInstance(Kryo kryo) { + if (this.kryoRegistrar != null) { + this.kryoRegistrar.registerTypes(kryo); + } + kryo.setReferences(this.useReferences); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/package-info.java new file mode 100644 index 0000000000..a95392bea7 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/package-info.java @@ -0,0 +1,4 @@ +/** + * The Kryo specific {@code Codec} classes. + */ +package org.springframework.integration.codec.kryo; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/package-info.java new file mode 100644 index 0000000000..1da5eb9555 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides base classes for the {@code Codec} abstraction. + */ +package org.springframework.integration.codec; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/codec/CompositeCodecTests.java b/spring-integration-core/src/test/java/org/springframework/integration/codec/CompositeCodecTests.java new file mode 100644 index 0000000000..75035eac25 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/codec/CompositeCodecTests.java @@ -0,0 +1,83 @@ +/* + * 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.integration.codec; + +import static org.junit.Assert.assertEquals; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; + +import org.springframework.integration.codec.kryo.PojoCodec; + +/** + * @author David Turanski + * @since 4.2 + */ +public class CompositeCodecTests { + + private Codec codec; + + + @Before + public void setup() { + Map, Codec> codecs = new HashMap, Codec>(); + this.codec = new CompositeCodec(codecs, new PojoCodec()); + } + + @Test + public void testPojoSerialization() throws IOException { + SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("hello", 123); + SomeClassWithNoDefaultConstructors foo2 = this.codec.decode( + this.codec.encode(foo), + 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) && this.val2 == that.val2); + } + + @Override + public int hashCode() { + int result = this.val1.hashCode(); + result = 31 * result + this.val2; + return result; + } + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/FileKryoRegistrarTests.java b/spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/FileKryoRegistrarTests.java new file mode 100644 index 0000000000..82e091b64e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/FileKryoRegistrarTests.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.integration.codec.kryo; + +import static org.junit.Assert.assertEquals; + +import java.io.File; +import java.io.IOException; + +import org.junit.Test; + + +/** + * @author David Turanski + * @since 4.2 + */ +public class FileKryoRegistrarTests { + + @Test + public void test() throws IOException { + PojoCodec pc = new PojoCodec(new FileKryoRegistrar()); + File file = new File("/foo/bar"); + File file2 = pc.decode(pc.encode(file), File.class); + assertEquals(file, file2); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/KryoCodecTests.java b/spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/KryoCodecTests.java new file mode 100644 index 0000000000..586192e400 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/KryoCodecTests.java @@ -0,0 +1,174 @@ +/* + * 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.integration.codec.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 4.2 + */ +public class KryoCodecTests { + + @Test + public void testStringSerialization() throws IOException { + String str = "hello"; + PojoCodec codec = new PojoCodec(); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + + codec.encode(str, bos); + + String s2 = codec.decode(bos.toByteArray(), String.class); + assertEquals(str, s2); + } + + @Test + public void testSerializationWithStreams() throws IOException { + String str = "hello"; + File file = new File("test.ser"); + PojoCodec codec = new PojoCodec(); + FileOutputStream fos = new FileOutputStream(file); + codec.encode(str, fos); + fos.close(); + + FileInputStream fis = new FileInputStream(file); + String s2 = codec.decode(fis, String.class); + file.delete(); + assertEquals(str, s2); + } + + @Test + public void testPojoSerialization() throws IOException { + PojoCodec codec = new PojoCodec(); + SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("foo", 123); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + codec.encode(foo, bos); + Object foo2 = codec.decode(bos.toByteArray(), SomeClassWithNoDefaultConstructors.class); + assertEquals(foo, foo2); + } + + @Test + public void testPrimitiveSerialization() throws IOException { + PojoCodec codec = new PojoCodec(); + + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + codec.encode(true, bos); + boolean b = codec.decode(bos.toByteArray(), Boolean.class); + assertEquals(true, b); + b = codec.decode(bos.toByteArray(), boolean.class); + assertEquals(true, b); + + bos = new ByteArrayOutputStream(); + codec.encode(3.14159, bos); + + double d = codec.decode(bos.toByteArray(), double.class); + assertEquals(3.14159, d, 0.00001); + + bos = new ByteArrayOutputStream(); + codec.encode(3.14159, bos); + + d = codec.decode(bos.toByteArray(), Double.class); + assertEquals(3.14159, d, 0.00001); + + } + + @Test + public void testMapSerialization() throws IOException { + PojoCodec codec = new PojoCodec(); + Map map = new HashMap(); + map.put("one", 1); + map.put("two", 2); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + codec.encode(map, bos); + Map m2 = (Map) codec.decode(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 codec = new PojoCodec(); + Foo foo = new Foo(); + foo.put("one", 1); + foo.put("two", 2); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + codec.encode(foo, bos); + + Foo foo2 = codec.decode(bos.toByteArray(), Foo.class); + assertEquals(1, foo2.get("one")); + assertEquals(2, foo2.get("two")); + } + + 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) && this.val2 == that.val2); + } + + @Override + public int hashCode() { + int result = this.val1.hashCode(); + result = 31 * result + this.val2; + return result; + } + + } + + static class Foo { + + private Map map; + + public Foo() { + map = new HashMap(); + } + + public void put(Object key, Object value) { + this.map.put(key, value); + } + + public Object get(Object key) { + return this.map.get(key); + } + + } + +}