diff --git a/build.gradle b/build.gradle index b716a2e451..2f44209983 100644 --- a/build.gradle +++ b/build.gradle @@ -77,6 +77,7 @@ subprojects { subproject -> aspectjVersion = '1.8.2' apacheSshdVersion = '0.10.1' boonVersion = '0.25' + chronicleVersion = '3.2.2' commonsDbcpVersion = '1.4' commonsIoVersion = '2.4' commonsNetVersion = '3.3' @@ -256,6 +257,7 @@ project('spring-integration-core') { compile("io.fastjson:boon:$boonVersion", optional) testCompile ("org.aspectj:aspectjweaver:$aspectjVersion") + testCompile ("net.openhft:chronicle:$chronicleVersion") } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java index 32c9514fde..c6c7dcf910 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java @@ -18,8 +18,10 @@ package org.springframework.integration.channel; import java.util.ArrayList; import java.util.List; +import java.util.Queue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import org.springframework.integration.core.MessageSelector; @@ -36,17 +38,20 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ public class QueueChannel extends AbstractPollableChannel implements QueueChannelOperations { - private final BlockingQueue> queue; + private final Queue> queue; + + protected final Semaphore queueSemaphore = new Semaphore(0); /** * Create a channel with the specified queue. * * @param queue The queue. */ - public QueueChannel(BlockingQueue> queue) { + public QueueChannel(Queue> queue) { Assert.notNull(queue, "'queue' must not be null"); this.queue = queue; } @@ -76,14 +81,25 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne protected boolean doSend(Message message, long timeout) { Assert.notNull(message, "'message' must not be null"); try { - if (timeout > 0) { - return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS); + if (this.queue instanceof BlockingQueue) { + BlockingQueue> blockingQueue = (BlockingQueue>) this.queue; + if (timeout > 0) { + return blockingQueue.offer(message, timeout, TimeUnit.MILLISECONDS); + } + if (timeout == 0) { + return blockingQueue.offer(message); + } + blockingQueue.put(message); + return true; } - if (timeout == 0) { - return this.queue.offer(message); + else { + try { + return this.queue.offer(message); + } + finally { + this.queueSemaphore.release(); + } } - queue.put(message); - return true; } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -95,12 +111,32 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne protected Message doReceive(long timeout) { try { if (timeout > 0) { - return queue.poll(timeout, TimeUnit.MILLISECONDS); + if (this.queue instanceof BlockingQueue) { + return ((BlockingQueue>) this.queue).poll(timeout, TimeUnit.MILLISECONDS); + } + else { + long nanos = TimeUnit.MILLISECONDS.toNanos(timeout); + long deadline = System.nanoTime() + nanos; + while (this.queue.size() == 0 && nanos > 0) { + this.queueSemaphore.tryAcquire(nanos, TimeUnit.NANOSECONDS); + nanos = deadline - System.nanoTime(); + } + return this.queue.poll(); + } } if (timeout == 0) { - return queue.poll(); + return this.queue.poll(); + } + + if (this.queue instanceof BlockingQueue) { + return ((BlockingQueue>) this.queue).take(); + } + else { + while (this.queue.size() == 0) { + this.queueSemaphore.tryAcquire(50, TimeUnit.MILLISECONDS); + } + return this.queue.poll(); } - return queue.take(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -111,7 +147,15 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne @Override public List> clear() { List> clearedMessages = new ArrayList>(); - this.queue.drainTo(clearedMessages); + if (this.queue instanceof BlockingQueue) { + ((BlockingQueue>) this.queue).drainTo(clearedMessages); + } + else { + Message message = null; + while ((message = this.queue.poll()) != null) { + clearedMessages.add(message); + } + } return clearedMessages; } @@ -138,7 +182,13 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne @Override public int getRemainingCapacity() { - return this.queue.remainingCapacity(); + if (this.queue instanceof BlockingQueue) { + return ((BlockingQueue>) this.queue).remainingCapacity(); + } + else { + //Assume that underlying Queue implementation takes care of its size on "offer". + return Integer.MAX_VALUE; + } } } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index 82401f1247..f21f7aa3b1 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -241,13 +241,12 @@ - Reference to a BlockingQueue that can be used to buffer the messages. This attribute is - mutually - exclusive with the "message-store" attribute (only one can be specified). + Reference to a Queue that can be used to buffer the messages. This attribute is + mutually exclusive with the "message-store" attribute (only one can be specified). - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java index e150447929..4bad9f8005 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2014 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. @@ -28,15 +28,22 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; -import org.springframework.messaging.Message; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.selector.UnexpiredMessageSelector; import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +import reactor.io.encoding.JavaSerializationCodec; +import reactor.queue.PersistentQueue; +import reactor.queue.spec.PersistentQueueSpec; /** * @author Mark Fisher + * @author Artem Bilan */ public class QueueChannelTests { @@ -237,4 +244,113 @@ public class QueueChannelTests { assertTrue(channel.send(new GenericMessage("roomAvailable"), 0)); } + @Rule + public final TemporaryFolder tempFolder = new TemporaryFolder(); + + @Test + public void testReactorPersistentQueue() throws InterruptedException { + final AtomicBoolean messageReceived = new AtomicBoolean(false); + final CountDownLatch latch = new CountDownLatch(1); + PersistentQueue> queue = new PersistentQueueSpec>() + .codec(new JavaSerializationCodec>()) + .basePath(this.tempFolder.getRoot().getAbsolutePath()) + .get(); + final QueueChannel channel = new QueueChannel(queue); + new Thread(new Runnable() { + public void run() { + Message message = channel.receive(); + if (message != null) { + messageReceived.set(true); + latch.countDown(); + } + } + }).start(); + assertFalse(messageReceived.get()); + channel.send(new GenericMessage("testing")); + latch.await(1000, TimeUnit.MILLISECONDS); + assertTrue(messageReceived.get()); + + final CountDownLatch latch1 = new CountDownLatch(2); + + Thread thread = new Thread(new Runnable() { + public void run() { + while (true) { + Message message = channel.receive(100); + if (message != null) { + latch1.countDown(); + if (latch1.getCount() == 0) { + break; + } + } + } + } + }); + thread.start(); + + Thread.sleep(200); + channel.send(new GenericMessage("testing")); + channel.send(new GenericMessage("testing")); + assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS)); + + final AtomicBoolean receiveInterrupted = new AtomicBoolean(false); + final CountDownLatch latch2 = new CountDownLatch(1); + Thread t = new Thread(new Runnable() { + public void run() { + Message message = channel.receive(10000); + receiveInterrupted.set(true); + assertTrue(message == null); + latch2.countDown(); + } + }); + t.start(); + assertFalse(receiveInterrupted.get()); + t.interrupt(); + latch2.await(); + assertTrue(receiveInterrupted.get()); + + receiveInterrupted.set(false); + final CountDownLatch latch3 = new CountDownLatch(1); + t = new Thread(new Runnable() { + public void run() { + Message message = channel.receive(); + receiveInterrupted.set(true); + assertTrue(message == null); + latch3.countDown(); + } + }); + t.start(); + assertFalse(receiveInterrupted.get()); + t.interrupt(); + latch3.await(); + assertTrue(receiveInterrupted.get()); + + GenericMessage message1 = new GenericMessage("test1"); + GenericMessage message2 = new GenericMessage("test2"); + assertTrue(channel.send(message1)); + assertTrue(channel.send(message2)); + List> clearedMessages = channel.clear(); + assertNotNull(clearedMessages); + assertEquals(2, clearedMessages.size()); + + clearedMessages = channel.clear(); + assertNotNull(clearedMessages); + assertEquals(0, clearedMessages.size()); + + // Test on artificial infinite wait + // channel.receive(); + + // Distributed scenario + final CountDownLatch latch4 = new CountDownLatch(1); + new Thread(new Runnable() { + public void run() { + Message message = channel.receive(); + if (message != null) { + latch4.countDown(); + } + } + }).start(); + queue.add(new GenericMessage("foo")); + assertTrue(latch4.await(1000, TimeUnit.MILLISECONDS)); + } + } diff --git a/src/reference/docbook/channel.xml b/src/reference/docbook/channel.xml index bc3dc14671..082ec98df2 100644 --- a/src/reference/docbook/channel.xml +++ b/src/reference/docbook/channel.xml @@ -557,9 +557,9 @@ payload to an Integer. is polled from a QueueChannel, it is removed from the Message Store. - By default any QueueChannel only stores its Messages in an in-memory Queue + By default, a QueueChannel stores its Messages in an in-memory Queue and can therefore lead to the lost message scenario mentioned above. However Spring Integration - provides a JdbcMessageStore to allow a QueueChannel to be backed by an RDBMS. + provides persistent stores, such as the JdbcMessageStore. You can configure a Message Store for any QueueChannel by adding the @@ -593,6 +593,21 @@ payload to an Integer. is a ChannelPriorityMessageStore the messages will be received in FIFO within priority order. The notion of priority is determined by the message store implementation. + + Another option to customize the QueueChannel environment is provided by the ref attribute of the + <int:queue> sub-element. This attribute implies the reference to any + java.util.Queue implementation. An implementation is provided + by the Project Reactor and its + reactor.queue.PersistentQueue implementation for the + IndexedChronicle: + + >() + .codec(new JavaSerializationCodec>()) + .basePath(System.getProperty("java.io.tmpdir") + "/reactor-queue") + .get()); +}]]>
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index fca87c7418..e947fc4eea 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -237,5 +237,16 @@ See .
+
+ QueueChannel: backed Queue type + + The QueueChannel backed Queue type has been changed + from BlockingQueue to the more generic + Queue. It allows the use of any external + Queue implementation, for example Reactor's + PersistentQueue. + See . + +