diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java index a6b5081dc6..1aba0d5e60 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java @@ -26,6 +26,7 @@ import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; import org.springframework.integration.EiMessageHeaderAccessor; import org.springframework.integration.amqp.AmqpHeaders; +import org.springframework.integration.json.JsonHeaders; import org.springframework.integration.mapping.AbstractHeaderMapper; import org.springframework.util.StringUtils; @@ -45,6 +46,8 @@ import org.springframework.util.StringUtils; * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper implements AmqpHeaderMapper { @@ -70,6 +73,9 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper headers = amqpMessageProperties.getHeaders(); headers.remove(AmqpHeaders.STACKED_CORRELATION_HEADER); headers.remove(AmqpHeaders.STACKED_REPLY_TO_HEADER); + return headers; } @@ -272,6 +287,18 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper) { + value = ((Class) value).getName(); + } + amqpMessageProperties.setHeader(jsonHeader.replaceFirst(JsonHeaders.PREFIX, ""), value.toString()); + } + } + String replyCorrelation = getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_CORRELATION, String.class); if (StringUtils.hasLength(replyCorrelation)) { amqpMessageProperties.setHeader("spring_reply_correlation", replyCorrelation); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java new file mode 100644 index 0000000000..cfa0eb620a --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -0,0 +1,168 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.inbound; + +import static org.junit.Assert.assertEquals; +import static org.mockito.Matchers.anyBoolean; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.junit.Test; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.connection.Connection; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.support.converter.JsonMessageConverter; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.integration.Message; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.json.JsonToObjectTransformer; +import org.springframework.integration.json.ObjectToJsonTransformer; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.transformer.Transformer; + +import com.rabbitmq.client.Channel; + +/** + * @author Artem Bilan + * @since 3.0 + */ +public class InboundEndpointTests { + + @Test + public void testInt2809JavaTypePropertiesToAmqp() { + Connection connection = mock(Connection.class); + doAnswer(new Answer() { + public Channel answer(InvocationOnMock invocation) throws Throwable { + return mock(Channel.class); + } + }).when(connection).createChannel(anyBoolean()); + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + when(connectionFactory.createConnection()).thenReturn(connection); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + + AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container); + adapter.setMessageConverter(new JsonMessageConverter()); + + PollableChannel channel = new QueueChannel(); + + adapter.setOutputChannel(channel); + adapter.afterPropertiesSet(); + + Object payload = new Foo("bar1"); + + Transformer objectToJsonTransformer = new ObjectToJsonTransformer(); + Message jsonMessage = objectToJsonTransformer.transform(new GenericMessage(payload)); + + MessageProperties amqpMessageProperties = new MessageProperties(); + org.springframework.amqp.core.Message amqpMessage = + new SimpleMessageConverter().toMessage(jsonMessage.getPayload(), amqpMessageProperties); + new DefaultAmqpHeaderMapper().fromHeadersToRequest(jsonMessage.getHeaders(), amqpMessageProperties); + + MessageListener listener = (MessageListener) container.getMessageListener(); + listener.onMessage(amqpMessage); + + Message result = channel.receive(1000); + assertEquals(payload, result.getPayload()); + } + + @Test + public void testInt2809JavaTypePropertiesFromAmqp() { + Connection connection = mock(Connection.class); + doAnswer(new Answer() { + public Channel answer(InvocationOnMock invocation) throws Throwable { + return mock(Channel.class); + } + }).when(connection).createChannel(anyBoolean()); + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + when(connectionFactory.createConnection()).thenReturn(connection); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + + AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container); + + PollableChannel channel = new QueueChannel(); + + adapter.setOutputChannel(channel); + adapter.afterPropertiesSet(); + + Object payload = new Foo("bar1"); + + MessageProperties amqpMessageProperties = new MessageProperties(); + org.springframework.amqp.core.Message amqpMessage = new JsonMessageConverter().toMessage(payload, amqpMessageProperties); + + MessageListener listener = (MessageListener) container.getMessageListener(); + listener.onMessage(amqpMessage); + + Message receive = channel.receive(1000); + + Message result = new JsonToObjectTransformer().transform(receive); + + assertEquals(payload, result.getPayload()); + } + + + + public static class Foo { + + private String bar; + + public Foo() { + } + + public Foo(String bar) { + this.bar = bar; + } + + public String getBar() { + return bar; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + + Foo foo = (Foo) o; + + if (bar != null ? !bar.equals(foo.bar) : foo.bar != null) { + return false; + } + + return true; + } + + @Override + public int hashCode() { + return bar != null ? bar.hashCode() : 0; + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/ChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/ChannelRegistry.java deleted file mode 100644 index 6cff673b4e..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/ChannelRegistry.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on - * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ - -package org.springframework.integration.channel.registry; - -import org.springframework.messaging.MessageChannel; - -/** - * A strategy interface used to bind a {@link MessageChannel} to a logical name. The name - * is intended to identify a logical consumer or producer of messages. This may be a - * queue, a channel adapter, another message channel, a Spring bean, etc. - * - * @author Mark Fisher - * @author David Turanski - * @since 3.0 - */ -public interface ChannelRegistry { - - /** - * Register a message consumer - * @param name the logical identity of the message source - * @param channel the channel bound as a consumer - */ - void inbound(String name, MessageChannel channel); - - /** - * Register a message producer - * @param name the logical identity of the message target - * @param channel the channel bound as a producer - */ - void outbound(String name, MessageChannel channel); - - /** - * Create a tap on an already registered inbound channel - * @param name the registered name - * @param channel the channel that will receive messages from the tap - */ - void tap(String name, MessageChannel channel); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/LocalChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/LocalChannelRegistry.java deleted file mode 100644 index 49f70a3a82..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/LocalChannelRegistry.java +++ /dev/null @@ -1,168 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on - * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ - -package org.springframework.integration.channel.registry; - -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.support.AbstractApplicationContext; -import org.springframework.messaging.MessageChannel; -import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.channel.interceptor.WireTap; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.integration.handler.BridgeHandler; -import org.springframework.util.Assert; - -/** - * A simple implementation of {@link ChannelRegistry} for in-process use. For inbound and - * outbound, creates a {@link DirectChannel} and bridges the passed - * {@link MessageChannel} to the channel which is registered in the given application - * context. If that channel does not yet exist, it will be created. For tap, it adds a - * {@link WireTap} for an inbound channel whose name matches the one provided. If no such - * inbound channel exists at the time of the method invocation, it will throw an - * Exception. Otherwise the provided channel instance will receive messages from the wire - * tap on that inbound channel. - * - * @author David Turanski - * @author Mark Fisher - * @since 3.0 - */ -public class LocalChannelRegistry implements ChannelRegistry, ApplicationContextAware, InitializingBean { - - private volatile AbstractApplicationContext applicationContext; - - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - Assert.isInstanceOf(AbstractApplicationContext.class, applicationContext); - this.applicationContext = (AbstractApplicationContext) applicationContext; - } - - @Override - public void afterPropertiesSet() throws Exception { - Assert.notNull(applicationContext, "The 'applicationContext' property cannot be null"); - } - - /** - * Looks up or creates a DirectChannel with the given name and creates a bridge from - * that channel to the provided channel instance. Also registers a wire tap if the - * channel for the given name had been created. The target of the wire tap is a - * publish-subscribe channel. - */ - @Override - public void inbound(String name, MessageChannel channel) { - Assert.hasText(name, "a valid name is required to register an inbound channel"); - Assert.notNull(channel, "channel must not be null"); - DirectChannel registeredChannel = lookupOrCreateSharedChannel(name, DirectChannel.class); - bridge(registeredChannel, channel); - createSharedTapChannelIfNecessary(registeredChannel); - } - - /** - * Looks up or creates a DirectChannel with the given name and creates a bridge to - * that channel from the provided channel instance. - */ - @Override - public void outbound(String name, MessageChannel channel) { - Assert.hasText(name, "a valid name is required to register an outbound channel"); - Assert.notNull(channel, "channel must not be null"); - Assert.isTrue(channel instanceof SubscribableChannel, - "channel must be of type " + SubscribableChannel.class.getName()); - DirectChannel registeredChannel = lookupOrCreateSharedChannel(name, DirectChannel.class); - bridge((SubscribableChannel) channel, registeredChannel); - } - - /** - * Looks up a wiretap for the inbound channel with the given name and creates a - * bridge from that wiretap's output channel to the provided channel instance. - * Will throw an Exception if no such wiretap exists. - */ - @Override - public void tap(String name, MessageChannel channel) { - Assert.hasText(name, "a valid name is required to register a tap channel"); - Assert.notNull(channel, "channel must not be null"); - SubscribableChannel tapChannel = null; - String tapName = name + ".tap"; - try { - tapChannel = applicationContext.getBean(tapName, SubscribableChannel.class); - } - catch (Exception e) { - throw new IllegalArgumentException("No tap channel exists for '" + name - + "'. A tap is only valid for a registered inbound channel."); - } - bridge(tapChannel, channel); - } - - protected synchronized T lookupOrCreateSharedChannel(String name, Class requiredType) { - T channel = null; - if (applicationContext.containsBean(name)) { - try { - channel = applicationContext.getBean(name, requiredType); - } - catch (Exception e) { - throw new IllegalArgumentException("bean '" + name - + "' is already registered but does not match the required type"); - } - } - else { - channel = createSharedChannel(name, requiredType); - } - return channel; - } - - protected T createSharedChannel(String name, Class requiredType) { - try { - T channel = requiredType.newInstance(); - channel.setComponentName(name); - channel.setBeanFactory(applicationContext); - channel.setBeanName(name); - channel.afterPropertiesSet(); - applicationContext.getBeanFactory().registerSingleton(name, channel); - return channel; - } - catch (Exception e) { - throw new IllegalArgumentException("failed to create channel: " + name, e); - } - } - - private synchronized void createSharedTapChannelIfNecessary(AbstractMessageChannel channel) { - String tapName = channel.getComponentName() + ".tap"; - PublishSubscribeChannel tapChannel = null; - if (!applicationContext.containsBean(tapName)) { - tapChannel = createSharedChannel(tapName, PublishSubscribeChannel.class); - WireTap wireTap = new WireTap(tapChannel); - channel.addInterceptor(wireTap); - } - else { - try { - tapChannel = applicationContext.getBean(tapName, PublishSubscribeChannel.class); - } - catch (Exception e) { - throw new IllegalArgumentException("bean '" + tapName - + "' is already registered but does not match the required type"); - } - } - } - - protected BridgeHandler bridge(SubscribableChannel from, MessageChannel to) { - BridgeHandler handler = new BridgeHandler(); - handler.setOutputChannel(to); - handler.afterPropertiesSet(); - from.subscribe(handler); - return handler; - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/package-info.java deleted file mode 100644 index d9b84f22f3..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -/** - * Provides classes representing channel registries. - */ -package org.springframework.integration.channel.registry; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java index d5d6b9a574..229da3e767 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java @@ -73,7 +73,7 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF } /* * Return a reply-producing message handler so that we still get 'produced no reply' messages - * and the super class will inject the advice chain to advise the handler if needed. + * and the super class will inject the advice chain to advise the handler method if needed. */ handler = new AbstractReplyProducingMessageHandler() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/JsonToObjectTransformerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/JsonToObjectTransformerParser.java index 588432554d..bb8c856836 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/JsonToObjectTransformerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/JsonToObjectTransformerParser.java @@ -16,12 +16,12 @@ package org.springframework.integration.config.xml; -import org.springframework.util.StringUtils; import org.w3c.dom.Element; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.json.JsonToObjectTransformer; +import org.springframework.util.StringUtils; /** * @author Mark Fisher @@ -39,7 +39,9 @@ public class JsonToObjectTransformerParser extends AbstractTransformerParser { protected void parseTransformer(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { String type = element.getAttribute("type"); String objectMapper = element.getAttribute("object-mapper"); - builder.addConstructorArgValue(type); + if (StringUtils.hasText(type)) { + builder.addConstructorArgValue(type); + } if (StringUtils.hasText(objectMapper)) { builder.addConstructorArgReference(objectMapper); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonHeaders.java b/spring-integration-core/src/main/java/org/springframework/integration/json/JsonHeaders.java new file mode 100644 index 0000000000..5381bdf472 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/json/JsonHeaders.java @@ -0,0 +1,43 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.json; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; + +/** + * Pre-defined names and prefixes to be used for setting and/or retrieving JSON + * entries from/to Message Headers and other adapter, e.g. AMQP. + * + * @author Artem Bilan + * @since 3.0 + */ +public class JsonHeaders { + + public static final String PREFIX = "json"; + + public static final String TYPE_ID = PREFIX + "__TypeId__"; + + public static final String CONTENT_TYPE_ID = PREFIX + "__ContentTypeId__"; + + public static final String KEY_TYPE_ID = PREFIX + "__KeyTypeId__"; + + public static final Collection HEADERS = + Collections.unmodifiableList(Arrays.asList(TYPE_ID, CONTENT_TYPE_ID, KEY_TYPE_ID)); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java index 6bbf10ec2f..92d895ef5a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/json/JsonToObjectTransformer.java @@ -16,10 +16,13 @@ package org.springframework.integration.json; +import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.integration.Message; +import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.json.JacksonJsonObjectMapper; import org.springframework.integration.support.json.JacksonJsonObjectMapperProvider; import org.springframework.integration.support.json.JsonObjectMapper; -import org.springframework.integration.transformer.AbstractPayloadTransformer; +import org.springframework.integration.transformer.AbstractTransformer; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -36,13 +39,17 @@ import org.springframework.util.ClassUtils; * @see JacksonJsonObjectMapperProvider * @since 2.0 */ -public class JsonToObjectTransformer extends AbstractPayloadTransformer { +public class JsonToObjectTransformer extends AbstractTransformer implements BeanClassLoaderAware { - private final Class targetClass; + private final Class targetClass; private final JsonObjectMapper jsonObjectMapper; - public JsonToObjectTransformer(Class targetClass) { + public JsonToObjectTransformer() { + this((Class) null); + } + + public JsonToObjectTransformer(Class targetClass) { this(targetClass, null); } @@ -52,8 +59,7 @@ public class JsonToObjectTransformer extends AbstractPayloadTransformer targetClass, Object objectMapper) throws ClassNotFoundException { - Assert.notNull(targetClass, "targetClass must not be null"); + public JsonToObjectTransformer(Class targetClass, Object objectMapper) throws ClassNotFoundException { this.targetClass = targetClass; if (objectMapper != null) { try { @@ -70,15 +76,34 @@ public class JsonToObjectTransformer extends AbstractPayloadTransformer targetClass, JsonObjectMapper jsonObjectMapper) { - Assert.notNull(targetClass, "targetClass must not be null"); + public JsonToObjectTransformer(JsonObjectMapper jsonObjectMapper) { + this(null, jsonObjectMapper); + } + + public JsonToObjectTransformer(Class targetClass, JsonObjectMapper jsonObjectMapper) { this.targetClass = targetClass; this.jsonObjectMapper = (jsonObjectMapper != null) ? jsonObjectMapper : JacksonJsonObjectMapperProvider.newInstance(); } @Override - protected T transformPayload(String payload) throws Exception { - return this.jsonObjectMapper.fromJson(payload, this.targetClass); + public void setBeanClassLoader(ClassLoader classLoader) { + if (this.jsonObjectMapper instanceof BeanClassLoaderAware) { + ((BeanClassLoaderAware) this.jsonObjectMapper).setBeanClassLoader(classLoader); + } + } + + @Override + protected Object doTransform(Message message) throws Exception { + if (this.targetClass != null) { + return this.jsonObjectMapper.fromJson(message.getPayload(), this.targetClass); + } + else { + Object result = this.jsonObjectMapper.fromJson(message.getPayload(), message.getHeaders()); + MessageBuilder messageBuilder = MessageBuilder.withPayload(result) + .copyHeaders(message.getHeaders()) + .removeHeaders(JsonHeaders.HEADERS.toArray(new String[3])); + return messageBuilder.build(); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/ObjectToJsonTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/json/ObjectToJsonTransformer.java index b9e0c37185..633ed3d462 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/json/ObjectToJsonTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/json/ObjectToJsonTransformer.java @@ -109,6 +109,9 @@ public class ObjectToJsonTransformer extends AbstractTransformer { else if (StringUtils.hasLength(this.contentType)) { headers.put(MessageHeaders.CONTENT_TYPE, this.contentType); } + + this.jsonObjectMapper.populateJavaTypes(headers, message.getPayload().getClass()); + messageBuilder.copyHeaders(headers); return messageBuilder.build(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java index c91a3b9d33..82d8ddc70b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java @@ -26,6 +26,8 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.json.JsonHeaders; import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -248,7 +250,7 @@ public abstract class AbstractHeaderMapper implements RequestReplyHeaderMappe if (!type.isAssignableFrom(value.getClass())) { if (logger.isWarnEnabled()) { logger.warn("skipping header '" + name + "' since it is not of expected type [" + type + "], it is [" + - value.getClass() + "]"); + value.getClass() + "]"); } return null; } @@ -271,9 +273,14 @@ public abstract class AbstractHeaderMapper implements RequestReplyHeaderMappe */ private String addPrefixIfNecessary(String prefix, String propertyName) { String headerName = propertyName; - if (StringUtils.hasText(prefix) && !headerName.startsWith(prefix) && !headerName.equals(MessageHeaders.CONTENT_TYPE)) { + if (StringUtils.hasText(prefix) && !headerName.startsWith(prefix) && + !headerName.equals(MessageHeaders.CONTENT_TYPE) && + (!JsonHeaders.HEADERS.contains(headerName) || !JsonHeaders.HEADERS.contains(JsonHeaders.PREFIX + headerName))) { headerName = prefix + propertyName; } + if (JsonHeaders.HEADERS.contains(JsonHeaders.PREFIX + headerName)) { + headerName = JsonHeaders.PREFIX + headerName; + } return headerName; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java index 53f66d0d19..3393468cfd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java @@ -16,10 +16,14 @@ package org.springframework.integration.store.metadata; +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; import java.util.Properties; import org.apache.commons.logging.Log; @@ -37,6 +41,7 @@ import org.springframework.util.DefaultPropertiesPersister; * * @author Oleg Zhurakousky * @author Mark Fisher + * @author Gary Russell * @since 2.0 */ public class PropertiesPersistingMetadataStore implements MetadataStore, InitializingBean, DisposableBean { @@ -86,9 +91,9 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial } private void saveMetadata() { - FileOutputStream outputStream = null; + OutputStream outputStream = null; try { - outputStream = new FileOutputStream(this.file); + outputStream = new BufferedOutputStream(new FileOutputStream(this.file)); this.persister.store(this.metadata, outputStream, "Last feed entry"); } catch (IOException e) { @@ -104,15 +109,15 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial } catch (IOException e) { // not fatal for the functionality of the component - logger.warn("Failed to close FileOutputStream to " + this.file.getAbsolutePath(), e); + logger.warn("Failed to close OutputStream to " + this.file.getAbsolutePath(), e); } } } private void loadMetadata() { - FileInputStream inputStream = null; + InputStream inputStream = null; try { - inputStream = new FileInputStream(this.file); + inputStream = new BufferedInputStream(new FileInputStream(this.file)); this.persister.load(this.metadata, inputStream); } catch (Exception e) { @@ -128,7 +133,7 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial } catch (Exception e2) { // non fatal - logger.warn("Failed to close FileInputStream for: " + this.file.getAbsolutePath()); + logger.warn("Failed to close InputStream for: " + this.file.getAbsolutePath()); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java index 8d16b229fb..948fc0c6ee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java @@ -16,8 +16,10 @@ package org.springframework.integration.support.json; -import org.springframework.messaging.Message; +import java.lang.reflect.Type; + import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; /** * Base {@link JsonInboundMessageMapper.JsonMessageParser} implementation for Jackson processors. @@ -68,7 +70,7 @@ abstract class AbstractJacksonJsonMessageParser

implements JsonInboundMessage Class headerType = this.messageMapper.getHeaderTypes().containsKey(headerName) ? this.messageMapper.getHeaderTypes().get(headerName) : Object.class; try { - return this.objectMapper.fromJson(parser, headerType); + return this.objectMapper.fromJson(parser, (Type) headerType); } catch (Exception e) { throw new IllegalArgumentException("Mapping header '" + headerName + "' of JSON message '" + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonObjectMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonObjectMapper.java new file mode 100644 index 0000000000..5f2ff9b87b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonObjectMapper.java @@ -0,0 +1,84 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.support.json; + +import java.io.File; +import java.io.InputStream; +import java.io.Reader; +import java.lang.reflect.Type; +import java.net.URL; +import java.util.Arrays; +import java.util.Collection; +import java.util.Map; + +import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.util.ClassUtils; + +/** + * Base class for Jackson {@link JsonObjectMapper} implementations. + * + * @author Artem Bilan + * @since 3.0 + */ +public abstract class AbstractJacksonJsonObjectMapper implements JsonObjectMapper

, BeanClassLoaderAware { + + protected static final Collection> supportedJsonTypes = + Arrays.> asList(String.class, byte[].class, File.class, URL.class, InputStream.class, Reader.class); + + private volatile ClassLoader classLoader = ClassUtils.getDefaultClassLoader(); + + @Override + public void setBeanClassLoader(ClassLoader classLoader) { + this.classLoader = classLoader; + } + + @Override + public T fromJson(Object json, Class valueType) throws Exception { + return this.fromJson(json, this.constructType(valueType)); + } + + @Override + public T fromJson(Object json, Map javaTypes) throws Exception { + J javaType = this.extractJavaType(javaTypes); + return this.fromJson(json, javaType); + } + + protected J createJavaType(Map javaTypes, String javaTypeKey) throws Exception { + Object classValue = javaTypes.get(javaTypeKey); + if (classValue == null) { + throw new IllegalArgumentException("Could not resolve '" + javaTypeKey + "' in 'javaTypes'."); + } + else { + Class aClass = null; + if (classValue instanceof Class) { + aClass = (Class) classValue; + } + else { + aClass = ClassUtils.forName(classValue.toString(), this.classLoader); + } + + return this.constructType(aClass); + } + } + + protected abstract T fromJson(Object json, J type) throws Exception; + + protected abstract J extractJavaType(Map javaTypes) throws Exception; + + protected abstract J constructType(Type type); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java index 14bcb1d32b..0d802e5bc4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * 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. @@ -16,15 +16,22 @@ package org.springframework.integration.support.json; +import java.io.File; +import java.io.InputStream; import java.io.Reader; import java.io.Writer; import java.lang.reflect.Type; +import java.net.URL; +import java.util.Collection; +import java.util.Map; + +import org.springframework.integration.json.JsonHeaders; +import org.springframework.util.Assert; import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.JavaType; import com.fasterxml.jackson.databind.ObjectMapper; -import org.springframework.util.Assert; - /** * Jackson 2 JSON-processor (@link https://github.com/FasterXML) {@linkplain JsonObjectMapper} implementation. * Delegates toJson and fromJson @@ -33,7 +40,7 @@ import org.springframework.util.Assert; * @author Artem Bilan * @since 3.0 */ -public class Jackson2JsonObjectMapper implements JsonObjectMapper { +public class Jackson2JsonObjectMapper extends AbstractJacksonJsonObjectMapper { private final ObjectMapper objectMapper; @@ -46,6 +53,7 @@ public class Jackson2JsonObjectMapper implements JsonObjectMapper { this.objectMapper = objectMapper; } + @Override public String toJson(Object value) throws Exception { return this.objectMapper.writeValueAsString(value); } @@ -55,18 +63,72 @@ public class Jackson2JsonObjectMapper implements JsonObjectMapper { this.objectMapper.writeValue(writer, value); } - public T fromJson(String json, Class valueType) throws Exception { - return this.objectMapper.readValue(json, valueType); - } - @Override - public T fromJson(Reader json, Class valueType) throws Exception { - return this.objectMapper.readValue(json, valueType); + protected T fromJson(Object json, JavaType type) throws Exception { + if (json instanceof String) { + return this.objectMapper.readValue((String) json, type); + } + else if(json instanceof byte[]) { + return this.objectMapper.readValue((byte[]) json, type); + } + else if (json instanceof File) { + return this.objectMapper.readValue((File) json, type); + } + else if (json instanceof URL) { + return this.objectMapper.readValue((URL) json, type); + } + else if (json instanceof InputStream) { + return this.objectMapper.readValue((InputStream) json, type); + } + else if (json instanceof Reader) { + return this.objectMapper.readValue((Reader) json, type); + } + else { + throw new IllegalArgumentException("'json' argument must be an instance of: " + supportedJsonTypes); + } } @Override public T fromJson(JsonParser parser, Type valueType) throws Exception { - return this.objectMapper.readValue(parser, this.objectMapper.constructType(valueType)); + return this.objectMapper.readValue(parser, this.constructType(valueType)); + } + + @Override + public void populateJavaTypes(Map map, Class sourceClass) { + JavaType javaType = this.objectMapper.constructType(sourceClass); + map.put(JsonHeaders.TYPE_ID, javaType.getRawClass()); + + if (javaType.isContainerType() && !javaType.isArrayType()) { + map.put(JsonHeaders.CONTENT_TYPE_ID, javaType.getContentType().getRawClass()); + } + + if (javaType.getKeyType() != null) { + map.put(JsonHeaders.KEY_TYPE_ID, javaType.getKeyType().getRawClass()); + } + } + + @Override + @SuppressWarnings({ "unchecked" }) + protected JavaType extractJavaType(Map javaTypes) throws Exception { + JavaType classType = this.createJavaType(javaTypes, JsonHeaders.TYPE_ID); + if (!classType.isContainerType() || classType.isArrayType()) { + return classType; + } + + JavaType contentClassType = this.createJavaType(javaTypes, JsonHeaders.CONTENT_TYPE_ID); + if (classType.getKeyType() == null) { + return this.objectMapper.getTypeFactory() + .constructCollectionType((Class>) classType.getRawClass(), contentClassType); + } + + JavaType keyClassType = this.createJavaType(javaTypes, JsonHeaders.KEY_TYPE_ID); + return this.objectMapper.getTypeFactory() + .constructMapType((Class>) classType.getRawClass(), keyClassType, contentClassType); + } + + @Override + protected JavaType constructType(Type type) { + return this.objectMapper.constructType(type); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonObjectMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonObjectMapper.java index 7b3ded0ac9..5297a2bb33 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonObjectMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonObjectMapper.java @@ -16,13 +16,20 @@ package org.springframework.integration.support.json; +import java.io.File; +import java.io.InputStream; import java.io.Reader; import java.io.Writer; import java.lang.reflect.Type; +import java.net.URL; +import java.util.Collection; +import java.util.Map; import org.codehaus.jackson.JsonParser; import org.codehaus.jackson.map.ObjectMapper; +import org.codehaus.jackson.type.JavaType; +import org.springframework.integration.json.JsonHeaders; import org.springframework.util.Assert; /** @@ -33,7 +40,7 @@ import org.springframework.util.Assert; * @author Artem Bilan * @since 3.0 */ -public class JacksonJsonObjectMapper implements JsonObjectMapper { +public class JacksonJsonObjectMapper extends AbstractJacksonJsonObjectMapper { private final ObjectMapper objectMapper; @@ -46,6 +53,7 @@ public class JacksonJsonObjectMapper implements JsonObjectMapper { this.objectMapper = objectMapper; } + @Override public String toJson(Object value) throws Exception { return this.objectMapper.writeValueAsString(value); } @@ -55,18 +63,72 @@ public class JacksonJsonObjectMapper implements JsonObjectMapper { this.objectMapper.writeValue(writer, value); } - public T fromJson(String json, Class valueType) throws Exception { - return this.objectMapper.readValue(json, valueType); - } - - @Override - public T fromJson(Reader json, Class valueType) throws Exception { - return this.objectMapper.readValue(json, valueType); - } - @Override public T fromJson(JsonParser parser, Type valueType) throws Exception { - return this.objectMapper.readValue(parser, this.objectMapper.constructType(valueType)); + return this.objectMapper.readValue(parser, this.constructType(valueType)); + } + + @Override + protected T fromJson(Object json, JavaType type) throws Exception { + if (json instanceof String) { + return this.objectMapper.readValue((String) json, type); + } + else if(json instanceof byte[]) { + return this.objectMapper.readValue((byte[]) json, type); + } + else if (json instanceof File) { + return this.objectMapper.readValue((File) json, type); + } + else if (json instanceof URL) { + return this.objectMapper.readValue((URL) json, type); + } + else if (json instanceof InputStream) { + return this.objectMapper.readValue((InputStream) json, type); + } + else if (json instanceof Reader) { + return this.objectMapper.readValue((Reader) json, type); + } + else { + throw new IllegalArgumentException("'json' argument must be an instance of: " + supportedJsonTypes); + } + } + + @Override + public void populateJavaTypes(Map map, Class sourceClass) { + JavaType javaType = this.constructType(sourceClass); + map.put(JsonHeaders.TYPE_ID, javaType.getRawClass()); + + if (javaType.isContainerType() && !javaType.isArrayType()) { + map.put(JsonHeaders.CONTENT_TYPE_ID, javaType.getContentType().getRawClass()); + } + + if (javaType.getKeyType() != null) { + map.put(JsonHeaders.KEY_TYPE_ID, javaType.getKeyType().getRawClass()); + } + } + + @Override + protected JavaType constructType(Type type) { + return this.objectMapper.constructType(type); + } + + @Override + @SuppressWarnings({ "unchecked" }) + protected JavaType extractJavaType(Map javaTypes) throws Exception { + JavaType classType = this.createJavaType(javaTypes, JsonHeaders.TYPE_ID); + if (!classType.isContainerType() || classType.isArrayType()) { + return classType; + } + + JavaType contentClassType = this.createJavaType(javaTypes, JsonHeaders.CONTENT_TYPE_ID); + if (classType.getKeyType() == null) { + return this.objectMapper.getTypeFactory() + .constructCollectionType((Class>) classType.getRawClass(), contentClassType); + } + + JavaType keyClassType = this.createJavaType(javaTypes, JsonHeaders.KEY_TYPE_ID); + return this.objectMapper.getTypeFactory() + .constructMapType((Class>) classType.getRawClass(), keyClassType, contentClassType); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapper.java index 682f7c0720..70b3c8c35e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapper.java @@ -16,9 +16,9 @@ package org.springframework.integration.support.json; -import java.io.Reader; import java.io.Writer; import java.lang.reflect.Type; +import java.util.Map; /** * Strategy interface to convert an Object to/from the JSON representation. @@ -33,10 +33,11 @@ public interface JsonObjectMapper

{ void toJson(Object value, Writer writer) throws Exception; - T fromJson(String json, Class valueType) throws Exception; + T fromJson(Object json, Class valueType) throws Exception; - T fromJson(Reader json, Class valueType) throws Exception; + T fromJson(Object json, Map javaTypes) throws Exception; T fromJson(P parser, Type valueType) throws Exception; + void populateJavaTypes(Map map, Class sourceClass); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperAdapter.java index 8f7f1282f1..e5feaeec83 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperAdapter.java @@ -16,9 +16,9 @@ package org.springframework.integration.support.json; -import java.io.Reader; import java.io.Writer; import java.lang.reflect.Type; +import java.util.Map; /** * Simple {@linkplain JsonObjectMapper} adapter implementation, if there is no need @@ -39,12 +39,7 @@ public abstract class JsonObjectMapperAdapter

implements JsonObjectMapper

} @Override - public T fromJson(String json, Class valueType) throws Exception { - return null; - } - - @Override - public T fromJson(Reader json, Class valueType) throws Exception { + public T fromJson(Object json, Class valueType) throws Exception { return null; } @@ -53,4 +48,13 @@ public abstract class JsonObjectMapperAdapter

implements JsonObjectMapper

return null; } + @Override + public T fromJson(Object json, Map javaTypes) throws Exception { + return null; + } + + @Override + public void populateJavaTypes(Map map, Class sourceClass) { + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/ClassUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/ClassUtils.java index 5f2140e975..9558a76eb7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/ClassUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/ClassUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-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. @@ -16,6 +16,8 @@ package org.springframework.integration.util; +import java.util.HashMap; +import java.util.Map; import java.util.Set; /** @@ -24,6 +26,24 @@ import java.util.Set; */ public abstract class ClassUtils { + /** + * Map with primitive wrapper type as key and corresponding primitive + * type as value, for example: Integer.class -> int.class. + */ + private static final Map, Class> primitiveWrapperTypeMap = new HashMap, Class>(8); + + + static { + primitiveWrapperTypeMap.put(Boolean.class, boolean.class); + primitiveWrapperTypeMap.put(Byte.class, byte.class); + primitiveWrapperTypeMap.put(Character.class, char.class); + primitiveWrapperTypeMap.put(Double.class, double.class); + primitiveWrapperTypeMap.put(Float.class, float.class); + primitiveWrapperTypeMap.put(Integer.class, int.class); + primitiveWrapperTypeMap.put(Long.class, long.class); + primitiveWrapperTypeMap.put(Short.class, short.class); + } + public static Class findClosestMatch(Class type, Set> candidates, boolean failOnTie) { int minTypeDiffWeight = Integer.MAX_VALUE; Class closestMatch = null; @@ -35,7 +55,7 @@ public abstract class ClassUtils { } else if (failOnTie && typeDiffWeight < Integer.MAX_VALUE && (typeDiffWeight == minTypeDiffWeight)) { throw new IllegalStateException("Unresolvable ambiguity while attempting to find closest match for [" + - type.getName() + "]. Candidate types [" + closestMatch.getName() + "] and [" + candidate.getName() + + type.getName() + "]. Candidate types [" + closestMatch.getName() + "] and [" + candidate.getName() + "] have equal weight."); } } @@ -67,4 +87,14 @@ public abstract class ClassUtils { return result; } + /** + * Resolve the given class if it is a primitive wrapper class, + * returning the corresponding primitive type instead. + * @param clazz the wrapper class to check + * @return the corresponding primitive if the clazz is a wrapper, otherwise null + */ + public static Class resolvePrimitiveType(Class clazz) { + return primitiveWrapperTypeMap.get(clazz); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/StackTraceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/StackTraceUtils.java new file mode 100644 index 0000000000..f76c2ee1cf --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/StackTraceUtils.java @@ -0,0 +1,49 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.util; + +/** + * Utility methods for analyzing stack traces. + * + * @author Gary Russell + * @since 3.0 + * + */ +public class StackTraceUtils { + + private StackTraceUtils() {} + + /** + * Traverses the stack trace element array looking for instances that contain the first or second + * Strings in the className property. + * @param firstClass The first class to look for. + * @param secondClass The second class to look for. + * @param stackTrace The stack trace. + * @return true if the first class appears first, false if the second appears first + * @throws IllegalArgumentException if neither class is found. + */ + public static boolean isFrameContainingXBeforeFrameContainingY(String firstClass, String secondClass, StackTraceElement[] stackTrace) { + for (StackTraceElement element : stackTrace) { + if (element.getClassName().contains(firstClass)) { + return true; + } + else if (element.getClassName().contains(secondClass)) { + return false; + } + } + throw new IllegalArgumentException("Neither " + firstClass + " nor " + secondClass + " class found"); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/LocalChannelRegistryTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/LocalChannelRegistryTests.java deleted file mode 100644 index 23760928fa..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/LocalChannelRegistryTests.java +++ /dev/null @@ -1,171 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on - * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ - -package org.springframework.integration.channel.registry; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - -import java.util.concurrent.atomic.AtomicBoolean; - -import org.junit.Before; -import org.junit.Test; - -import org.springframework.context.support.GenericApplicationContext; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessagingException; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.messaging.support.GenericMessage; - -/** - * @author David Turanski - * @author Mark Fisher - * @since 3.0 - */ -public class LocalChannelRegistryTests { - - private LocalChannelRegistry registry = new LocalChannelRegistry(); - - private GenericApplicationContext context = new GenericApplicationContext(); - - @Before - public void setUp() { - registry.setApplicationContext(context); - context.refresh(); - } - - @Test - public void testInbound() { - DirectChannel channel = new DirectChannel(); - registry.inbound("inbound", channel); - assertTrue(context.containsBean("inbound")); - - final AtomicBoolean messageReceived = new AtomicBoolean(); - channel.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - messageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - SubscribableChannel registeredChannel = context.getBean("inbound", SubscribableChannel.class); - registeredChannel.send(new GenericMessage("hello")); - assertTrue(messageReceived.get()); - } - - @Test - public void testOutbound() { - DirectChannel channel = new DirectChannel(); - registry.outbound("outbound", channel); - assertTrue(context.containsBean("outbound")); - - final AtomicBoolean messageReceived = new AtomicBoolean(); - SubscribableChannel registeredChannel = context.getBean("outbound", SubscribableChannel.class); - registeredChannel.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - messageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - channel.send(new GenericMessage("hello")); - assertTrue(messageReceived.get()); - } - - @Test(expected = IllegalArgumentException.class) - public void testOutboundTapShouldFail() { - DirectChannel channel = new DirectChannel(); - registry.outbound("outbound", channel); - DirectChannel tapChannel = new DirectChannel(); - registry.tap("outbound", tapChannel); - } - - @Test - public void testInboundTap() { - DirectChannel channel = new DirectChannel(); - registry.inbound("inbound", channel); - DirectChannel tapChannel = new DirectChannel(); - registry.tap("inbound", tapChannel); - final AtomicBoolean originalMessageReceived = new AtomicBoolean(); - final AtomicBoolean tapMessageReceived = new AtomicBoolean(); - tapChannel.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - tapMessageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - channel.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - originalMessageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - MessageChannel registeredChannel = context.getBean("inbound", MessageChannel.class); - registeredChannel.send(new GenericMessage("hello")); - assertTrue(originalMessageReceived.get()); - assertTrue(tapMessageReceived.get()); - } - - @Test - public void testFlowThroughRegisteredChannelFromOutboundToInbound() { - DirectChannel outbound = new DirectChannel(); - DirectChannel inbound = new DirectChannel(); - registry.outbound("foo", outbound); - registry.inbound("foo", inbound); - final AtomicBoolean messageReceived = new AtomicBoolean(); - inbound.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - messageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - outbound.send(new GenericMessage("hello")); - assertTrue(messageReceived.get()); - } - - @Test - public void testFlowThroughRegisteredChannelFromOutboundToInboundWithTap() { - DirectChannel outbound = new DirectChannel(); - DirectChannel inbound = new DirectChannel(); - DirectChannel tap = new DirectChannel(); - registry.outbound("foo", outbound); - registry.inbound("foo", inbound); - registry.tap("foo", tap); - final AtomicBoolean originalMessageReceived = new AtomicBoolean(); - final AtomicBoolean tapMessageReceived = new AtomicBoolean(); - inbound.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - originalMessageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - tap.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - tapMessageReceived.set(true); - assertEquals("hello", message.getPayload()); - } - }); - outbound.send(new GenericMessage("hello")); - assertTrue(originalMessageReceived.get()); - assertTrue(tapMessageReceived.get()); - } - -} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml index 5b2da86780..508e3cc386 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml @@ -22,19 +22,19 @@ - - - diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-fail-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-fail-context.xml new file mode 100644 index 0000000000..abbc47f5fb --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-fail-context.xml @@ -0,0 +1,19 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java index b6766077e7..e81fa6a847 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java @@ -17,16 +17,24 @@ package org.springframework.integration.handler; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.StackTraceUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -90,7 +98,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("TEST", reply.getPayload()); assertEquals("replyingHandlerTestInputChannel,replyingHandlerTestService", reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertEquals("doDispatch", st[3].getMethodName()); // close to the metal + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); // close to the metal } @Test @@ -103,7 +111,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("optimizedRefReplyingHandlerTestInputChannel,optimizedRefReplyingHandlerTestService", reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertEquals("doDispatch", st[3].getMethodName()); + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); // close to the metal } @Test @@ -115,7 +123,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("TEST", reply.getPayload()); assertEquals("replyingHandlerWithStandardMethodTestInputChannel,replyingHandlerWithStandardMethodTestService", reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertEquals("doDispatch", st[3].getMethodName()); // close to the metal + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); // close to the metal } @Test @@ -149,6 +157,22 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("processorTestInputChannel,processorTestService", reply.getHeaders().get("history").toString()); } + @Test + public void testFailOnDoubleReference() { + try { + new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "-fail-context.xml", + this.getClass()); + fail("Expected exception due to 2 endpoints referencing the same bean"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(BeanCreationException.class)); + assertThat(e.getCause(), Matchers.instanceOf(BeanCreationException.class)); + assertThat(e.getCause().getCause(), Matchers.instanceOf(IllegalArgumentException.class)); + assertThat(e.getCause().getCause().getMessage(), + Matchers.containsString("An AbstractReplyProducingMessageHandler may only be referenced once")); + } + + } @SuppressWarnings("unused") private static class TestReplyingMessageHandler extends AbstractReplyProducingMessageHandler { @@ -162,6 +186,10 @@ public class ServiceActivatorDefaultFrameworkMethodTests { } public String foo(String in) { + Exception e = new RuntimeException(); + StackTraceElement[] st = e.getStackTrace(); + // use this to test that StackTraceUtils works as expected and returns false + assertFalse(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); return "bar"; } @@ -174,7 +202,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { public void handleMessage(Message requestMessage) { Exception e = new RuntimeException(); StackTraceElement[] st = e.getStackTrace(); - assertEquals("doDispatch", st[4].getMethodName()); + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); // close to the metal } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerParserTests-context.xml index 7034732677..6df09b437e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerParserTests-context.xml @@ -8,16 +8,16 @@ http://www.springframework.org/schema/integration/spring-integration.xsd"> + type="org.springframework.integration.json.TestPerson"/> diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerParserTests.java index ff45f7fe84..cfc04444e8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerParserTests.java @@ -134,79 +134,6 @@ public class JsonToObjectTransformerParserTests { } - static class TestPerson { - - private String firstName; - - private String lastName; - - private int age; - - private TestAddress address; - - - public String getFirstName() { - return this.firstName; - } - - public void setFirstName(String firstName) { - this.firstName = firstName; - } - - public void setLastName(String lastName) { - this.lastName = lastName; - } - - public String getLastName() { - return this.lastName; - } - - public void setAge(int age) { - this.age = age; - } - - public int getAge() { - return this.age; - } - - public void setAddress(TestAddress address) { - this.address = address; - } - - public TestAddress getAddress() { - return this.address; - } - - @Override - public String toString() { - return "name=" + this.firstName + " " + this.lastName - + ", age=" + this.age + ", address=" + this.address; - } - } - - - static class TestAddress { - - private int number; - - private String street; - - - public void setNumber(int number) { - this.number = number; - } - - public void setStreet(String street) { - this.street = street; - } - - @Override - public String toString() { - return this.number + " " + this.street; - } - } - - static class CustomObjectMapper extends ObjectMapper { public CustomObjectMapper() { @@ -219,8 +146,8 @@ public class JsonToObjectTransformerParserTests { static class CustomJsonObjectMapper extends JsonObjectMapperAdapter { @Override - public Object fromJson(String json, Class valueType) throws Exception { - return new TestJsonContainer(json); + public Object fromJson(Object json, Class valueType) throws Exception { + return new TestJsonContainer((String) json); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerTests.java index 9ae4a6771a..aa583bc275 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonToObjectTransformerTests.java @@ -22,6 +22,8 @@ import org.codehaus.jackson.JsonParser.Feature; import org.codehaus.jackson.map.ObjectMapper; import org.junit.Test; +import org.springframework.integration.Message; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.json.JacksonJsonObjectMapper; /** @@ -33,9 +35,11 @@ public class JsonToObjectTransformerTests { @Test public void objectPayload() throws Exception { - JsonToObjectTransformer transformer = new JsonToObjectTransformer(TestPerson.class); + JsonToObjectTransformer transformer = new JsonToObjectTransformer(TestPerson.class); String jsonString = "{\"firstName\":\"John\",\"lastName\":\"Doe\",\"age\":42,\"address\":{\"number\":123,\"street\":\"Main Street\"}}"; - TestPerson person = transformer.transformPayload(jsonString); + Message message = transformer.transform(new GenericMessage(jsonString)); + @SuppressWarnings("unchecked") + TestPerson person = (TestPerson) message.getPayload(); assertEquals("John", person.getFirstName()); assertEquals("Doe", person.getLastName()); assertEquals(42, person.getAge()); @@ -47,10 +51,12 @@ public class JsonToObjectTransformerTests { ObjectMapper customMapper = new ObjectMapper(); customMapper.configure(Feature.ALLOW_UNQUOTED_FIELD_NAMES, Boolean.TRUE); customMapper.configure(Feature.ALLOW_SINGLE_QUOTES, Boolean.TRUE); - JsonToObjectTransformer transformer = - new JsonToObjectTransformer(TestPerson.class, new JacksonJsonObjectMapper(customMapper)); + JsonToObjectTransformer transformer = + new JsonToObjectTransformer(TestPerson.class, new JacksonJsonObjectMapper(customMapper)); String jsonString = "{firstName:'John', lastName:'Doe', age:42, address:{number:123, street:'Main Street'}}"; - TestPerson person = transformer.transformPayload(jsonString); + Message message = transformer.transform(new GenericMessage(jsonString)); + @SuppressWarnings("unchecked") + TestPerson person = (TestPerson) message.getPayload(); assertEquals("John", person.getFirstName()); assertEquals("Doe", person.getLastName()); assertEquals(42, person.getAge()); @@ -60,82 +66,8 @@ public class JsonToObjectTransformerTests { @SuppressWarnings("deprecation") @Test(expected = IllegalArgumentException.class) public void testInt2831IllegalArgument() throws Exception { - new JsonToObjectTransformer(String.class, new Object()); + new JsonToObjectTransformer(String.class, new Object()); } - @SuppressWarnings("unused") - private static class TestPerson { - - private String firstName; - - private String lastName; - - private int age; - - private TestAddress address; - - - public String getFirstName() { - return this.firstName; - } - - public void setFirstName(String firstName) { - this.firstName = firstName; - } - - public void setLastName(String lastName) { - this.lastName = lastName; - } - - public String getLastName() { - return this.lastName; - } - - public void setAge(int age) { - this.age = age; - } - - public int getAge() { - return this.age; - } - - public void setAddress(TestAddress address) { - this.address = address; - } - - public TestAddress getAddress() { - return this.address; - } - - @Override - public String toString() { - return "name=" + this.firstName + " " + this.lastName - + ", age=" + this.age + ", address=" + this.address; - } - } - - - @SuppressWarnings("unused") - private static class TestAddress { - - private int number; - - private String street; - - - public void setNumber(int number) { - this.number = number; - } - - public void setStreet(String street) { - this.street = street; - } - - @Override - public String toString() { - return this.number + " " + this.street; - } - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonTransformersSymmetricalTests.java b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonTransformersSymmetricalTests.java new file mode 100644 index 0000000000..5f81466485 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonTransformersSymmetricalTests.java @@ -0,0 +1,47 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.json; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; + +import org.springframework.integration.Message; +import org.springframework.integration.message.GenericMessage; + +/** + * @author Artem Bilan + * @since 3.0 + */ +public class JsonTransformersSymmetricalTests { + + @Test + public void testInt2809ObjectToJson_JsonToObject() { + + TestPerson person = new TestPerson("John", "Doe", 42); + person.setAddress(new TestAddress(123, "Main Street")); + + ObjectToJsonTransformer objectToJsonTransformer = new ObjectToJsonTransformer(); + Message jsonMessage = objectToJsonTransformer.transform(new GenericMessage(person)); + + JsonToObjectTransformer jsonToObjectTransformer = new JsonToObjectTransformer(); + Message result = jsonToObjectTransformer.transform(jsonMessage); + + assertEquals(person, result.getPayload()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerParserTests.java index 3cf107ff94..5bb4550754 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerParserTests.java @@ -174,87 +174,6 @@ public class ObjectToJsonTransformerParserTests { } - static class TestPerson { - - private String firstName; - - private String lastName; - - private int age; - - private TestAddress address; - - - public String getFirstName() { - return this.firstName; - } - - public void setFirstName(String firstName) { - this.firstName = firstName; - } - - public void setLastName(String lastName) { - this.lastName = lastName; - } - - public String getLastName() { - return this.lastName; - } - - public void setAge(int age) { - this.age = age; - } - - public int getAge() { - return this.age; - } - - public void setAddress(TestAddress address) { - this.address = address; - } - - public TestAddress getAddress() { - return this.address; - } - - @Override - public String toString() { - return "\"name\":\"" + this.firstName + " " + this.lastName - + "\", \"age\":" + this.age + ", \"address\":\"" + this.address + "\""; - } - } - - - static class TestAddress { - - private int number; - - private String street; - - - public int getNumber() { - return this.number; - } - - public void setNumber(int number) { - this.number = number; - } - - public String getStreet() { - return this.street; - } - - public void setStreet(String street) { - this.street = street; - } - - @Override - public String toString() { - return this.number + " " + this.street; - } - } - - static class CustomObjectMapper extends ObjectMapper { public CustomObjectMapper() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.java index 0f5ea8e1df..24ee63232f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.java @@ -147,66 +147,4 @@ public class ObjectToJsonTransformerTests { new ObjectToJsonTransformer(new Object()); } - @SuppressWarnings("unused") - private static class TestPerson { - - private final String firstName; - - private final String lastName; - - private final int age; - - private TestAddress address; - - - public TestPerson(String firstName, String lastName, int age) { - this.firstName = firstName; - this.lastName = lastName; - this.age = age; - } - - public String getFirstName() { - return firstName; - } - - public String getLastName() { - return lastName; - } - - public int getAge() { - return age; - } - - public TestAddress getAddress() { - return address; - } - - public void setAddress(TestAddress address) { - this.address = address; - } - } - - - @SuppressWarnings("unused") - private static class TestAddress { - - private final int number; - - private final String street; - - - public TestAddress(int number, String street) { - this.number = number; - this.street = street; - } - - public int getNumber() { - return number; - } - - public String getStreet() { - return street; - } - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/TestAddress.java b/spring-integration-core/src/test/java/org/springframework/integration/json/TestAddress.java new file mode 100644 index 0000000000..ac6039e1c3 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/TestAddress.java @@ -0,0 +1,80 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.json; + +/** + * @author Mark Fisher + * @since 2.0 +*/ +@SuppressWarnings("unused") +class TestAddress { + + private volatile int number; + + private volatile String street; + + TestAddress() { + } + + public TestAddress(int number, String street) { + this.number = number; + this.street = street; + } + + + public int getNumber() { + return this.number; + } + + public void setNumber(int number) { + this.number = number; + } + + public String getStreet() { + return this.street; + } + + public void setStreet(String street) { + this.street = street; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + TestAddress that = (TestAddress) o; + + if (number != that.number) return false; + if (street != null ? !street.equals(that.street) : that.street != null) return false; + + return true; + } + + @Override + public int hashCode() { + int result = number; + result = 31 * result + (street != null ? street.hashCode() : 0); + return result; + } + + @Override + public String toString() { + return this.number + " " + this.street; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/TestPerson.java b/spring-integration-core/src/test/java/org/springframework/integration/json/TestPerson.java new file mode 100644 index 0000000000..e77bf4d52a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/TestPerson.java @@ -0,0 +1,105 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.json; + +/** + * @author Mark Fisher + * @since 2.0 +*/ +@SuppressWarnings("unused") +class TestPerson { + + private volatile String firstName; + + private volatile String lastName; + + private volatile int age; + + private volatile TestAddress address; + + TestPerson() { + } + + public TestPerson(String firstName, String lastName, int age) { + this.firstName = firstName; + this.lastName = lastName; + this.age = age; + } + + + public String getFirstName() { + return this.firstName; + } + + public void setFirstName(String firstName) { + this.firstName = firstName; + } + + public void setLastName(String lastName) { + this.lastName = lastName; + } + + public String getLastName() { + return this.lastName; + } + + public void setAge(int age) { + this.age = age; + } + + public int getAge() { + return this.age; + } + + public void setAddress(TestAddress address) { + this.address = address; + } + + public TestAddress getAddress() { + return this.address; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + TestPerson that = (TestPerson) o; + + if (age != that.age) return false; + if (address != null ? !address.equals(that.address) : that.address != null) return false; + if (firstName != null ? !firstName.equals(that.firstName) : that.firstName != null) return false; + if (lastName != null ? !lastName.equals(that.lastName) : that.lastName != null) return false; + + return true; + } + + @Override + public int hashCode() { + int result = firstName != null ? firstName.hashCode() : 0; + result = 31 * result + (lastName != null ? lastName.hashCode() : 0); + result = 31 * result + age; + result = 31 * result + (address != null ? address.hashCode() : 0); + return result; + } + + @Override + public String toString() { + return "name=" + this.firstName + " " + this.lastName + + ", age=" + this.age + ", address=" + this.address; + } +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index a8a6dbee69..04d0109a26 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -16,11 +16,15 @@ package org.springframework.integration.file; +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.BufferedWriter; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; import java.io.OutputStreamWriter; +import java.io.Writer; import java.nio.charset.Charset; import org.apache.commons.logging.Log; @@ -329,12 +333,12 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand private File handleFileMessage(final File sourceFile, File tempFile, final File resultFile) throws IOException { if (FileExistsMode.APPEND.equals(this.fileExistsMode)){ File fileToWriteTo = this.determineFileToWrite(resultFile, tempFile); - final FileOutputStream fos = new FileOutputStream(fileToWriteTo, true); - final FileInputStream fis = new FileInputStream(sourceFile); + final BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(fileToWriteTo, true)); + final BufferedInputStream bis = new BufferedInputStream(new FileInputStream(sourceFile)); WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry, fileToWriteTo.getAbsolutePath()){ @Override protected void whileLocked() throws IOException { - FileCopyUtils.copy(fis, fos); + FileCopyUtils.copy(bis, bos); } }; whileLockedProcessor.doWhileLocked(); @@ -362,11 +366,11 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode); - final FileOutputStream fos = new FileOutputStream(fileToWriteTo, append); + final BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(fileToWriteTo, append)); WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry, fileToWriteTo.getAbsolutePath()){ @Override protected void whileLocked() throws IOException { - FileCopyUtils.copy(bytes, fos); + FileCopyUtils.copy(bytes, bos); } }; @@ -380,7 +384,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand final boolean append = FileExistsMode.APPEND.equals(this.fileExistsMode); - final OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(fileToWriteTo, append), this.charset); + final Writer writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(fileToWriteTo, append), this.charset)); WhileLockedProcessor whileLockedProcessor = new WhileLockedProcessor(this.lockRegistry, fileToWriteTo.getAbsolutePath()){ @Override protected void whileLocked() throws IOException { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index f182e46633..6606a873b8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -16,6 +16,7 @@ package org.springframework.integration.file.remote.gateway; +import java.io.BufferedOutputStream; import java.io.File; import java.io.FileNotFoundException; import java.io.FileOutputStream; @@ -521,9 +522,9 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply if (!localFile.exists()) { String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix; File tempFile = new File(tempFileName); - FileOutputStream fileOutputStream = new FileOutputStream(tempFile); + BufferedOutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile)); try { - session.read(remoteFilePath, fileOutputStream); + session.read(remoteFilePath, outputStream); } catch (Exception e) { if (e instanceof RuntimeException){ @@ -535,7 +536,7 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply } finally { try { - fileOutputStream.close(); + outputStream.close(); } catch (Exception ignored2) { //Ignore it diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index 0ccb791287..fd20dbf108 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java @@ -16,9 +16,11 @@ package org.springframework.integration.file.remote.synchronizer; +import java.io.BufferedOutputStream; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; +import java.io.OutputStream; import java.util.Arrays; import java.util.Collection; import java.util.List; @@ -202,9 +204,9 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS if (!localFile.exists()) { String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix; File tempFile = new File(tempFileName); - FileOutputStream fileOutputStream = new FileOutputStream(tempFile); + OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile)); try { - session.read(remoteFilePath, fileOutputStream); + session.read(remoteFilePath, outputStream); } catch (Exception e) { if (e instanceof RuntimeException){ @@ -216,7 +218,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS } finally { try { - fileOutputStream.close(); + outputStream.close(); } catch (Exception ignored2) { } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/transformer/FileToStringTransformer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/transformer/FileToStringTransformer.java index d8731be3c9..f86327f661 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/transformer/FileToStringTransformer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/transformer/FileToStringTransformer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-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. @@ -16,9 +16,11 @@ package org.springframework.integration.file.transformer; +import java.io.BufferedReader; import java.io.File; import java.io.FileInputStream; import java.io.InputStreamReader; +import java.io.Reader; import java.nio.charset.Charset; import org.springframework.util.Assert; @@ -26,8 +28,9 @@ import org.springframework.util.FileCopyUtils; /** * A payload transformer that copies a File's contents to a String. - * + * * @author Mark Fisher + * @author Gary Russell */ public class FileToStringTransformer extends AbstractFilePayloadTransformer { @@ -45,7 +48,7 @@ public class FileToStringTransformer extends AbstractFilePayloadTransformer valueClass = value.getClass(); + if (valueClass.getName().equals(paramInfo.getType())) { + return true; + } + else { + Class primitiveType = ClassUtils.resolvePrimitiveType(valueClass); + return primitiveType != null && primitiveType.getName().equals(paramInfo.getType()); + } + } + /** * First checks if defaultObjectName is set, otherwise falls back on {@link JmxHeaders#OBJECT_NAME} header. */ @@ -229,4 +240,4 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa return map; } -} \ No newline at end of file +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java index c03d788bf3..e41560c10f 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2012 the original author or authors. - * + * Copyright 2002-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. @@ -33,7 +33,7 @@ import org.springframework.util.StringUtils; /** * Parser for the 'mbean-export' element of the integration JMX namespace. - * + * * @author Mark Fisher * @author Gary Russell * @since 2.0 @@ -61,7 +61,8 @@ public class MBeanExporterParser extends AbstractSingleBeanDefinitionParser { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "object-name-static-properties"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "managed-components", "componentNamePatterns"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "shutdown-executor"); - + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "object-naming-strategy", "namingStrategy"); + builder.addPropertyValue("server", mbeanServer); this.registerMBeanExporterHelper(parserContext.getRegistry()); } @@ -70,7 +71,7 @@ public class MBeanExporterParser extends AbstractSingleBeanDefinitionParser { BeanDefinitionBuilder mBeanExporterHelperBuilder = BeanDefinitionBuilder.rootBeanDefinition(MBeanExporterHelper.class); BeanDefinitionReaderUtils.registerWithGeneratedName(mBeanExporterHelperBuilder.getBeanDefinition(), registry); } - + private Object getMBeanServer(Element element, ParserContext parserContext) { String mbeanServer = element.getAttribute("server"); if (StringUtils.hasText(mbeanServer)) { diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index 2365ccf5d8..91efa883df 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -27,6 +27,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; import javax.management.DynamicMBean; @@ -80,6 +81,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; import org.springframework.util.PatternMatchUtils; import org.springframework.util.ReflectionUtils; +import org.springframework.util.ReflectionUtils.FieldCallback; +import org.springframework.util.ReflectionUtils.FieldFilter; /** *

@@ -106,6 +109,7 @@ import org.springframework.util.ReflectionUtils; * @author Helena Edelson * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ @ManagedResource public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostProcessor, BeanFactoryAware, @@ -163,7 +167,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private final MetadataMBeanInfoAssembler assembler = new MetadataMBeanInfoAssembler(attributeSource); - private final MetadataNamingStrategy namingStrategy = new MetadataNamingStrategy(attributeSource); + private final MetadataNamingStrategy defaultNamingStrategy = new MetadataNamingStrategy(attributeSource); private String[] componentNamePatterns = { "*" }; @@ -179,7 +183,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP super(); // Shouldn't be necessary, but to be on the safe side... setAutodetect(false); - setNamingStrategy(namingStrategy); + setNamingStrategy(defaultNamingStrategy); setAssembler(assembler); } @@ -206,7 +210,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP */ public void setDefaultDomain(String domain) { this.domain = domain; - this.namingStrategy.setDefaultDomain(domain); + this.defaultNamingStrategy.setDefaultDomain(domain); } public void setComponentNamePatterns(String[] componentNamePatterns) { @@ -217,7 +221,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { super.setBeanFactory(beanFactory); - Assert.isTrue(beanFactory instanceof ListableBeanFactory, "A ListableBeanFactory is required."); + Assert.isInstanceOf(ListableBeanFactory.class, beanFactory, "A ListableBeanFactory is required."); this.beanFactory = (ListableBeanFactory) beanFactory; } @@ -246,6 +250,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } if (bean instanceof MessageHandler) { + if (this.handlerInAnonymousWrapper(bean) != null) { + if (logger.isDebugEnabled()) { + logger.debug("Skipping " + beanName + " because it wraps another handler"); + } + return bean; + } SimpleMessageHandlerMetrics monitor = new SimpleMessageHandlerMetrics((MessageHandler) bean); Object advised = applyHandlerInterceptor(bean, monitor, beanClassLoader); handlers.add(monitor); @@ -282,6 +292,33 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } + private MessageHandler handlerInAnonymousWrapper(final Object bean) { + if (bean != null && bean.getClass().isAnonymousClass()) { + final AtomicReference wrapped = new AtomicReference(); + ReflectionUtils.doWithFields(bean.getClass(), new FieldCallback() { + + @Override + public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException { + field.setAccessible(true); + Object handler = field.get(bean); + if (handler instanceof MessageHandler) { + wrapped.set((MessageHandler) handler); + } + } + }, new FieldFilter() { + + @Override + public boolean matches(Field field) { + return wrapped.get() == null && field.getName().startsWith("val$"); + } + }); + return wrapped.get(); + } + else { + return null; + } + } + /** * Copy of private method in super class. Needed so we can avoid using the bean factory to extract the bean again, * and risk it being a proxy (which it almost certainly is by now). @@ -987,20 +1024,22 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP String source = "endpoint"; Object endpoint = null; + MessageHandler messageHandler = monitor.getMessageHandler(); + for (String beanName : names) { endpoint = beanFactory.getBean(beanName); - Object field = null; try { - field = extractTarget(getField(endpoint, "handler")); + Object field = extractTarget(getField(endpoint, "handler")); + if (field == messageHandler || + this.extractTarget(this.handlerInAnonymousWrapper(field)) == messageHandler) { + name = beanName; + endpointName = beanName; + break; + } } catch (Exception e) { logger.trace("Could not get handler from bean = " + beanName); } - if (field == monitor.getMessageHandler()) { - name = beanName; - endpointName = beanName; - break; - } } if (name != null && endpoint != null && name.startsWith("_org.springframework.integration")) { name = getInternalComponentName(name); @@ -1044,11 +1083,11 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } if (name == null) { - if (monitor.getMessageHandler() instanceof NamedComponent) { - name = ((NamedComponent) monitor.getMessageHandler()).getComponentName(); + if (messageHandler instanceof NamedComponent) { + name = ((NamedComponent) messageHandler).getComponentName(); } if (name == null) { - name = monitor.getMessageHandler().toString(); + name = messageHandler.toString(); } source = "handler"; } diff --git a/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-3.0.xsd b/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-3.0.xsd index 04f8b34b2c..28dc8ecf52 100644 --- a/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-3.0.xsd +++ b/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-3.0.xsd @@ -204,6 +204,19 @@ + + + + + + + + + An ObjectNamingStrategy to generate the MBean's ObjectName. See the reference documentation + for details of the default ObjectName. Also see 'object-name-static-properties'. + + + diff --git a/spring-integration-jmx/src/test/java/log4j.properties b/spring-integration-jmx/src/test/java/log4j.properties index 9246efaccb..0c503ac468 100644 --- a/spring-integration-jmx/src/test/java/log4j.properties +++ b/spring-integration-jmx/src/test/java/log4j.properties @@ -6,6 +6,6 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m log4j.category.org.springframework=WARN -log4j.category.org.springframework.integration=DEBUG -log4j.category.org.springframework.beans.factory=DEBUG +#log4j.category.org.springframework.integration=DEBUG +#log4j.category.org.springframework.beans.factory=DEBUG #log4j.category.org.springframework.integration.monitor=TRACE diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests-context.xml index a222fe42dc..0e5a5f7c13 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests-context.xml @@ -27,19 +27,19 @@ - - - diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests-fail-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests-fail-context.xml new file mode 100644 index 0000000000..942235110a --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests-fail-context.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java index aaec3d8b47..16da50f033 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/ServiceActivatorDefaultFrameworkMethodTests.java @@ -17,18 +17,26 @@ package org.springframework.integration.jmx; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.StackTraceUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -92,7 +100,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("TEST", reply.getPayload()); assertEquals("replyingHandlerTestInputChannel,replyingHandlerTestService", reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertEquals("doDispatch", st[15].getMethodName()); // close to the metal + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); } @Test @@ -105,7 +113,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("optimizedRefReplyingHandlerTestInputChannel,optimizedRefReplyingHandlerTestService", reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertEquals("doDispatch", st[15].getMethodName()); + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); } @Test @@ -117,7 +125,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("TEST", reply.getPayload()); assertEquals("replyingHandlerWithStandardMethodTestInputChannel,replyingHandlerWithStandardMethodTestService", reply.getHeaders().get("history").toString()); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertEquals("doDispatch", st[15].getMethodName()); // close to the metal + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); } @Test @@ -151,6 +159,23 @@ public class ServiceActivatorDefaultFrameworkMethodTests { assertEquals("processorTestInputChannel,processorTestService", reply.getHeaders().get("history").toString()); } + @Test + public void testFailOnDoubleReference() { + try { + new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "-fail-context.xml", + this.getClass()); + fail("Expected exception due to 2 endpoints referencing the same bean"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(BeanCreationException.class)); + assertThat(e.getCause(), Matchers.instanceOf(BeanCreationException.class)); + assertThat(e.getCause().getCause(), Matchers.instanceOf(IllegalArgumentException.class)); + assertThat(e.getCause().getCause().getMessage(), + Matchers.containsString("An AbstractReplyProducingMessageHandler may only be referenced once")); + } + + } + private interface Foo { public String foo(String in); @@ -169,6 +194,10 @@ public class ServiceActivatorDefaultFrameworkMethodTests { } public String foo(String in) { + Exception e = new RuntimeException(); + StackTraceElement[] st = e.getStackTrace(); + // use this to test that StackTraceUtils works as expected and returns false + assertFalse(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); return "bar"; } @@ -181,7 +210,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests { public void handleMessage(Message requestMessage) { Exception e = new RuntimeException(); StackTraceElement[] st = e.getStackTrace(); - assertEquals("doDispatch", st[28].getMethodName()); + assertTrue(StackTraceUtils.isFrameContainingXBeforeFrameContainingY("Dispatcher", "MethodInvokerHelper", st)); } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml index fc496855ff..3b419513d5 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml @@ -18,11 +18,14 @@ + object-name-static-properties="appProperties" + object-naming-strategy="keyNamer"/> foo bar + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java index 03c6c870f1..1565113abe 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.jmx.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import java.util.Properties; @@ -57,6 +58,7 @@ public class MBeanExporterParserTests { assertTrue(properties.containsKey("foo")); assertTrue(properties.containsKey("bar")); assertEquals(server, exporter.getServer()); + assertSame(context.getBean("keyNamer"), TestUtils.getPropertyValue(exporter, "namingStrategy")); exporter.destroy(); } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanRegistrationCustomNamingTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanRegistrationCustomNamingTests-context.xml new file mode 100644 index 0000000000..2e41b34f61 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanRegistrationCustomNamingTests-context.xml @@ -0,0 +1,35 @@ + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanRegistrationCustomNamingTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanRegistrationCustomNamingTests.java new file mode 100644 index 0000000000..c47cc0363d --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanRegistrationCustomNamingTests.java @@ -0,0 +1,72 @@ +/* + * Copyright 2013 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.jmx.config; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.Set; + +import javax.management.MBeanServer; +import javax.management.MalformedObjectNameException; +import javax.management.ObjectName; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jmx.export.naming.KeyNamingStrategy; +import org.springframework.jmx.export.naming.ObjectNamingStrategy; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * @author Gary Russell + * @since 3.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class MBeanRegistrationCustomNamingTests { + + @Autowired + private MBeanServer server; + + @Test + public void testHandlerMBeanRegistration() throws Exception { + Set names = server.queryNames(new ObjectName("test.MBeanRegistration:type=Integration,componentType=MessageHandler,*"), null); + assertEquals(6, names.size()); + assertTrue(names.contains(new ObjectName("test.MBeanRegistration:type=Integration,componentType=MessageHandler,name=chain,bean=endpoint"))); + assertTrue(names.contains(new ObjectName("test.MBeanRegistration:type=Integration,componentType=MessageHandler,name=chain$child.t1,bean=handler"))); + assertTrue(names.contains(new ObjectName("test.MBeanRegistration:type=Integration,componentType=MessageHandler,name=chain$child.f1,bean=handler"))); + } + + public static class Source { + public String get() { + return "foo"; + } + } + + public static class Namer implements ObjectNamingStrategy { + + private final ObjectNamingStrategy realNamer = new KeyNamingStrategy(); + @Override + public ObjectName getObjectName(Object managedBean, String beanKey) throws MalformedObjectNameException { + String actualBeanKey = beanKey.replace("type=", "type=Integration,componentType="); + return realNamer.getObjectName(managedBean, actualBeanKey); + } + + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests-context.xml index f412ab7bd6..e9e3dd8e8e 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests-context.xml @@ -33,6 +33,14 @@ + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests.java index 2a2946cbba..f0d0578a72 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayTests.java @@ -17,31 +17,40 @@ package org.springframework.integration.jmx.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import org.hamcrest.Matchers; import org.junit.After; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; +import org.springframework.integration.jmx.OperationInvokingMessageHandler; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; -import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; -import org.springframework.integration.jmx.OperationInvokingMessageHandler; import org.springframework.messaging.support.GenericMessage; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + /** * @author Oleg Zhurakousky * @author Artem Bilan + * @author Gary Russell * */ @ContextConfiguration @@ -49,15 +58,15 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class OperationInvokingOutboundGatewayTests { @Autowired - @Qualifier("withReplyChannel") private MessageChannel withReplyChannel; @Autowired - @Qualifier("withReplyChannelOutput") + private MessageChannel primitiveChannel; + + @Autowired private PollableChannel withReplyChannelOutput; @Autowired - @Qualifier("withNoReplyChannel") private MessageChannel withNoReplyChannel; @Autowired @@ -90,6 +99,42 @@ public class OperationInvokingOutboundGatewayTests { assertEquals(3, adviceCalled); } + @Test + public void gatewayWithPrimitiveArgs() throws Exception { + primitiveChannel.send(new GenericMessage(new Object[] { true, 0L, 1 })); + assertEquals(1, testBean.messages.size()); + List argList = new ArrayList(); + argList.add(false); + argList.add(123L); + argList.add(42); + primitiveChannel.send(new GenericMessage>(argList)); + assertEquals(2, testBean.messages.size()); + Map argMap = new HashMap(); + argMap.put("p1", true); + argMap.put("p2", 0L); + argMap.put("p3", 42); + primitiveChannel.send(new GenericMessage>(argMap)); + assertEquals(3, testBean.messages.size()); + argMap.put("p2", true); + argMap.put("p1", 0L); + argMap.put("p3", 42); + try { + primitiveChannel.send(new GenericMessage>(argMap)); + fail("Expected Exception"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(MessagingException.class)); + assertThat(e.getMessage(), Matchers.containsString("failed to find JMX operation")); + } + // TODO: Uncomment when Spring Framework minimum is 3.2.3 +// argMap = new HashMap(); +// argMap.put("bool", true); +// argMap.put("time", 0L); +// argMap.put("foo", 42); +// primitiveChannel.send(new GenericMessage>(argMap)); +// assertEquals(4, testBean.messages.size()); + } + @Test public void gatewayWithNoReplyChannel() throws Exception { withNoReplyChannel.send(new GenericMessage("1")); @@ -102,7 +147,7 @@ public class OperationInvokingOutboundGatewayTests { @Test //INT-1029, INT-2822 public void testOutboundGatewayInsideChain() throws Exception { - List handlers = TestUtils.getPropertyValue(this.operationInvokingWithinChain, "handlers", List.class); + List handlers = TestUtils.getPropertyValue(this.operationInvokingWithinChain, "handlers", List.class); assertEquals(1, handlers.size()); Object handler = handlers.get(0); assertTrue(handler instanceof OperationInvokingMessageHandler); diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/TestBean.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/TestBean.java index fa34a4a236..e492db08f9 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/TestBean.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/TestBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-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. @@ -26,6 +26,7 @@ import org.springframework.jmx.export.annotation.ManagedResource; /** * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.0 */ @ManagedResource @@ -42,11 +43,16 @@ public class TestBean { public void test(String text) { this.messages.add(text); } - + @ManagedOperation public List testWithReturn(String text) { this.messages.add(text); return messages; } + @ManagedOperation + public void testPrimitiveArgs(boolean bool, long time, int foo) { + this.messages.add(bool + " " + time + " " + foo); + } + } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java new file mode 100644 index 0000000000..587d2277e3 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java @@ -0,0 +1,245 @@ +/* + * Copyright 2013 the original author or authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.redis.inbound; + +import java.util.concurrent.TimeUnit; + +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.BoundListOperations; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessagingException; +import org.springframework.integration.channel.MessagePublishingErrorHandler; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.util.ErrorHandlingTaskExecutor; +import org.springframework.jmx.export.annotation.ManagedMetric; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @author Gunnar Hillert + * @author Artem Bilan + * @since 3.0 + */ +@ManagedResource +public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport { + + public static final long DEFAULT_RECEIVE_TIMEOUT = 1000; + + private final BoundListOperations boundListOperations; + + private MessageChannel errorChannel; + + private volatile TaskExecutor taskExecutor; + + private volatile RedisSerializer serializer = new JdkSerializationRedisSerializer(); + + private volatile boolean expectMessage = false; + + private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + + private volatile boolean active; + + private volatile boolean listening; + + /** + * @param queueName Must not be an empty String + * @param connectionFactory Must not be null + */ + public RedisQueueMessageDrivenEndpoint(String queueName, RedisConnectionFactory connectionFactory) { + Assert.hasText(queueName, "'queueName' is required"); + Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); + RedisTemplate template = new RedisTemplate(); + template.setConnectionFactory(connectionFactory); + template.setEnableDefaultSerializer(false); + template.setKeySerializer(new StringRedisSerializer()); + template.afterPropertiesSet(); + this.boundListOperations = template.boundListOps(queueName); + } + + public void setSerializer(RedisSerializer serializer) { + this.serializer = serializer; + } + + /** + * When data is retrieved from the Redis queue, does the returned data represent + * just the payload for a Message, or does the data represent a serialized + * {@link Message}?. {@code expectMessage} defaults to false. This means + * the retrieved data will be used as the payload for a new Spring Integration + * Message. Otherwise, the data is deserialized as Spring Integration + * Message. + * + * @param expectMessage Defaults to false + */ + public void setExpectMessage(boolean expectMessage) { + this.expectMessage = expectMessage; + } + + /** + * This timeout (milliseconds) is used when retrieving elements from the queue + * specified by {@link #boundListOperations}. + *

+ * If the queue does contain elements, the data is retrieved immediately. However, + * if the queue is empty, the Redis connection is blocked until either an element + * can be retrieved from the queue or until the specified timeout passes. + *

+ * A timeout of zero can be used to block indefinitely. If not set explicitly + * the timeout value will default to {@code 1000} + *

+ * See also: http://redis.io/commands/brpop + * + * @param receiveTimeout Must be non-negative. Specified in milliseconds. + */ + public void setReceiveTimeout(long receiveTimeout) { + Assert.isTrue(receiveTimeout > 0, "'receiveTimeout' must be > 0."); + this.receiveTimeout = receiveTimeout; + } + + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + @Override + public void setErrorChannel(MessageChannel errorChannel) { + super.setErrorChannel(errorChannel); + this.errorChannel = errorChannel; + } + + @Override + protected void onInit() { + super.onInit(); + if (this.expectMessage) { + Assert.notNull(this.serializer, "'serializer' has to be provided where 'expectMessage == true'."); + } + if (this.taskExecutor == null) { + String beanName = this.getComponentName(); + this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType()); + } + if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) { + MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler(); + errorHandler.setDefaultErrorChannel(this.errorChannel); + this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler); + } + } + + @Override + public String getComponentType() { + return "int-redis:message-driven-channel-adapter"; + } + + @SuppressWarnings("unchecked") + private void popMessageAndSend() { + Message message = null; + + byte[] value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); + + if (value != null) { + if (this.expectMessage) { + try { + message = (Message) this.serializer.deserialize(value); + } + catch (Exception e) { + throw new MessagingException("Deserialization of Message failed.", e); + } + } + else { + Object payload = value; + if (this.serializer != null) { + payload = this.serializer.deserialize(value); + } + message = MessageBuilder.withPayload(payload).build(); + } + } + + if (message != null) { + this.sendMessage(message); + } + } + + @Override + protected void doStart() { + if (!this.active) { + this.active = true; + this.restart(); + } + } + + private void restart() { + this.taskExecutor.execute(new ListenerTask()); + } + + @Override + protected void doStop() { + this.active = false; + } + + public boolean isListening() { + return listening; + } + + /** + * Returns the size of the Queue specified by {@link #boundListOperations}. The queue is + * represented by a Redis list. If the queue does not exist 0 + * is returned. See also http://redis.io/commands/llen + * + * @return Size of the queue. Never negative. + */ + @ManagedMetric + public long getQueueSize() { + return this.boundListOperations.size(); + } + + /** + * Clear the Redis Queue specified by {@link #boundListOperations}. + */ + @ManagedOperation + public void clearQueue() { + this.boundListOperations.getOperations().delete(this.boundListOperations.getKey()); + } + + + private class ListenerTask implements Runnable { + + @Override + public void run() { + RedisQueueMessageDrivenEndpoint.this.listening = true; + try { + while (RedisQueueMessageDrivenEndpoint.this.active) { + RedisQueueMessageDrivenEndpoint.this.popMessageAndSend(); + } + } + finally { + if (RedisQueueMessageDrivenEndpoint.this.active) { + RedisQueueMessageDrivenEndpoint.this.restart(); + } + else { + RedisQueueMessageDrivenEndpoint.this.listening = false; + } + } + } + + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java new file mode 100644 index 0000000000..bff2a74832 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java @@ -0,0 +1,112 @@ +/* + * Copyright 2013 the original author or authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.redis.outbound; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.Message; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; +import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @author Gunnar Hillert + * @author Artem Bilan + * @since 3.0 + */ +public class RedisQueueOutboundChannelAdapter extends AbstractMessageHandler implements IntegrationEvaluationContextAware { + + private final RedisSerializer stringSerializer = new StringRedisSerializer(); + + private final RedisTemplate template; + + private final Expression queueNameExpression; + + private EvaluationContext evaluationContext; + + private volatile boolean extractPayload = true; + + private volatile RedisSerializer serializer = new JdkSerializationRedisSerializer(); + + private volatile boolean serializerExplicitlySet; + + public RedisQueueOutboundChannelAdapter(String queueName, RedisConnectionFactory connectionFactory) { + this(new LiteralExpression(queueName), connectionFactory); + } + + public RedisQueueOutboundChannelAdapter(Expression queueNameExpression, RedisConnectionFactory connectionFactory) { + Assert.notNull(queueNameExpression, "'queueNameExpression' is required"); + Assert.hasText(queueNameExpression.getExpressionString(), "'queueNameExpression.getExpressionString()' is required"); + Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); + this.queueNameExpression = queueNameExpression; + this.template = new RedisTemplate(); + this.template.setConnectionFactory(connectionFactory); + this.template.setEnableDefaultSerializer(false); + this.template.setKeySerializer(new StringRedisSerializer()); + this.template.afterPropertiesSet(); + } + + @Override + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; + } + + public void setExtractPayload(boolean extractPayload) { + this.extractPayload = extractPayload; + } + + public void setSerializer(RedisSerializer serializer) { + Assert.notNull(serializer, "'serializer' must not be null"); + this.serializer = serializer; + this.serializerExplicitlySet = true; + } + + @Override + public String getComponentType() { + return "int-redis:outbound-channel-adapter"; + } + + @Override + @SuppressWarnings("unchecked") + protected void handleMessageInternal(Message message) throws Exception { + Object value = message; + + if (this.extractPayload) { + value = message.getPayload(); + } + + if (!(value instanceof byte[])) { + if (value instanceof String && !this.serializerExplicitlySet) { + value = this.stringSerializer.serialize((String) value); + } + else { + value = ((RedisSerializer) this.serializer).serialize(value); + } + } + + String queueName = this.queueNameExpression.getValue(this.evaluationContext, message, String.class); + this.template.boundListOps(queueName).leftPush(value); + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java new file mode 100644 index 0000000000..9184e0eff5 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -0,0 +1,157 @@ +/* + * Copyright 2013 the original author or authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.redis.inbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +import java.util.Date; + +import org.hamcrest.Matchers; +import org.junit.Test; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.support.MessageBuilder; + +/** + * @author Gunnar Hillert + * @author Artem Bilan + * @since 3.0 + */ +public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { + + @Test + @RedisAvailable + @SuppressWarnings("unchecked") + public void testInt3014Default() throws Exception { + + String queueName = "si.test.redisQueueInboundChannelAdapterTests"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setEnableDefaultSerializer(false); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); + + String payload = "testing"; + + redisTemplate.boundListOps(queueName).leftPush(payload); + + Date payload2 = new Date(); + + redisTemplate.boundListOps(queueName).leftPush(payload2); + + PollableChannel channel = new QueueChannel(); + + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, connectionFactory); + endpoint.setOutputChannel(channel); + endpoint.setReceiveTimeout(1000); + endpoint.afterPropertiesSet(); + endpoint.start(); + + Message receive = (Message) channel.receive(2000); + assertNotNull(receive); + assertEquals(payload, receive.getPayload()); + + receive = (Message) channel.receive(2000); + assertNotNull(receive); + assertEquals(payload2, receive.getPayload()); + + endpoint.stop(); + this.waitUntilListening(endpoint); + } + + @Test + @RedisAvailable + @SuppressWarnings("unchecked") + public void testInt3014ExpectMessageTrue() throws Exception { + + final String queueName = "si.test.redisQueueInboundChannelAdapterTests2"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setEnableDefaultSerializer(false); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); + + Message message = MessageBuilder.withPayload("testing").build(); + + redisTemplate.boundListOps(queueName).leftPush(message); + + redisTemplate.boundListOps(queueName).leftPush("test"); + + PollableChannel channel = new QueueChannel(); + + PollableChannel errorChannel = new QueueChannel(); + + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, connectionFactory); + endpoint.setExpectMessage(true); + endpoint.setOutputChannel(channel); + endpoint.setErrorChannel(errorChannel); + endpoint.setReceiveTimeout(1000); + endpoint.afterPropertiesSet(); + endpoint.start(); + + Message receive = (Message) channel.receive(2000); + assertNotNull(receive); + + assertEquals(message, receive); + + receive = (Message) errorChannel.receive(2000); + assertNotNull(receive); + assertThat(receive, Matchers.instanceOf(ErrorMessage.class)); + assertThat(receive.getPayload(), Matchers.instanceOf(MessagingException.class)); + assertThat(((Exception) receive.getPayload()).getMessage(), Matchers.containsString("Deserialization of Message failed.")); + assertThat(((Exception) receive.getPayload()).getCause(), Matchers.instanceOf(ClassCastException.class)); + assertThat(((Exception) receive.getPayload()).getCause().getMessage(), + Matchers.containsString("java.lang.String cannot be cast to org.springframework.integration.Message")); + + + endpoint.stop(); + this.waitUntilListening(endpoint); + } + + + public void waitUntilListening(RedisQueueMessageDrivenEndpoint endpoint) throws Exception { + int n = 0; + while (endpoint.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + throw new Exception("RedisQueueMessageDrivenEndpoint failed to stop."); + } + } + + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java new file mode 100644 index 0000000000..47a0c8c35c --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java @@ -0,0 +1,143 @@ +/* + * Copyright 2013 the original author or authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.redis.outbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; +import java.util.Date; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.serializer.JacksonJsonRedisSerializer; +import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.Message; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.support.MessageBuilder; + +/** + * @author Gunnar Hillert + * @author Artem Bilan + * @since 3.0 + */ +public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests { + + @Test + @RedisAvailable + public void testInt3015Default() throws Exception { + + final String queueName = "si.test.testRedisQueueOutboundChannelAdapter"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + + String payload = "testing"; + handler.handleMessage(MessageBuilder.withPayload(payload).build()); + + RedisTemplate redisTemplate = new StringRedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.afterPropertiesSet(); + + Object result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + + assertEquals(payload, result); + + Date payload2 = new Date(); + handler.handleMessage(MessageBuilder.withPayload(payload2).build()); + + RedisTemplate redisTemplate2 = new RedisTemplate(); + redisTemplate2.setConnectionFactory(connectionFactory); + redisTemplate2.setEnableDefaultSerializer(false); + redisTemplate2.setKeySerializer(new StringRedisSerializer()); + redisTemplate2.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate2.afterPropertiesSet(); + + Object result2 = redisTemplate2.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result2); + + assertEquals(payload2, result2); + } + + @Test + @RedisAvailable + public void testInt3015ExtractPayloadFalse() throws Exception { + + final String queueName = "si.test.testRedisQueueOutboundChannelAdapter2"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + handler.setExtractPayload(false); + + Message message = MessageBuilder.withPayload("testing").build(); + handler.handleMessage(message); + + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.setEnableDefaultSerializer(false); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); + + Object result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + + assertEquals(message, result); + + } + + @Test + @RedisAvailable + public void testInt3015ExplicitSerializer() throws Exception { + + final String queueName = "si.test.testRedisQueueOutboundChannelAdapter2"; + + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + + final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory); + handler.setSerializer(new JacksonJsonRedisSerializer(Object.class)); + + RedisTemplate redisTemplate = new StringRedisTemplate(); + redisTemplate.setConnectionFactory(connectionFactory); + redisTemplate.afterPropertiesSet(); + + handler.handleMessage(new GenericMessage(Arrays.asList("foo", "bar", "baz"))); + + Object result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + + assertEquals("[\"foo\",\"bar\",\"baz\"]", result); + + handler.handleMessage(new GenericMessage("test")); + + result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS); + assertNotNull(result); + + assertEquals("\"test\"", result); + } + +} diff --git a/spring-integration-redis/src/test/resources/log4j.properties b/spring-integration-redis/src/test/resources/log4j.properties new file mode 100644 index 0000000000..7ebfb6af9b --- /dev/null +++ b/spring-integration-redis/src/test/resources/log4j.properties @@ -0,0 +1,8 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n + +log4j.category.org.springframework.integration=WARN +log4j.category.org.springframework.integration.redis=DEBUG diff --git a/src/reference/docbook/ftp.xml b/src/reference/docbook/ftp.xml index 967934adc8..1e04a9960f 100644 --- a/src/reference/docbook/ftp.xml +++ b/src/reference/docbook/ftp.xml @@ -108,8 +108,9 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/ftp - DefaultFtpSessionFactory provides an abstraction over the underlying client API which, in the current release of - Spring Integration, is Apache Commons Net. This spares you from the low level configuration details + DefaultFtpSessionFactory provides an abstraction over the underlying client API which, + since Spring Integration 2.0, is Apache Commons Net. + This spares you from the low level configuration details of the org.apache.commons.net.ftp.FTPClient. However there are times when access to lower level FTPClient details is necessary to achieve more advanced configuration (e.g., setting data timeout, default timeout etc.). For that purpose, AbstractFtpSessionFactory (the base class for all FTP Session Factories) exposes hooks, in the form of the two post-processing methods below. diff --git a/src/reference/docbook/http.xml b/src/reference/docbook/http.xml index 21a162a18a..8ef55aeb5e 100644 --- a/src/reference/docbook/http.xml +++ b/src/reference/docbook/http.xml @@ -66,7 +66,7 @@ custom HttpMessageConverter to add the default converters after the custom converters. By default this flag is set to false, meaning that the custom converters replace the default list. - Starting with this release MultiPart File support was implemented. If the request has been wrapped as a + Starting with Spring Integration 2.0, MultiPart File support is implemented. If the request has been wrapped as a MultipartHttpServletRequest, when using the default converters, that request will be converted to a Message payload that is a MultiValueMap containing values that may be byte arrays, Strings, or instances of Spring's MultipartFile depending on the content type of the individual parts. @@ -619,7 +619,7 @@ By default the HTTP request will be generated using an instance of Si HttpComponentsClientHttpRequestFactory - - Uses Apache HttpComponents HttpClient (Since Spring 3.1) + - Uses Apache HttpComponents HttpClient (Since Spring 3.1) ClientHttpRequestFactory @@ -652,6 +652,13 @@ By default the HTTP request will be generated using an instance of Si + + When using the Apache HttpComponents HttpClient with a Pooling Connection Manager, be aware that, by + default, the connection manager will create no more than 2 concurrent connections per given route and no more than 20 connections + in total. For many real-world applications these limits may prove too constraining. Refer to the Apache documentation + (link above) for information about configuring this important component. + + Here is an example of how to configure an HTTP Outbound Gateway using a SimpleClientHttpRequestFactory, configured diff --git a/src/reference/docbook/jmx.xml b/src/reference/docbook/jmx.xml index b41b366eda..d39e8742d8 100644 --- a/src/reference/docbook/jmx.xml +++ b/src/reference/docbook/jmx.xml @@ -81,7 +81,7 @@ relatively simple. It only requires a JMX ObjectName in its configuration as shown below. - + default-notification-type attribute provided in the configuration. - + org.springframework.integration. - + @@ -250,7 +251,7 @@ Once the exporter is defined, start up your application with: - -Dcom.sun.management.jmxremote + -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=6969 -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false @@ -262,16 +263,20 @@ sophisticated features than JConsole.) - + + The MBean exporter is orthogonal to the one provided in Spring core - it registers message channels and message handlers, but not itself. You can expose the exporter itself, and certain other components in Spring Integration, using the standard <context:mbean-export/> - tag. The exporter has a couple of useful metrics attached to it, for + tag. The exporter has a some metrics attached to it, for instance a count of the number of active handlers and the number of - queued messages (these would both be important if you wanted to - shutdown the context without losing any messages). - + queued messages. + + + It also has a useful operation, as discussed in . + +
MBean ObjectNames @@ -361,6 +366,46 @@ + + + Custom elements can be appended to the object name by providing a reference to a + Properties object in the object-name-static-properties attribute. + + + Also, since Spring Integration 3.0, you can use a custom + ObjectNamingStrategy + using the object-naming-strategy attribute. This permits greater control over the naming of the + MBeans. For example, to group all Integration MBeans under + an 'Integration' type. A simple custom naming strategy implementation might be: + + + + The beanKey argument is a String containing the standard object name beginning with + the default-domain and including any additional static properties. + This example simply moves the standard type part to componentType and + sets the type to 'Integration', + enabling selection of all Integration MBeans in one query: + "my.domain:type=Integration,*. This also groups the beans under one tree entry under the + domain in tools like VisualVM. + + + The default naming strategy is a + MetadataNamingStrategy. The exporter propagates the default-domain to that object to allow it + to generate a fallback object name if parsing of the bean key fails. If your custom naming strategy is a + MetadataNamingStrategy (or subclass), the exporter will not + propagate the default-domain; you will need to configure it on your strategy bean. +
diff --git a/src/reference/docbook/shutdown.xml b/src/reference/docbook/shutdown.xml index 4ebeb1ecde..21c76edddb 100644 --- a/src/reference/docbook/shutdown.xml +++ b/src/reference/docbook/shutdown.xml @@ -58,4 +58,22 @@ If no time is left when we get to step 6, it probably means some thread is hung; in which case, the operation attempts a forced shutdown on all schedulers and executors before exiting. + + As discussed in this operation can be invoked using JMX. If you + wish to programmatically invoke the method, you will need to inject, or otherwise get a reference to, + the IntegrationMBeanExporter. If no id attribute is provided on + the <int-jmx:mbean-export/> definition, the bean will have a generated name. This + name contains a random component to avoid ObjectName collisions if multiple + Spring Integration contexts exist in the same JVM (MBeanServer). + + + For this reason, if you wish to invoke the method programmatically, it is recommended that you + provide the exporter with an id attribute so it can easily be accessed in the + application context. + + + Finally, the operation can be invoked using the <control-bus>; see the + monitoring Spring Integration sample application for details. +
diff --git a/src/reference/docbook/transactions.xml b/src/reference/docbook/transactions.xml index 5f8e656393..04f8368cd6 100644 --- a/src/reference/docbook/transactions.xml +++ b/src/reference/docbook/transactions.xml @@ -205,7 +205,7 @@ public interface TransactionSynchronizationProcessor { }]]> The factory is responsible for creating a - TransactionSynchronization + TransactionSynchronization object. You can implement your own, or use the one provided by the framework: DefaultTransactionSynchronizationFactory. This implementation returns a TransactionSynchronization that delegates to a default implementation of diff --git a/src/reference/docbook/transformer.xml b/src/reference/docbook/transformer.xml index 8a959e3794..ee1a6d87e4 100644 --- a/src/reference/docbook/transformer.xml +++ b/src/reference/docbook/transformer.xml @@ -206,7 +206,7 @@ public class Kid { a BeanCreationException will be thrown.  - JSON Transformers + JSON Transformers Object to JSON and JSON to Object transformers are provided. @@ -279,7 +279,7 @@ public class Foo { - Beginning with version 2.2, the object-to-json-transformer sets the content-type + Beginning with version 2.2, the object-to-json-transformer sets the content-type header to application/json, by default, if the input message does not already have that header present. @@ -290,66 +290,50 @@ public class Foo { attribute to an empty string (""). This will result in a message with no content-type header, unless such a header was present on the input message. - - The behavior of adding the default header has a side affect - causing applications with the following - sequence to fail: - - - ->object-to-json-transformer->amqp-outbound-adapter----> - - - ---->amqp-inbound-adapter->json-to-object-transformer-> - - - This is because the default SimpleMessageConverter used by the inbound adapter doesn't - recognize this content type and the adapter emits a message with a byte[] payload instead of - String, which was the case with earlier versions. - - - If you are using this pattern, there are a number of ways to configure the environment so that JSON - conversion will be performed correctly. - - - One solution is to set the content type to a text type, so the inbound converter will convert the JSON to - String. This solution requires a change to just the outbound application. - - - ]]> - - - The second solution is to eliminate the json transformers altogether and use an - org.springframework.amqp.support.converter.JsonMessageConverter on both the - outbound and inbound adapters. This configures the adapters to perform the JSON conversion and - the transformers are not necessary. The converter on the outbound adapter adds - type information to the message properties; the inbound converter uses this type information for the conversion. - The converter is provided to the adapters using the message-converter attribute. - This solution requires a change to both the inbound and outbound applications. - - - The third solution is to eliminate the json-to-object-transformer in just - the inbound application and use an - org.springframework.amqp.support.converter.JsonMessageConverter on the - inbound adapter. The converter - is provided to the adapter using the message-converter attribute. - However, because there will be no type information in the message properties, - this also requires adding the defaultType to the converter, using the - same type as currently configured on the json-to-object-transformer. - This solution requires a change to just the inbound application. The configuration below shows - how to configure the message converter; it requires spring-amqp 1.1.3 or - above. - - - - - - - - - -]]> + + Beginning with version 3.0, the ObjectToJsonTransformer adds headers, + reflecting the source type, to the message. Similarly, the JsonToObjectTransformer can + use those type headers when converting the JSON to an object. These headers are mapped in the AMQP adapters so that + they are entirely compatible with the Spring-AMQP + JsonMessageConverter. + + + This enables the following flows to work without any special configuration... + + + ...->amqp-outbound-adapter----> + + + ---->amqp-inbound-adapter->json-to-object-transformer->... + + + Where the outbound adapter is configured with a JsonMessageConverter and the + inbound adapter uses the default SimpleMessageConverter. + + + ...->object-to-json-transformer->amqp-outbound-adapter----> + + + ---->amqp-inbound-adapter->... + + + Where the outbound adapter is configured with a SimpleMessageConverter and the + inbound adapter uses the default JsonMessageConverter. + + + ...->object-to-json-transformer->amqp-outbound-adapter----> + + + ---->amqp-inbound-adapter->json-to-object-transformer-> + + + Where both adapters are configured with a SimpleMessageConverter. + + + When using the headers to determine the type, you should not provide + a class attribute, because it takes precedence over the headers. + In addition to JSON Transformers, Spring Integration provides a built-in #jsonPath SpEL function for use in expressions. For more information see . diff --git a/src/reference/docbook/twitter.xml b/src/reference/docbook/twitter.xml index 267f665777..40c4d7bb97 100644 --- a/src/reference/docbook/twitter.xml +++ b/src/reference/docbook/twitter.xml @@ -124,7 +124,7 @@ twitter.oauth.accessTokenSecret=AbRxUAvyNCtqQtxFK8w5ZMtMj20KFhB6o]]>twitter messages, or tweets - The current release of Spring Integration provides support for receiving tweets as Timeline Updates, + Spring Integration version 2.0 and above provides support for receiving tweets as Timeline Updates, Direct Messages, Mention Messages as well as Search Results. @@ -241,7 +241,7 @@ received. Twitter outbound channel adapters allow you to send Twitter Messages, or tweets. - The current release of Spring Integration supports sending Status Update Messages and Direct Messages. + Spring Integration version 2.0 and above supports sending Status Update Messages and Direct Messages. Twitter outbound channel adapters will take the Message payload and send it as a Twitter message. Currently the only supported payload type is String, so consider adding a transformer if the payload of the incoming message is not a String. diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 5cb22379b3..8bd71c43ee 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -63,12 +63,23 @@
JMX Support - A new <int-jmx:tree-polling-channel-adapter/> is provided; this - adapter queries the JMX MBean tree and sends a message with a payload that is the - graph of objects that matches the query. By default the MBeans are mapped to - primitives and simple Objects like Map, List and arrays - permitting simple - transformation, for example, to JSON - . + + + A new <int-jmx:tree-polling-channel-adapter/> is provided; this + adapter queries the JMX MBean tree and sends a message with a payload that is the + graph of objects that matches the query. By default the MBeans are mapped to + primitives and simple Objects like Map, List and arrays - permitting simple + transformation, for example, to JSON. + + + The IntegrationMBeanExporter now allows the configuration of + a custom ObjectNamingStrategy using the naming-strategy + attribute. + + + + + For more information, see .
@@ -314,10 +325,20 @@
Jackson Support (JSON) - A new abstraction for JSON conversion has been introduced. Implementations for Jackson 1.x - and Jackson 2 are currently provided, with the version being determined by presence on - the classpath. Previously, only Jackson 1.x was supported. For more information, - see 'JSON Transformers' in . + + + A new abstraction for JSON conversion has been introduced. Implementations for Jackson 1.x + and Jackson 2 are currently provided, with the version being determined by presence on + the classpath. Previously, only Jackson 1.x was supported. + + + The ObjectToJsonTransformer and JsonToObjectTransformer + now emit/consume headers containing type information. + + + + + For more information, see 'JSON Transformers' in .