From 4a3fc43856ce19161dc7b5f34e91f1f530917eed Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 3 Jan 2008 18:41:04 +0000 Subject: [PATCH] Factored out common target adapter behavior from DefaultTargetAdapter into new AbstractTargetAdapter and implemented ApplicationEventTargetAdapter. --- .../adapter/AbstractTargetAdapter.java | 83 +++++++++++++++++++ .../adapter/DefaultTargetAdapter.java | 57 ++----------- .../integration/adapter/TargetAdapter.java | 2 +- .../event/ApplicationEventTargetAdapter.java | 51 ++++++++++++ .../adapter/event/MessagingEvent.java | 37 +++++++++ .../adapter/event/MessagingEventMapper.java | 38 +++++++++ .../integration/bus/MessageBus.java | 20 ++--- .../endpoint/GenericMessageEndpoint.java | 4 +- .../integration/endpoint/MessageEndpoint.java | 2 +- .../integration/message/MessageReceiver.java | 4 +- .../ApplicationEventTargetAdapterTests.java | 61 ++++++++++++++ .../bus/FixedRateConsumerTests.java | 8 +- .../bus/UnicastMessageDispatcherTests.java | 8 +- 13 files changed, 299 insertions(+), 76 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapter.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEventMapper.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java new file mode 100644 index 0000000000..105684fdca --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java @@ -0,0 +1,83 @@ +/* + * Copyright 2002-2007 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.adapter; + +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageMapper; +import org.springframework.integration.message.SimplePayloadMessageMapper; +import org.springframework.util.Assert; + +/** + * Base class providing common behavior for target adapters. + * + * @author Mark Fisher + */ +public abstract class AbstractTargetAdapter implements TargetAdapter { + + private String name; + + private MessageChannel channel; + + private MessageMapper mapper = new SimplePayloadMessageMapper(); + + private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(5); + + + public void setName(String name) { + this.name = name; + } + + public String getName() { + return this.name; + } + + public void setChannel(MessageChannel channel) { + Assert.notNull(channel, "'channel' must not be null"); + this.channel = channel; + } + + public MessageChannel getChannel() { + return this.channel; + } + + public void setMessageMapper(MessageMapper mapper) { + Assert.notNull(mapper, "'mapper' must not be null"); + this.mapper = mapper; + } + + protected MessageMapper getMessageMapper() { + return this.mapper; + } + + public void setConsumerPolicy(ConsumerPolicy policy) { + Assert.notNull(policy, "'policy' must not be null"); + this.policy = policy; + } + + public ConsumerPolicy getConsumerPolicy() { + return this.policy; + } + + public final void messageReceived(Message message) { + this.sendToTarget(this.mapper.fromMessage(message)); + } + + protected abstract boolean sendToTarget(T object); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java index 1934bba092..3a487245b3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java @@ -16,11 +16,7 @@ package org.springframework.integration.adapter; -import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageMapper; -import org.springframework.integration.message.SimplePayloadMessageMapper; import org.springframework.util.Assert; /** @@ -29,61 +25,18 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class DefaultTargetAdapter implements TargetAdapter { +public class DefaultTargetAdapter extends AbstractTargetAdapter { - private String name; - - private MessageChannel channel; - - private Target target; - - private MessageMapper mapper = new SimplePayloadMessageMapper(); - - private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(5); + private Target target; - public DefaultTargetAdapter(Target target) { + public DefaultTargetAdapter(Target target) { Assert.notNull(target, "'target' must not be null"); this.target = target; } - public void setName(String name) { - this.name = name; - } - - public String getName() { - return this.name; - } - - public void setChannel(MessageChannel channel) { - Assert.notNull(channel, "'channel' must not be null"); - this.channel = channel; - } - - public MessageChannel getChannel() { - return this.channel; - } - - public void setMessageMapper(MessageMapper mapper) { - Assert.notNull(mapper, "'mapper' must not be null"); - this.mapper = mapper; - } - - protected MessageMapper getMessageMapper() { - return this.mapper; - } - - public void setConsumerPolicy(ConsumerPolicy policy) { - Assert.notNull(policy, "'policy' must not be null"); - this.policy = policy; - } - - public ConsumerPolicy getConsumerPolicy() { - return this.policy; - } - - public void messageReceived(Message message) { - this.target.send(this.mapper.fromMessage(message)); + public boolean sendToTarget(T object) { + return this.target.send(object); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java index b2380eca08..f9e3ffdab9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java @@ -25,7 +25,7 @@ import org.springframework.integration.message.MessageReceiver; * * @author Mark Fisher */ -public interface TargetAdapter extends MessageReceiver { +public interface TargetAdapter extends MessageReceiver { String getName(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapter.java new file mode 100644 index 0000000000..ba0c846bb8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapter.java @@ -0,0 +1,51 @@ +/* + * Copyright 2002-2007 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.adapter.event; + +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.integration.adapter.AbstractTargetAdapter; + +/** + * A target adapter for publishing {@link MessagingEvent MessagingEvents}. The + * {@link MessagingEvent} is a subclass of Spring's {@link ApplicationEvent} + * used by this adapter to wrap any {@link Message} received on its channel. + * + * @author Mark Fisher + */ +public class ApplicationEventTargetAdapter extends AbstractTargetAdapter implements + ApplicationEventPublisherAware { + + private ApplicationEventPublisher applicationEventPublisher; + + + public ApplicationEventTargetAdapter() { + this.setMessageMapper(new MessagingEventMapper()); + } + + + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + @Override + protected boolean sendToTarget(MessagingEvent event) { + this.applicationEventPublisher.publishEvent(event); + return true; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java new file mode 100644 index 0000000000..f83d22ca51 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java @@ -0,0 +1,37 @@ +/* + * Copyright 2002-2007 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.adapter.event; + +import org.springframework.context.ApplicationEvent; +import org.springframework.integration.message.Message; + +/** + * A subclass of {@link ApplicationEvent} that wraps a {@link Message}. + * + * @author Mark Fisher + */ +public class MessagingEvent extends ApplicationEvent { + + public MessagingEvent(Message message) { + super(message); + } + + public Message getMessage() { + return (Message) this.getSource(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEventMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEventMapper.java new file mode 100644 index 0000000000..bb8ebf539c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEventMapper.java @@ -0,0 +1,38 @@ +/* + * Copyright 2002-2007 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.adapter.event; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageMapper; + +/** + * A {@link MessageMapper} implementation for mapping to and from + * {@link MessagingEvent MessagingEvents}. + * + * @author Mark Fisher + */ +public class MessagingEventMapper implements MessageMapper> { + + public MessagingEvent fromMessage(Message message) { + return new MessagingEvent(message); + } + + public Message toMessage(MessagingEvent event) { + return event.getMessage(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index ef069b4ddf..68ec43bb56 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -30,7 +30,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.integration.MessagingException; -import org.springframework.integration.adapter.DefaultTargetAdapter; +import org.springframework.integration.adapter.AbstractTargetAdapter; import org.springframework.integration.adapter.SourceAdapter; import org.springframework.integration.adapter.TargetAdapter; import org.springframework.integration.channel.ChannelRegistry; @@ -54,13 +54,13 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - private Map endpoints = new ConcurrentHashMap(); + private Map> endpoints = new ConcurrentHashMap>(); - private Map targetAdapters = new ConcurrentHashMap(); + private Map> targetAdapters = new ConcurrentHashMap>(); private List dispatcherTasks = new CopyOnWriteArrayList(); - private Map receiverExecutors = new ConcurrentHashMap(); + private Map, MessageReceivingExecutor> receiverExecutors = new ConcurrentHashMap, MessageReceivingExecutor>(); private ScheduledThreadPoolExecutor dispatcherExecutor; @@ -153,7 +153,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif this.channelRegistry.registerChannel(name, channel); } - public void registerEndpoint(String name, MessageEndpoint endpoint) { + public void registerEndpoint(String name, MessageEndpoint endpoint) { Assert.notNull(name, "'name' must not be null"); Assert.notNull(endpoint, "'endpoint' must not be null"); this.endpoints.put(name, endpoint); @@ -189,9 +189,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - public void registerTargetAdapter(String name, TargetAdapter targetAdapter) { - if (targetAdapter instanceof DefaultTargetAdapter) { - DefaultTargetAdapter adapter = (DefaultTargetAdapter) targetAdapter; + public void registerTargetAdapter(String name, TargetAdapter targetAdapter) { + if (targetAdapter instanceof AbstractTargetAdapter) { + AbstractTargetAdapter adapter = (AbstractTargetAdapter) targetAdapter; adapter.setName(name); this.targetAdapters.put(name, targetAdapter); MessageChannel channel = adapter.getChannel(); @@ -222,7 +222,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif channel = new PointToPointChannel(); this.registerChannel(channelName, channel); } - MessageEndpoint endpoint = this.endpoints.get(endpointName); + MessageEndpoint endpoint = this.endpoints.get(endpointName); if (endpoint == null) { throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'"); } @@ -257,7 +257,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public int getActiveCountForReceiver(String receiverName) { - MessageReceiver receiver = this.endpoints.get(receiverName); + MessageReceiver receiver = this.endpoints.get(receiverName); if (receiver == null) { receiver = this.targetAdapters.get(receiverName); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java index 9d9a07e913..950ba69f59 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java @@ -37,7 +37,7 @@ import org.springframework.integration.message.Message; * * @author Mark Fisher */ -public class GenericMessageEndpoint implements MessageEndpoint { +public class GenericMessageEndpoint implements MessageEndpoint { private String inputChannelName; @@ -98,7 +98,7 @@ public class GenericMessageEndpoint implements MessageEndpoint { } - public void messageReceived(Message message) { + public void messageReceived(Message message) { if (this.handler == null) { if (this.defaultOutputChannelName == null) { throw new MessagingConfigurationException( diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index c326a72713..a9d0dc161d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -25,7 +25,7 @@ import org.springframework.integration.message.MessageReceiver; * * @author Mark Fisher */ -public interface MessageEndpoint extends MessageReceiver { +public interface MessageEndpoint extends MessageReceiver { void setInputChannelName(String inputChannelName); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java index d58ec6546e..931e9ed253 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java @@ -22,8 +22,8 @@ package org.springframework.integration.message; * * @author Mark Fisher */ -public interface MessageReceiver { +public interface MessageReceiver { - void messageReceived(Message message); + void messageReceived(Message message); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java new file mode 100644 index 0000000000..89b2d24126 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java @@ -0,0 +1,61 @@ +/* + * Copyright 2002-2007 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.adapter.event; + +import static org.junit.Assert.assertEquals; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; + +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class ApplicationEventTargetAdapterTests { + + @Test + public void testSendingEvent() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + ApplicationEventPublisher publisher = new ApplicationEventPublisher() { + public void publishEvent(ApplicationEvent event) { + latch.countDown(); + } + }; + MessageChannel channel = new PointToPointChannel(); + ApplicationEventTargetAdapter adapter = new ApplicationEventTargetAdapter(); + adapter.setApplicationEventPublisher(publisher); + adapter.setChannel(channel); + MessageBus bus = new MessageBus(); + bus.registerChannel("channel", channel); + bus.registerTargetAdapter("adapter", adapter); + bus.start(); + assertEquals(1, latch.getCount()); + channel.send(new StringMessage("123", "testing")); + latch.await(100, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + bus.stop(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java index b14b071365..2a347a5d0c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java @@ -42,8 +42,8 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageEndpoint endpoint = new GenericMessageEndpoint() { - public void messageReceived(Message message) { + MessageEndpoint endpoint = new GenericMessageEndpoint() { + public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); } @@ -73,8 +73,8 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); PointToPointChannel channel = new PointToPointChannel(); - MessageEndpoint endpoint = new GenericMessageEndpoint() { - public void messageReceived(Message message) { + MessageEndpoint endpoint = new GenericMessageEndpoint() { + public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java index 0900e4d01e..1a3d19ca05 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java @@ -40,16 +40,16 @@ public class UnicastMessageDispatcherTests { final AtomicBoolean endpoint1Received = new AtomicBoolean(); final AtomicBoolean endpoint2Received = new AtomicBoolean(); final CountDownLatch latch = new CountDownLatch(1); - MessageEndpoint endpoint1 = new GenericMessageEndpoint() { + MessageEndpoint endpoint1 = new GenericMessageEndpoint() { @Override - public void messageReceived(Message message) { + public void messageReceived(Message message) { endpoint1Received.set(true); latch.countDown(); } }; - MessageEndpoint endpoint2 = new GenericMessageEndpoint() { + MessageEndpoint endpoint2 = new GenericMessageEndpoint() { @Override - public void messageReceived(Message message) { + public void messageReceived(Message message) { endpoint2Received.set(true); latch.countDown(); }