Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Factored out common target adapter behavior from DefaultTargetAdapter into new AbstractTargetAdapter and implemented ApplicationEventTargetAdapter.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-03 18:41:04 +00:00
parent 5d9602d497
commit 4a3fc43856
13 changed files with 299 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java Normal file
View File

@@ -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<T> implements TargetAdapter<T> {
private String name;
private MessageChannel channel;
private MessageMapper<?,T> mapper = new SimplePayloadMessageMapper<T>();
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<?,T> mapper) {
Assert.notNull(mapper, "'mapper' must not be null");
this.mapper = mapper;
}
protected MessageMapper<?,T> 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);
}

57
spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java
View File

@@ -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<T> extends AbstractTargetAdapter<T> {
private String name;
private MessageChannel channel;
private Target target;
private MessageMapper mapper = new SimplePayloadMessageMapper();
private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(5);
private Target<T> target;
public DefaultTargetAdapter(Target target) {
public DefaultTargetAdapter(Target<T> 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);
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java
View File

@@ -25,7 +25,7 @@ import org.springframework.integration.message.MessageReceiver;
*
* @author Mark Fisher
*/
public interface TargetAdapter extends MessageReceiver {
public interface TargetAdapter<T> extends MessageReceiver<T> {
String getName();

51
spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapter.java Normal file
View File

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

37
spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java Normal file
View File

@@ -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<T> extends ApplicationEvent {
public MessagingEvent(Message<T> message) {
super(message);
}
public Message<T> getMessage() {
return (Message<T>) this.getSource();
}
}

38
spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEventMapper.java Normal file
View File

@@ -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<T> implements MessageMapper<T, MessagingEvent<T>> {
public MessagingEvent<T> fromMessage(Message<T> message) {
return new MessagingEvent<T>(message);
}
public Message<T> toMessage(MessagingEvent<T> event) {
return event.getMessage();
}
}

20
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
private Map<String, MessageEndpoint<?>> endpoints = new ConcurrentHashMap<String, MessageEndpoint<?>>();
private Map<String, TargetAdapter> targetAdapters = new ConcurrentHashMap<String, TargetAdapter>();
private Map<String, TargetAdapter<?>> targetAdapters = new ConcurrentHashMap<String, TargetAdapter<?>>();
private List<DispatcherTask> dispatcherTasks = new CopyOnWriteArrayList<DispatcherTask>();
private Map<MessageReceiver, MessageReceivingExecutor> receiverExecutors = new ConcurrentHashMap<MessageReceiver, MessageReceivingExecutor>();
private Map<MessageReceiver<?>, MessageReceivingExecutor> receiverExecutors = new ConcurrentHashMap<MessageReceiver<?>, 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);
}

4
spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java
View File

@@ -37,7 +37,7 @@ import org.springframework.integration.message.Message;
*
* @author Mark Fisher
*/
public class GenericMessageEndpoint implements MessageEndpoint {
public class GenericMessageEndpoint<T> implements MessageEndpoint<T> {
private String inputChannelName;
@@ -98,7 +98,7 @@ public class GenericMessageEndpoint implements MessageEndpoint {
}
public void messageReceived(Message<?> message) {
public void messageReceived(Message<T> message) {
if (this.handler == null) {
if (this.defaultOutputChannelName == null) {
throw new MessagingConfigurationException(

2
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -25,7 +25,7 @@ import org.springframework.integration.message.MessageReceiver;
*
* @author Mark Fisher
*/
public interface MessageEndpoint extends MessageReceiver {
public interface MessageEndpoint<T> extends MessageReceiver<T> {
void setInputChannelName(String inputChannelName);

4
spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java
View File

@@ -22,8 +22,8 @@ package org.springframework.integration.message;
*
* @author Mark Fisher
*/
public interface MessageReceiver {
public interface MessageReceiver<T> {
void messageReceived(Message<?> message);
void messageReceived(Message<T> message);
}