INT-4553: Store-backed QueueChannel improvements

JIRA: https://jira.spring.io/browse/INT-4553
Fixes #2628
Fixes #2629

- Avoid `size()` calls on the MGS, use `poll()` instead.
- Optimize the indexes for the `INT_CHANNEL_MESSAGE` table.

Avoid size call when no timeout too.

Polishing - PR Comments

Missed a doc fix

Another missed %PREFIX%

Fix underscores

Polishing; PR comments; make MGQ extendable.

Fix version in doc.

* Polishing `@since`
* Use diamonds whenever it is possible

**Cherry-pick to 5.0.x**

Author: garyrussell

spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -66,7 +66,7 @@ public QueueChannel(Queue<Message<?>> queue) {
 	public QueueChannel(int capacity) {
 		Assert.isTrue(capacity > 0, "The capacity must be a positive integer. " +
 				"For a zero-capacity alternative, consider using a 'RendezvousChannel'.");
-		this.queue = new LinkedBlockingQueue<Message<?>>(capacity);
+		this.queue = new LinkedBlockingQueue<>(capacity);
 	}
 
 	/**
@@ -75,7 +75,7 @@ public QueueChannel(int capacity) {
 	 * unbounded queue may lead to OutOfMemoryErrors.
 	 */
 	public QueueChannel() {
-		this(new LinkedBlockingQueue<Message<?>>());
+		this(new LinkedBlockingQueue<>());
 	}
 
 	@Override
@@ -116,13 +116,19 @@ protected Message<?> doReceive(long timeout) {
 					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); // NOSONAR - ok to ignore result
-						nanos = deadline - System.nanoTime();
+					Message<?> message = this.queue.poll();
+					if (message == null) {
+						long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
+						long deadline = System.nanoTime() + nanos;
+						while (message == null && nanos > 0) {
+							this.queueSemaphore.tryAcquire(nanos, TimeUnit.NANOSECONDS); // NOSONAR ok to ignore result
+							message = this.queue.poll();
+							if (message == null) {
+								nanos = deadline - System.nanoTime();
+							}
+						}
 					}
-					return this.queue.poll();
+					return message;
 				}
 			}
 			if (timeout == 0) {
@@ -133,10 +139,12 @@ protected Message<?> doReceive(long timeout) {
 				return ((BlockingQueue<Message<?>>) this.queue).take();
 			}
 			else {
-				while (this.queue.size() == 0) {
+				Message<?> message = this.queue.poll();
+				while (message == null) {
 					this.queueSemaphore.tryAcquire(50, TimeUnit.MILLISECONDS);
+					message = this.queue.poll();
 				}
-				return this.queue.poll();
+				return message;
 			}
 		}
 		catch (InterruptedException e) {
@@ -147,7 +155,7 @@ protected Message<?> doReceive(long timeout) {
 
 	@Override
 	public List<Message<?>> clear() {
-		List<Message<?>> clearedMessages = new ArrayList<Message<?>>();
+		List<Message<?>> clearedMessages = new ArrayList<>();
 		if (this.queue instanceof BlockingQueue) {
 			((BlockingQueue<Message<?>>) this.queue).drainTo(clearedMessages);
 		}
@@ -165,7 +173,7 @@ public List<Message<?>> purge(MessageSelector selector) {
 		if (selector == null) {
 			return this.clear();
 		}
-		List<Message<?>> purgedMessages = new ArrayList<Message<?>>();
+		List<Message<?>> purgedMessages = new ArrayList<>();
 		Object[] array = this.queue.toArray();
 		for (Object o : array) {
 			Message<?> message = (Message<?>) o;

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -121,6 +121,42 @@ public Iterator<Message<?>> iterator() {
 		return getMessages().iterator();
 	}
 
+	/**
+	 * Get the store.
+	 * @return the store.
+	 * @since 5.0.11
+	 */
+	protected BasicMessageGroupStore getMessageGroupStore() {
+		return this.messageGroupStore;
+	}
+
+	/**
+	 * Get the store lock.
+	 * @return the lock.
+	 * @since 5.0.11
+	 */
+	protected Lock getStoreLock() {
+		return this.storeLock;
+	}
+
+	/**
+	 * Get the not full condition.
+	 * @return the condition.
+	 * @since 5.0.11
+	 */
+	protected Condition getMessageStoreNotFull() {
+		return this.messageStoreNotFull;
+	}
+
+	/**
+	 * Get the not empty condition.
+	 * @return the condition.
+	 * @since 5.0.11
+	 */
+	protected Condition getMessageStoreNotEmpty() {
+		return this.messageStoreNotEmpty;
+	}
+
 	@Override
 	public int size() {
 		return this.messageGroupStore.messageGroupSize(this.groupId);
@@ -156,11 +192,11 @@ public Message<?> poll(long timeout, TimeUnit unit) throws InterruptedException
 		storeLock.lockInterruptibly();
 
 		try {
-			while (this.size() == 0 && timeoutInNanos > 0) {
+			message = doPoll();
+			while (message == null && timeoutInNanos > 0) {
 				timeoutInNanos = this.messageStoreNotEmpty.awaitNanos(timeoutInNanos);
+				message = doPoll();
 			}
-			message = this.doPoll();
-
 		}
 		finally {
 			storeLock.unlock();
@@ -196,7 +232,7 @@ public int drainTo(Collection<? super Message<?>> c) {
 	public int drainTo(Collection<? super Message<?>> collection, int maxElements) {
 		Assert.notNull(collection, "'collection' must not be null");
 		int originalSize = collection.size();
-		ArrayList<Message<?>> list = new ArrayList<Message<?>>();
+		ArrayList<Message<?>> list = new ArrayList<>();
 		final Lock storeLock = this.storeLock;
 		try {
 			storeLock.lockInterruptibly();
@@ -297,7 +333,7 @@ public Message<?> take() throws InterruptedException {
 			while (this.size() == 0) {
 				this.messageStoreNotEmpty.await();
 			}
-			message = this.doPoll();
+			message = doPoll();
 
 		}
 		finally {
@@ -306,15 +342,15 @@ public Message<?> take() throws InterruptedException {
 		return message;
 	}
 
-	private Collection<Message<?>> getMessages() {
+	protected Collection<Message<?>> getMessages() {
 		return this.messageGroupStore.getMessageGroup(this.groupId).getMessages();
 	}
 
 	/**
 	 * It is assumed that the 'storeLock' is being held by the caller, otherwise
 	 * IllegalMonitorStateException may be thrown
 	 */
-	private Message<?> doPoll() {
+	protected Message<?> doPoll() {
 		Message<?> message = this.messageGroupStore.pollMessageFromGroup(this.groupId);
 		this.messageStoreNotFull.signal();
 		return message;
@@ -323,8 +359,9 @@ private Message<?> doPoll() {
 	/**
 	 * It is assumed that the 'storeLock' is being held by the caller, otherwise
 	 * IllegalMonitorStateException may be thrown
+	 * @param message the message to offer.
 	 */
-	private boolean doOffer(Message<?> message) {
+	protected boolean doOffer(Message<?> message) {
 		boolean offered = false;
 		if (this.capacity == Integer.MAX_VALUE || this.size() < this.capacity) {
 			this.messageGroupStore.addMessageToGroup(this.groupId, message);
@@ -333,4 +370,5 @@ private boolean doOffer(Message<?> message) {
 		}
 		return offered;
 	}
+
 }