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

Added support for 'apply-sequence' on <publish-subscribe-channel/>. The BroadcastingDispatcher has the 'applySequence' property (INT-321).

Browse Source
This commit is contained in:
Mark Fisher
2008-07-30 16:02:47 +00:00
parent 59a4696d81
commit b9bb70b634
8 changed files with 193 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
org.springframework.integration/src/main/java/org/springframework/integration/config/PublishSubscribeChannelParser.java
View File

@@ -41,6 +41,7 @@ public class PublishSubscribeChannelParser extends AbstractChannelParser {
if (StringUtils.hasText(taskExecutorRef)) {
builder.addConstructorArgReference(taskExecutorRef);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
}
}

1
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -73,6 +73,7 @@
<xsd:complexContent>
<xsd:extension base="channelType">
<xsd:attribute name="task-executor" type="xsd:string"/>
<xsd:attribute name="apply-sequence" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

26
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.dispatcher;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageTarget;
/**
@@ -30,18 +31,37 @@ import org.springframework.integration.message.MessageTarget;
*/
public class BroadcastingDispatcher extends AbstractDispatcher {
public boolean send(final Message<?> message) {
private volatile boolean applySequence;
/**
* Specify whether to apply sequence numbers to the messages
* prior to sending to the targets. By default, sequence
* numbers will <em>not</em> be applied
*/
public void setApplySequence(boolean applySequence) {
this.applySequence = applySequence;
}
public boolean send(Message<?> message) {
int sequenceNumber = 1;
int sequenceSize = this.targets.size();
for (final MessageTarget target : this.targets) {
final Message<?> messageToSend = (!this.applySequence) ? message
: MessageBuilder.fromMessage(message)
.setSequenceNumber(sequenceNumber++)
.setSequenceSize(sequenceSize)
.build();
TaskExecutor executor = this.getTaskExecutor();
if (executor != null) {
executor.execute(new Runnable() {
public void run() {
sendMessageToTarget(message, target);
sendMessageToTarget(messageToSend, target);
}
});
}
else {
this.sendMessageToTarget(message, target);
this.sendMessageToTarget(messageToSend, target);
}
}
return true;

10
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PublishSubscribeChannel.java
View File

@@ -33,9 +33,6 @@ public class PublishSubscribeChannel extends AbstractMessageChannel implements S
private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher();
public PublishSubscribeChannel() {
}
/**
* Create a PublishSubscribeChannel that will use a {@link TaskExecutor}
* to publish its Messages.
@@ -46,6 +43,13 @@ public class PublishSubscribeChannel extends AbstractMessageChannel implements S
}
}
public PublishSubscribeChannel() {
}
public void setApplySequence(boolean applySequence) {
this.dispatcher.setApplySequence(applySequence);
}
public boolean subscribe(MessageTarget target) {
return this.dispatcher.addTarget(target);

79
org.springframework.integration/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java Normal file
View File

@@ -0,0 +1,79 @@
/*
* Copyright 2002-2008 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.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.PublishSubscribeChannel;
/**
* @author Mark Fisher
*/
public class PublishSubscribeChannelParserTests {
@Test
public void defaultChannel() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("defaultChannel");
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
assertNull(dispatcherAccessor.getPropertyValue("taskExecutor"));
assertFalse((Boolean) dispatcherAccessor.getPropertyValue("applySequence"));
}
@Test
public void applySequenceEnabled() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("channelWithApplySequenceEnabled");
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
assertTrue((Boolean) new DirectFieldAccessor(dispatcher).getPropertyValue("applySequence"));
}
@Test
public void channelWithTaskExecutor() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("channelWithTaskExecutor");
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
TaskExecutor executor = (TaskExecutor) dispatcherAccessor.getPropertyValue("taskExecutor");
assertNotNull(executor);
assertEquals(context.getBean("pool"), executor);
}
}

18
org.springframework.integration/src/test/java/org/springframework/integration/config/publishSubscribeChannelParserTests.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<publish-subscribe-channel id="defaultChannel"/>
<publish-subscribe-channel id="channelWithApplySequenceEnabled" apply-sequence="true"/>
<publish-subscribe-channel id="channelWithTaskExecutor" task-executor="pool"/>
<pool-executor id="pool"/>
</beans:beans>

62
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java
View File

@@ -16,18 +16,31 @@
package org.springframework.integration.dispatcher;
import static org.junit.Assert.*;
import static org.easymock.EasyMock.*;
import static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.getCurrentArguments;
import static org.easymock.EasyMock.isA;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.reset;
import static org.easymock.EasyMock.verify;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.easymock.IAnswer;
import org.junit.Before;
import org.junit.Test;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.StringMessage;
/**
* @author Mark Fisher
@@ -137,6 +150,51 @@ public class BroadcastingDispatcherTests {
assertTrue("Timing out Runnable not executed", timingOutStarted.get());
}
@Test
public void applySequenceDisabledByDefault() {
BroadcastingDispatcher dispatcher = new BroadcastingDispatcher();
final List<Message<?>> messages = new ArrayList<Message<?>>();
MessageTarget target = new MessageTarget() {
public boolean send(Message<?> message) {
messages.add(message);
return true;
}
};
dispatcher.addTarget(target);
dispatcher.addTarget(target);
dispatcher.send(new StringMessage("test"));
assertEquals(2, messages.size());
assertEquals(0, (int) messages.get(0).getHeaders().getSequenceNumber());
assertEquals(0, (int) messages.get(0).getHeaders().getSequenceSize());
assertEquals(0, (int) messages.get(1).getHeaders().getSequenceNumber());
assertEquals(0, (int) messages.get(1).getHeaders().getSequenceSize());
}
@Test
public void applySequenceEnabled() {
BroadcastingDispatcher dispatcher = new BroadcastingDispatcher();
dispatcher.setApplySequence(true);
final List<Message<?>> messages = new ArrayList<Message<?>>();
MessageTarget target = new MessageTarget() {
public boolean send(Message<?> message) {
messages.add(message);
return true;
}
};
dispatcher.addTarget(target);
dispatcher.addTarget(target);
dispatcher.addTarget(target);
dispatcher.send(new StringMessage("test"));
assertEquals(3, messages.size());
assertEquals(1, (int) messages.get(0).getHeaders().getSequenceNumber());
assertEquals(3, (int) messages.get(0).getHeaders().getSequenceSize());
assertEquals(2, (int) messages.get(1).getHeaders().getSequenceNumber());
assertEquals(3, (int) messages.get(1).getHeaders().getSequenceSize());
assertEquals(3, (int) messages.get(2).getHeaders().getSequenceNumber());
assertEquals(3, (int) messages.get(2).getHeaders().getSequenceSize());
}
private void defaultTaskExecutorMock() {
taskExecutorMock.execute(isA(Runnable.class));
expectLastCall().andAnswer(new IAnswer<Object>() {