INT-3520: Add Reactor's PersistentQueue Support

JIRA: https://jira.spring.io/browse/INT-3520

INT-3520: Use `Condition` to wait items in the `Queue`

INT-3520: Reworking to the `Semaphore(0)`

* Implement 'artificial infinite wait'
* Add distributed test with infinite `receive()`

INT-3520: Correct usage of `Semaphore`

Minor Doc Polishing

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")
 	}
 }
 

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<Message<?>> queue;
+	private final Queue<Message<?>> queue;
+
+	protected final Semaphore queueSemaphore = new Semaphore(0);
 
 	/**
 	 * Create a channel with the specified queue.
 	 *
 	 * @param queue The queue.
 	 */
-	public QueueChannel(BlockingQueue<Message<?>> queue) {
+	public QueueChannel(Queue<Message<?>> 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<Message<?>> blockingQueue = (BlockingQueue<Message<?>>) 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<Message<?>>) 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<Message<?>>) 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<Message<?>> clear() {
 		List<Message<?>> clearedMessages = new ArrayList<Message<?>>();
-		this.queue.drainTo(clearedMessages);
+		if (this.queue instanceof BlockingQueue) {
+			((BlockingQueue<Message<?>>) 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<Message<?>>) this.queue).remainingCapacity();
+		}
+		else {
+			//Assume that underlying Queue implementation takes care of its size on "offer".
+			return Integer.MAX_VALUE;
+		}
 	}
 
 }

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd

@@ -241,13 +241,12 @@
 		<xsd:attribute name="ref" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
-					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).
 				</xsd:documentation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
-						<tool:expected-type type="java.util.concurrent.BlockingQueue" />
+						<tool:expected-type type="java.util.Queue" />
 					</tool:annotation>
 				</xsd:appinfo>
 			</xsd:annotation>

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<String>("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<Message<?>> queue = new PersistentQueueSpec<Message<?>>()
+				.codec(new JavaSerializationCodec<Message<?>>())
+				.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<String>("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<String>("testing"));
+		channel.send(new GenericMessage<String>("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<String> message1 = new GenericMessage<String>("test1");
+		GenericMessage<String> message2 = new GenericMessage<String>("test2");
+		assertTrue(channel.send(message1));
+		assertTrue(channel.send(message2));
+		List<Message<?>> 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<String>("foo"));
+		assertTrue(latch4.await(1000, TimeUnit.MILLISECONDS));
+	}
+
 }

src/reference/docbook/channel.xml

@@ -557,9 +557,9 @@ payload to an Integer.
 	  is polled from a <classname>QueueChannel</classname>, it is removed from the Message Store.
       </para>
       <para>
-      By default any <classname>QueueChannel</classname> only stores its Messages in an in-memory Queue
+      By default, a <classname>QueueChannel</classname> stores its Messages in an in-memory Queue
       and can therefore lead to the lost message scenario mentioned above. However Spring Integration
-      provides a <classname>JdbcMessageStore</classname> to allow a <classname>QueueChannel</classname> to be backed by an RDBMS.
+      provides persistent stores, such as the <classname>JdbcMessageStore</classname>.
       </para>
       <para>
       You can configure a Message Store for any <classname>QueueChannel</classname> by adding the
@@ -593,6 +593,21 @@ payload to an Integer.
         is a <interfacename>ChannelPriorityMessageStore</interfacename> the messages will be received in
         FIFO within priority order. The notion of priority is determined by the message store implementation.
       </para>
+	  <para>
+		  Another option to customize the QueueChannel environment is provided by the <code>ref</code> attribute of the
+		  <code>&lt;int:queue&gt;</code> sub-element. This attribute implies the reference to any
+		  <interfacename>java.util.Queue</interfacename> implementation. An implementation is provided
+		  by the <ulink url="https://github.com/reactor/reactor">Project Reactor</ulink> and its
+		  <classname>reactor.queue.PersistentQueue</classname> implementation for the
+		  <ulink url="https://github.com/OpenHFT/Chronicle-Queue">IndexedChronicle</ulink>:
+	  </para>
+		<programlisting language="java"><![CDATA[@Bean
+public QueueChannel reactorQueue() {
+    return new QueueChannel(new PersistentQueueSpec<Message<?>>()
+                    .codec(new JavaSerializationCodec<Message<?>>())
+                    .basePath(System.getProperty("java.io.tmpdir") + "/reactor-queue")
+                    .get());
+}]]></programlisting>
     </section>
 
     <section id="channel-configuration-pubsubchannel">

src/reference/docbook/whats-new.xml

@@ -237,5 +237,16 @@
 				See <xref linkend="service-activator-namespace"/>.
 			</para>
 		</section>
+		<section id="4.1-queue-channel-queue.typ">
+			<title>QueueChannel: backed Queue type</title>
+			<para>
+				The <classname>QueueChannel</classname> backed <classname>Queue type</classname> has been changed
+				from <interfacename>BlockingQueue</interfacename> to the more generic
+				<interfacename>Queue</interfacename>. It allows the use of any external
+				<interfacename>Queue</interfacename> implementation, for example Reactor's
+				<classname>PersistentQueue</classname>.
+				See <xref linkend="channel-configuration-queuechannel"/>.
+			</para>
+		</section>
 	</section>
 </chapter>