diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java new file mode 100644 index 000000000..ea8c7e768 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java @@ -0,0 +1,204 @@ +/* + * 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.stream.converter; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.AbstractMessageConverter; +import org.springframework.messaging.converter.ContentTypeResolver; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.MimeType; + + +/** + * Base class for converters applied via Spring Integration 4.x data type channels. + * + * Extend this class to implement {@link org.springframework.messaging.converter.MessageConverter MessageConverters} + * used with custom Message conversion. Only {@link #fromMessage} is supported. + * + * @author David Turanski + */ +public abstract class AbstractFromMessageConverter extends AbstractMessageConverter { + + protected Logger logger = LoggerFactory.getLogger(this.getClass()); + + protected final List targetMimeTypes; + + /** + * Creates a converter that ignores content-type message headers + * + * @param targetMimeType the required target type + */ + protected AbstractFromMessageConverter(MimeType targetMimeType) { + this(new ArrayList(), targetMimeType, new StrictContentTypeResolver(targetMimeType)); + } + + protected AbstractFromMessageConverter(Collection targetMimeTypes) { + this(new ArrayList(), targetMimeTypes, new StringConvertingContentTypeResolver()); + } + + /** + * Creates a converter that handles one or more content-type message headers + * + * @param supportedSourceMimeTypes list of {@link MimeType} that may present in content-type header + * @param targetMimeType the required target type + */ + protected AbstractFromMessageConverter(Collection supportedSourceMimeTypes, MimeType targetMimeType, + ContentTypeResolver contentTypeResolver) { + super(supportedSourceMimeTypes); + Assert.notNull(targetMimeType, "'targetMimeType' cannot be null"); + setContentTypeResolver(contentTypeResolver); + this.targetMimeTypes = Collections.singletonList(targetMimeType); + } + + /** + * Creates a converter that handles one or more content-type message headers and one or more target MIME types + * + * @param supportedSourceMimeTypes a list of supported content types + * @param targetMimeTypes a list of supported target types + * @param contentTypeResolver the {@link ContentTypeResolver} to use + */ + protected AbstractFromMessageConverter(Collection supportedSourceMimeTypes, + Collection targetMimeTypes, + ContentTypeResolver contentTypeResolver) { + super(supportedSourceMimeTypes); + Assert.notNull(targetMimeTypes, "'targetMimeTypes' cannot be null"); + setContentTypeResolver(contentTypeResolver); + this.targetMimeTypes = new ArrayList(targetMimeTypes); + } + + /** + * Creates a converter that requires a specific content-type message header + * + * @param supportedSourceMimeType {@link MimeType} that must be present in content-type header + * @param targetMimeType the required target type + */ + protected AbstractFromMessageConverter(MimeType supportedSourceMimeType, MimeType targetMimeType) { + this(Collections.singletonList(supportedSourceMimeType), targetMimeType, new StrictContentTypeResolver( + supportedSourceMimeType)); + } + + /** + * Creates a converter that requires a specific content-type message header and supports multiple target MIME types. + * + * @param supportedSourceMimeType {@link MimeType} that must be present in content-type header + * @param targetMimeTypes a list of supported target types + */ + protected AbstractFromMessageConverter(MimeType supportedSourceMimeType, Collection targetMimeTypes) { + this(Collections.singletonList(supportedSourceMimeType), targetMimeTypes, new StrictContentTypeResolver( + supportedSourceMimeType)); + } + + /** + * Subclasses implement this to specify supported target types + * + * @return an array of supported classes or null if any target type is supported + */ + protected abstract Class[] supportedTargetTypes(); + + /** + * Subclasses implement this to specify supported payload types + * + * @return an array of supported classes or null if any target type is supported + */ + protected abstract Class[] supportedPayloadTypes(); + + protected boolean supportsPayloadType(Class clazz) { + return supportsType(clazz, supportedPayloadTypes()); + } + + @Override + protected boolean supports(Class clazz) { + return supportsType(clazz, supportedTargetTypes()); + } + + private boolean supportsType(Class clazz, Class[] supportedTypes) { + if (supportedTypes != null) { + for (Class targetType : supportedTypes) { + if (ClassUtils.isAssignable(clazz, targetType)) { + return true; + } + } + return false; + } + return true; + } + + @Override + protected boolean canConvertFrom(Message message, Class targetClass) { + return super.canConvertFrom(message, targetClass) && supportsPayloadType(message.getPayload().getClass()); + } + + public boolean supportsTargetMimeType(MimeType mimeType) { + for (MimeType targetMimeType : targetMimeTypes) { + if (mimeType.getType().equals(targetMimeType.getType()) && mimeType.getSubtype().equals( + targetMimeType.getSubtype())) { + return true; + } + } + return false; + } + + @Override + // TODO: This will likely be fixed in core Spring + public void setContentTypeResolver(ContentTypeResolver resolver) { + if (getContentTypeResolver() == null) { + super.setContentTypeResolver(resolver); + } + } + + /** + * Not supported by default + */ + @Override + protected boolean canConvertTo(Object payload, MessageHeaders headers) { + return false; + } + + /** + * Not supported by default + */ + @Override + public Object convertToInternal(Object payload, MessageHeaders headers) { + throw new UnsupportedOperationException("'convertTo' not supported"); + } + + /** + * Convenience method to construct a converted message + * + * @param payload the converted payload + * @param headers the existing message headers + * @param contentType the value of the content-type header + * @return the converted message + */ + protected final Message buildConvertedMessage(Object payload, MessageHeaders headers, MimeType contentType) { + return MessageBuilder.withPayload(payload).copyHeaders(headers) + .copyHeaders( + Collections.singletonMap(MessageHeaders.CONTENT_TYPE, + contentType)).build(); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java new file mode 100644 index 000000000..e1346e2a1 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java @@ -0,0 +1,89 @@ +/* + * 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.stream.converter; + +import java.io.UnsupportedEncodingException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.ContentTypeResolver; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; + + +/** + * A {@link org.springframework.messaging.converter.MessageConverter} + * to convert from byte[] to String applying the Charset provided in + * the content-type header if any. + * + * @author David Turanski + */ +public class ByteArrayToStringMessageConverter extends AbstractFromMessageConverter { + + private final static ContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver(); + + private final static List targetMimeTypes = new ArrayList(); + static { + targetMimeTypes.add(MessageConverterUtils.X_SPRING_STRING); + targetMimeTypes.add(MessageConverterUtils.X_JAVA_OBJECT); + targetMimeTypes.add(MimeTypeUtils.TEXT_PLAIN); + } + + public ByteArrayToStringMessageConverter() { + super(Arrays.asList(new MimeType[] { MimeTypeUtils.APPLICATION_OCTET_STREAM, MimeTypeUtils.TEXT_PLAIN }), + targetMimeTypes, contentTypeResolver); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] { String.class }; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] { byte[].class }; + } + + /** + * Don't need to manipulate message headers. Just return payload + */ + @Override + public Object convertFromInternal(Message message, Class targetClass) { + MimeType mimeType = contentTypeResolver.resolve(message.getHeaders()); + + String converted = null; + + if (mimeType == null || mimeType.getParameter("Charset") == null) { + converted = new String((byte[]) message.getPayload()); + } + else { + String encoding = mimeType.getParameter("Charset"); + if (encoding != null) { + try { + converted = new String((byte[]) message.getPayload(), encoding); + } + catch (UnsupportedEncodingException e) { + logger.error(e.getMessage(), e); + } + } + } + return converted; + + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java new file mode 100644 index 000000000..64d3951ad --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.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.cloud.stream.converter; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.springframework.messaging.converter.CompositeMessageConverter; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; +import org.springframework.util.MimeType; + + +/** + * A factory for creating an instance of {@link CompositeMessageConverter} for a given target MIME type + * + * @author David Turanski + */ +public class CompositeMessageConverterFactory { + + private final List converters; + + /** + * @param converters a list of {@link AbstractFromMessageConverter} + */ + public CompositeMessageConverterFactory(Collection converters) { + Assert.notNull(converters, "'converters' cannot be null"); + this.converters = new ArrayList(converters); + } + + /** + * Creation method. + * + * @param targetMimeType the target MIME type + * @return a converter for the target MIME type + */ + public CompositeMessageConverter newInstance(MimeType targetMimeType) { + List targetMimeTypeConverters = new ArrayList(); + for (AbstractFromMessageConverter converter : converters) { + if (converter.supportsTargetMimeType(targetMimeType)) { + targetMimeTypeConverters.add(converter); + } + } + if (CollectionUtils.isEmpty(targetMimeTypeConverters)) { + throw new ConversionException("No message converter is registered for " + + targetMimeType.toString()); + } + return new CompositeMessageConverter(targetMimeTypeConverters); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ConversionException.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ConversionException.java new file mode 100644 index 000000000..20fe52a18 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ConversionException.java @@ -0,0 +1,35 @@ +/* + * 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.stream.converter; + +/** + * Exception thrown when an error is encountered during message conversion. + * + * @author David Turanski + */ +@SuppressWarnings("serial") +public class ConversionException extends RuntimeException { + + public ConversionException(String message) { + super(message); + } + + public ConversionException(String message, Throwable t) { + super(message, t); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CustomMimeTypeConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CustomMimeTypeConverter.java new file mode 100644 index 000000000..3a58cddec --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CustomMimeTypeConverter.java @@ -0,0 +1,41 @@ +/* + * 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.cloud.stream.converter; + +import org.springframework.core.convert.converter.Converter; +import org.springframework.http.MediaType; +import org.springframework.util.MimeType; + +/** + * A custom converter for {@link MediaType} that accepts a plain java class name as a shorthand for + * {@code application/x-java-object;type=the.qualified.ClassName}. + * + * + * @author Eric Bottard + * @author David Turanski + */ +public class CustomMimeTypeConverter implements Converter { + + @Override + public MimeType convert(String source) { + if (!source.contains("/")) { + return MimeType.valueOf("application/x-java-object;type=" + source); + } + return MimeType.valueOf(source); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java new file mode 100644 index 000000000..427da06d9 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java @@ -0,0 +1,63 @@ +/* + * 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.stream.converter; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectOutputStream; +import java.io.Serializable; + +import org.springframework.messaging.Message; + + +/** + * A {@link org.springframework.messaging.converter.MessageConverter} + * to convert from a POJO to byte[] with Java.io serialization if any. + * + * @author David Turanski + */ +public class JavaToSerializedMessageConverter extends AbstractFromMessageConverter { + + public JavaToSerializedMessageConverter() { + super(MessageConverterUtils.X_JAVA_OBJECT, MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] { byte[].class }; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] { Serializable.class }; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass) { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try { + new ObjectOutputStream(bos).writeObject(message.getPayload()); + } + catch (IOException e) { + logger.error(e.getMessage(), e); + return null; + } + + return buildConvertedMessage(bos.toByteArray(), message.getHeaders(), + MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java new file mode 100644 index 000000000..652611dcd --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java @@ -0,0 +1,69 @@ +/* + * 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.stream.converter; + +import org.springframework.messaging.Message; +import org.springframework.util.MimeTypeUtils; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; + + +/** + * + * @author David Turanski + */ +public class JsonToPojoMessageConverter extends AbstractFromMessageConverter { + + private final ObjectMapper mapper = new ObjectMapper(); + + public JsonToPojoMessageConverter() { + super(MimeTypeUtils.APPLICATION_JSON, MessageConverterUtils.X_JAVA_OBJECT); + mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false); + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] {String.class, byte[].class}; + } + + @Override + protected Class[] supportedTargetTypes() { + return null; // any type + } + + @Override + public Object convertFromInternal(Message message, Class targetClass) { + Object result = null; + try { + Object payload = message.getPayload(); + + if (payload instanceof byte[]) { + result = mapper.readValue((byte[]) payload, targetClass); + } + else if (payload instanceof String) { + result = mapper.readValue((String) payload, targetClass); + } + } + catch (Exception e) { + logger.error(e.getMessage(), e); + return null; + } + return buildConvertedMessage(result, message.getHeaders(), + MessageConverterUtils.javaObjectMimeType(targetClass)); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java new file mode 100644 index 000000000..a9aff1ee0 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java @@ -0,0 +1,70 @@ +/* + * 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.stream.converter; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.messaging.Message; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; +import org.springframework.cloud.stream.tuple.Tuple; +import org.springframework.cloud.stream.tuple.TupleBuilder; + + +/** + * A {@link org.springframework.messaging.converter.MessageConverter} + * to convert from a JSON (byte[] or String) to a {@link Tuple}. + * + * @author David Turanski + */ +public class JsonToTupleMessageConverter extends AbstractFromMessageConverter { + + private final static List targetMimeTypes = new ArrayList(); + + static { + targetMimeTypes.add(MessageConverterUtils.X_SPRING_TUPLE); + targetMimeTypes.add(MessageConverterUtils.X_JAVA_OBJECT); + } + + public JsonToTupleMessageConverter() { + super(MimeTypeUtils.APPLICATION_JSON, targetMimeTypes); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] { Tuple.class }; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] { byte[].class, String.class }; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass) { + String source = null; + if (message.getPayload() instanceof byte[]) { + source = new String((byte[]) message.getPayload()); + } + else { + source = (String) message.getPayload(); + } + Tuple t = TupleBuilder.fromString(source); + return buildConvertedMessage(t, message.getHeaders(), MessageConverterUtils.javaObjectMimeType(t.getClass())); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java new file mode 100644 index 000000000..f2ee9588e --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java @@ -0,0 +1,135 @@ +/* + * 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.stream.converter; + +import static org.springframework.util.MimeType.valueOf; +import static org.springframework.util.MimeTypeUtils.APPLICATION_JSON; +import static org.springframework.util.MimeTypeUtils.APPLICATION_OCTET_STREAM; + +import org.springframework.cloud.stream.tuple.Tuple; +import org.springframework.util.ClassUtils; +import org.springframework.util.MimeType; +import org.springframework.cloud.stream.tuple.DefaultTuple; +import org.springframework.util.StringUtils; + + +/** + * Message conversion utility methods. + * + * @author David Turanski + * @author Gary Russell + */ +public class MessageConverterUtils { + + /** + * An MimeType specifying a {@link Tuple}. + */ + public static final MimeType X_SPRING_TUPLE = MimeType.valueOf("application/x-spring-tuple"); + + /** + * An MimeType for specifying a String. + */ + public static final MimeType X_SPRING_STRING = MimeType.valueOf("application/x-spring-string"); + + /** + * A general MimeType for Java Types. + */ + public static final MimeType X_JAVA_OBJECT = MimeType.valueOf("application/x-java-object"); + + /** + * A general MimeType for a Java serialized byte array. + */ + public static final MimeType X_JAVA_SERIALIZED_OBJECT = MimeType.valueOf("application/x-java-serialized-object"); + + /** + * Map the contentType to a target class. + * + * @param contentType the content type + * @param classLoader the class loader used to resolve the class + * @return the class for the content type + */ + public static Class getJavaTypeForContentType(MimeType contentType, ClassLoader classLoader) { + if (X_JAVA_OBJECT.includes(contentType)) { + if (contentType.getParameter("type") != null) { + try { + return ClassUtils.forName(contentType.getParameter("type"), classLoader); + } + catch (Exception e) { + throw new ConversionException(e.getMessage(), e); + } + } + else { + return Object.class; + } + } + else if (APPLICATION_JSON.equals(contentType)) { + return String.class; + } + else if (valueOf("text/*").includes(contentType)) { + return String.class; + } + else if (X_SPRING_TUPLE.includes(contentType)) { + return DefaultTuple.class; + } + else if (APPLICATION_OCTET_STREAM.includes(contentType)) { + return byte[].class; + } + else if (X_JAVA_SERIALIZED_OBJECT.includes(contentType)) { + return byte[].class; + } + else if (X_SPRING_STRING.includes(contentType)) { + return String.class; + } + return null; + } + + /** + * Build the conventional {@link MimeType} for a java object + * + * @param clazz the java type + * @return the MIME type + */ + public static MimeType javaObjectMimeType(Class clazz) { + return MimeType.valueOf("application/x-java-object;type=" + clazz.getName()); + } + + public static MimeType getMimeType(String contentTypeString) { + MimeType mimeType = null; + if (StringUtils.hasText(contentTypeString)) { + try { + mimeType = resolveContentType(contentTypeString); + } + catch (ClassNotFoundException cfe) { + throw new IllegalArgumentException("Could not find the class required for " + contentTypeString, cfe); + } + } + return mimeType; + } + + public static MimeType resolveContentType(String type) throws ClassNotFoundException, LinkageError { + if (!type.contains("/")) { + Class javaType = resolveJavaType(type); + return MessageConverterUtils.javaObjectMimeType(javaType); + } + return MimeType.valueOf(type); + } + + public static Class resolveJavaType(String type) throws ClassNotFoundException, LinkageError { + return ClassUtils.forName(type, Thread.currentThread().getContextClassLoader()); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java new file mode 100644 index 000000000..89b118ce0 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.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.cloud.stream.converter; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.messaging.Message; +import org.springframework.util.MimeTypeUtils; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; + + +/** + * A {@link org.springframework.messaging.converter.MessageConverter} + * to convert a Java object to a JSON String + * + * @author David Turanski + * @author David Liu + */ +public class PojoToJsonMessageConverter extends AbstractFromMessageConverter { + + private final ObjectMapper mapper = new ObjectMapper(); + + @Value("${typeconversion.json.prettyPrint:false}") + private volatile boolean prettyPrint; + + public PojoToJsonMessageConverter() { + super(MimeTypeUtils.APPLICATION_JSON); + mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] {String.class}; + } + + @Override + protected Class[] supportedPayloadTypes() { + return null; + } + + + public void setPrettyPrint(boolean prettyPrint) { + this.prettyPrint = prettyPrint; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass) { + Object result; + try { + if (prettyPrint) { + result = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(message.getPayload()); + } + else { + result = mapper.writeValueAsString(message.getPayload()); + } + } + catch (JsonProcessingException e) { + logger.error(e.getMessage(), e); + return null; + } + return buildConvertedMessage(result, message.getHeaders(), MimeTypeUtils.APPLICATION_JSON); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java new file mode 100644 index 000000000..292c9000d --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java @@ -0,0 +1,49 @@ +/* + * 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.stream.converter; + +import org.springframework.messaging.Message; +import org.springframework.util.MimeTypeUtils; + + +/** + * A {@link org.springframework.messaging.converter.MessageConverter} + * to convert a Java object to a String using toString() + * + * @author David Turanski + */ +public class PojoToStringMessageConverter extends AbstractFromMessageConverter { + + public PojoToStringMessageConverter() { + super(MimeTypeUtils.TEXT_PLAIN); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] { String.class }; + } + + @Override + protected Class[] supportedPayloadTypes() { + return null; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass) { + return buildConvertedMessage(message.getPayload().toString(), message.getHeaders(), MimeTypeUtils.TEXT_PLAIN); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java new file mode 100644 index 000000000..61b94e7bf --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java @@ -0,0 +1,63 @@ +/* + * 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.stream.converter; + +import java.io.ByteArrayInputStream; +import java.io.ObjectInputStream; +import java.io.Serializable; + +import org.springframework.messaging.Message; + + +/** + * A {@link org.springframework.messaging.converter.MessageConverter} + * to deserialize {@link Serializable} Java objects. + * + * @author David Turanski + */ +public class SerializedToJavaMessageConverter extends AbstractFromMessageConverter { + + public SerializedToJavaMessageConverter() { + super(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT, MessageConverterUtils.X_JAVA_OBJECT); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] { Serializable.class }; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] { byte[].class }; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass) { + ByteArrayInputStream bis = new ByteArrayInputStream((byte[]) (message.getPayload())); + Object result = null; + try { + result = new ObjectInputStream(bis).readObject(); + } + catch (Exception e) { + logger.error(e.getMessage(), e); + return null; + } + + return buildConvertedMessage(result, message.getHeaders(), + MessageConverterUtils.javaObjectMimeType(targetClass)); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StrictContentTypeResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StrictContentTypeResolver.java new file mode 100644 index 000000000..0155008f6 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StrictContentTypeResolver.java @@ -0,0 +1,37 @@ +/* + * 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.stream.converter; + +import org.springframework.util.MimeType; + + +/** + * A {@link StringConvertingContentTypeResolver} that requires a the content-type to be present. + * + * @author David Turanski + */ +// TODO: This will likely be pushed to core Spring +public class StrictContentTypeResolver extends StringConvertingContentTypeResolver { + + /** + * @param defaultMimeType the required {@link MimeType} + */ + public StrictContentTypeResolver(MimeType defaultMimeType) { + super(); + setDefaultMimeType(defaultMimeType); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StringConvertingContentTypeResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StringConvertingContentTypeResolver.java new file mode 100644 index 000000000..949df0a45 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StringConvertingContentTypeResolver.java @@ -0,0 +1,57 @@ +/* + * 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.stream.converter; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.DefaultContentTypeResolver; +import org.springframework.util.MimeType; + +/** + * A {@link DefaultContentTypeResolver} that can parse String values. + * + * @author David Turanski + */ +public class StringConvertingContentTypeResolver extends DefaultContentTypeResolver { + + private ConcurrentMap mimeTypeCache = new ConcurrentHashMap<>(); + + @Override + public MimeType resolve(MessageHeaders headers) { + return resolve((Map) headers); + } + + public MimeType resolve(Map headers) { + Object value = headers.get(MessageHeaders.CONTENT_TYPE); + if (value instanceof MimeType) { + return (MimeType) value; + } + else if (value instanceof String) { + MimeType mimeType = mimeTypeCache.get(value); + if (mimeType == null) { + String valueAsString = (String) value; + mimeType = MimeType.valueOf(valueAsString); + mimeTypeCache.put(valueAsString,mimeType); + } + return mimeType; + } + return getDefaultMimeType(); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StringToByteArrayMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StringToByteArrayMessageConverter.java new file mode 100644 index 000000000..17440a9c9 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StringToByteArrayMessageConverter.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.cloud.stream.converter; + +import java.io.UnsupportedEncodingException; +import java.util.ArrayList; +import java.util.List; + +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.ContentTypeResolver; +import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; + + +/** + * A {@link org.springframework.messaging.converter.MessageConverter} + * to convert a String to a byte[], applying the provided Charset in + * the content-type header if any. + * + * @author David Turanski + */ +public class StringToByteArrayMessageConverter extends AbstractFromMessageConverter { + + private final static ContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver(); + + private final static List targetMimeTypes = new ArrayList(); + static { + targetMimeTypes.add(MimeTypeUtils.APPLICATION_OCTET_STREAM); + } + + public StringToByteArrayMessageConverter() { + super(targetMimeTypes); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] { byte[].class }; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] { String.class }; + } + + /** + * Don't need to manipulate message headers. Just return the payload + */ + @Override + public Object convertFromInternal(Message message, Class targetClass) { + MimeType mimeType = contentTypeResolver.resolve(message.getHeaders()); + byte[] converted = null; + if (mimeType == null || mimeType.getParameter("Charset") == null) { + converted = ((String) message.getPayload()).getBytes(); + } + else { + String encoding = mimeType.getParameter("Charset"); + if (encoding != null) { + try { + converted = ((String) message.getPayload()).getBytes(encoding); + } + catch (UnsupportedEncodingException e) { + logger.error(e.getMessage(), e); + } + } + } + return converted; + + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java new file mode 100644 index 000000000..9fce50f66 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java @@ -0,0 +1,81 @@ +/* + * 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.stream.converter; + +import java.io.IOException; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.messaging.Message; +import org.springframework.util.MimeTypeUtils; +import org.springframework.cloud.stream.tuple.Tuple; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; + + +/** + * A {@link org.springframework.messaging.converter.MessageConverter} + * to convert a {@link Tuple} to a JSON String + * + * @author David Turanski + */ +public class TupleToJsonMessageConverter extends AbstractFromMessageConverter { + + @Value("${typeconversion.json.prettyPrint:false}") + private volatile boolean prettyPrint; + + public void setPrettyPrint(boolean prettyPrint) { + this.prettyPrint = prettyPrint; + } + + public TupleToJsonMessageConverter() { + super(MimeTypeUtils.APPLICATION_JSON); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] { String.class }; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] { Tuple.class }; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass) { + Tuple t = (Tuple) message.getPayload(); + String json; + if (prettyPrint) { + ObjectMapper mapper = new ObjectMapper(); + mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false); + try { + Object tmp = mapper.readValue(t.toString(), Object.class); + json = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(tmp); + } + catch (IOException e) { + logger.error(e.getMessage(), e); + return null; + } + } + else { + json = t.toString(); + } + return buildConvertedMessage(json, message.getHeaders(), MimeTypeUtils.APPLICATION_JSON); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/package-info.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/package-info.java new file mode 100644 index 000000000..eccaf6141 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/package-info.java @@ -0,0 +1,22 @@ +/* + * 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 for message converters. + */ + +package org.springframework.cloud.stream.converter;