From be6498008f4de7e5b4eff1f9e564dd84730cec94 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 16 Nov 2018 13:24:36 -0500 Subject: [PATCH] INT-4553: Store-backed QueueChannel improvements JIRA: https://jira.spring.io/browse/INT-4553 Fixes https://github.com/spring-projects/spring-integration/issues/2628 Fixes https://github.com/spring-projects/spring-integration/issues/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** # Conflicts: # src/reference/asciidoc/jdbc.adoc # src/reference/asciidoc/whats-new.adoc --- .../integration/channel/QueueChannel.java | 34 +-- .../integration/store/MessageGroupQueue.java | 56 ++++- .../channel/QueueChannelTests.java | 198 ++++++++++++------ .../store/MessageGroupQueueTests.java | 65 ++++-- ...racleChannelMessageStoreQueryProvider.java | 10 +- .../integration/jdbc/schema-db2.sql | 7 +- .../integration/jdbc/schema-derby.sql | 7 +- .../integration/jdbc/schema-h2.sql | 7 +- .../integration/jdbc/schema-hsqldb.sql | 7 +- .../integration/jdbc/schema-mysql.sql | 7 +- .../integration/jdbc/schema-oracle.sql | 7 +- .../integration/jdbc/schema-postgresql.sql | 7 +- .../integration/jdbc/schema-sqlserver.sql | 7 +- .../integration/jdbc/schema-sybase.sql | 7 +- src/reference/asciidoc/jdbc.adoc | 12 +- 15 files changed, 304 insertions(+), 134 deletions(-) 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 688a600241..24d61afd2a 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 @@ -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 class QueueChannel extends AbstractPollableChannel implements QueueChanne 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>(capacity); + this.queue = new LinkedBlockingQueue<>(capacity); } /** @@ -75,7 +75,7 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne * unbounded queue may lead to OutOfMemoryErrors. */ public QueueChannel() { - this(new LinkedBlockingQueue>()); + this(new LinkedBlockingQueue<>()); } @Override @@ -116,13 +116,19 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne 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); // 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 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne return ((BlockingQueue>) 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 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne @Override public List> clear() { - List> clearedMessages = new ArrayList>(); + List> clearedMessages = new ArrayList<>(); if (this.queue instanceof BlockingQueue) { ((BlockingQueue>) this.queue).drainTo(clearedMessages); } @@ -165,7 +173,7 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne if (selector == null) { return this.clear(); } - List> purgedMessages = new ArrayList>(); + List> purgedMessages = new ArrayList<>(); Object[] array = this.queue.toArray(); for (Object o : array) { Message message = (Message) o; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java index 7ed7f5af81..d69b4aa462 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java +++ b/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 class MessageGroupQueue extends AbstractQueue> implements Bloc 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 class MessageGroupQueue extends AbstractQueue> implements Bloc 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 class MessageGroupQueue extends AbstractQueue> implements Bloc public int drainTo(Collection> collection, int maxElements) { Assert.notNull(collection, "'collection' must not be null"); int originalSize = collection.size(); - ArrayList> list = new ArrayList>(); + ArrayList> list = new ArrayList<>(); final Lock storeLock = this.storeLock; try { storeLock.lockInterruptibly(); @@ -297,7 +333,7 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc while (this.size() == 0) { this.messageStoreNotEmpty.await(); } - message = this.doPoll(); + message = doPoll(); } finally { @@ -306,7 +342,7 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc return message; } - private Collection> getMessages() { + protected Collection> getMessages() { return this.messageGroupStore.getMessageGroup(this.groupId).getMessages(); } @@ -314,7 +350,7 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc * 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 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc /** * 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 @@ public class MessageGroupQueue extends AbstractQueue> implements Bloc } return offered; } + } 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 55def0220c..2f82b99ff3 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 @@ -20,8 +20,12 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.spy; +import java.util.ArrayDeque; import java.util.List; +import java.util.Queue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -46,147 +50,208 @@ public class QueueChannelTests { @Test public void testSimpleSendAndReceive() throws Exception { - final AtomicBoolean messageReceived = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); final QueueChannel channel = new QueueChannel(); - new Thread(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { Message message = channel.receive(); if (message != null) { - messageReceived.set(true); latch.countDown(); } - }).start(); - assertFalse(messageReceived.get()); - channel.send(new GenericMessage("testing")); - latch.await(10000, TimeUnit.MILLISECONDS); - assertTrue(messageReceived.get()); + }); + channel.send(new GenericMessage<>("testing")); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + exec.shutdownNow(); + } + + @Test + public void testSimpleSendAndReceiveNonBlockingQueue() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final QueueChannel channel = new QueueChannel(new ArrayDeque<>()); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + Message message = channel.receive(); + if (message != null) { + latch.countDown(); + } + }); + channel.send(new GenericMessage<>("testing")); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + exec.shutdownNow(); + } + + @Test + public void testSimpleSendAndReceiveNonBlockingQueueWithTimeout() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final QueueChannel channel = new QueueChannel(new ArrayDeque<>()); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + Message message = channel.receive(1); + if (message != null) { + latch.countDown(); + } + }); + channel.send(new GenericMessage<>("testing")); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + exec.shutdownNow(); } @Test public void testImmediateReceive() throws Exception { - final AtomicBoolean messageReceived = new AtomicBoolean(false); + final AtomicBoolean messageNull = new AtomicBoolean(false); final QueueChannel channel = new QueueChannel(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor(); Runnable receiveTask1 = () -> { Message message = channel.receive(0); - if (message != null) { - messageReceived.set(true); - } + messageNull.set(message == null); latch1.countDown(); }; - Runnable sendTask = () -> channel.send(new GenericMessage("testing")); + Runnable sendTask = () -> channel.send(new GenericMessage<>("testing")); singleThreadExecutor.execute(receiveTask1); - latch1.await(); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); singleThreadExecutor.execute(sendTask); - assertFalse(messageReceived.get()); Runnable receiveTask2 = () -> { Message message = channel.receive(0); if (message != null) { - messageReceived.set(true); + latch2.countDown(); } - latch2.countDown(); }; singleThreadExecutor.execute(receiveTask2); - latch2.await(); - assertTrue(messageReceived.get()); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); singleThreadExecutor.shutdownNow(); } @Test public void testBlockingReceiveWithNoTimeout() throws Exception { final QueueChannel channel = new QueueChannel(); - final AtomicBoolean receiveInterrupted = new AtomicBoolean(false); + final AtomicBoolean messageNull = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); - Thread t = new Thread(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { Message message = channel.receive(); - receiveInterrupted.set(true); - assertTrue(message == null); + messageNull.set(message == null); latch.countDown(); }); - t.start(); - assertFalse(receiveInterrupted.get()); - t.interrupt(); - latch.await(); - assertTrue(receiveInterrupted.get()); + exec.shutdownNow(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(messageNull.get()); } @Test public void testBlockingReceiveWithTimeout() throws Exception { final QueueChannel channel = new QueueChannel(); - final AtomicBoolean receiveInterrupted = new AtomicBoolean(false); + final AtomicBoolean messageNull = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); - Thread t = new Thread(() -> { + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { Message message = channel.receive(10000); - receiveInterrupted.set(true); - assertTrue(message == null); + messageNull.set(message == null); latch.countDown(); }); - t.start(); - assertFalse(receiveInterrupted.get()); - t.interrupt(); - latch.await(); - assertTrue(receiveInterrupted.get()); + exec.shutdownNow(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(messageNull.get()); + } + + @Test + public void testBlockingReceiveWithTimeoutEmptyThenSend() throws Exception { + Queue> queue = spy(new ArrayDeque<>()); + CountDownLatch pollLatch = new CountDownLatch(1); + AtomicBoolean first = new AtomicBoolean(true); + willAnswer(i -> { + pollLatch.countDown(); + return first.getAndSet(false) ? null : i.callRealMethod(); + }).given(queue).poll(); + final QueueChannel channel = new QueueChannel(queue); + final CountDownLatch latch = new CountDownLatch(1); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + Message message = channel.receive(10000); + if (message != null) { + latch.countDown(); + } + }); + assertTrue(pollLatch.await(10, TimeUnit.SECONDS)); + channel.send(new GenericMessage<>("testing")); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + exec.shutdownNow(); + } + + @Test + public void testBlockingReceiveNoTimeoutEmptyThenSend() throws Exception { + Queue> queue = spy(new ArrayDeque<>()); + CountDownLatch pollLatch = new CountDownLatch(1); + AtomicBoolean first = new AtomicBoolean(true); + willAnswer(i -> { + pollLatch.countDown(); + return first.getAndSet(false) ? null : i.callRealMethod(); + }).given(queue).poll(); + final QueueChannel channel = new QueueChannel(queue); + final CountDownLatch latch = new CountDownLatch(1); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + Message message = channel.receive(); + if (message != null) { + latch.countDown(); + } + }); + assertTrue(pollLatch.await(10, TimeUnit.SECONDS)); + channel.send(new GenericMessage<>("testing")); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + exec.shutdownNow(); } @Test public void testImmediateSend() { QueueChannel channel = new QueueChannel(3); - boolean result1 = channel.send(new GenericMessage("test-1")); + boolean result1 = channel.send(new GenericMessage<>("test-1")); assertTrue(result1); - boolean result2 = channel.send(new GenericMessage("test-2"), 100); + boolean result2 = channel.send(new GenericMessage<>("test-2"), 100); assertTrue(result2); - boolean result3 = channel.send(new GenericMessage("test-3"), 0); + boolean result3 = channel.send(new GenericMessage<>("test-3"), 0); assertTrue(result3); - boolean result4 = channel.send(new GenericMessage("test-4"), 0); + boolean result4 = channel.send(new GenericMessage<>("test-4"), 0); assertFalse(result4); } @Test public void testBlockingSendWithNoTimeout() throws Exception { final QueueChannel channel = new QueueChannel(1); - boolean result1 = channel.send(new GenericMessage("test-1")); + boolean result1 = channel.send(new GenericMessage<>("test-1")); assertTrue(result1); - final AtomicBoolean sendInterrupted = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); - Thread t = new Thread(() -> { - channel.send(new GenericMessage("test-2")); - sendInterrupted.set(true); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + channel.send(new GenericMessage<>("test-2")); latch.countDown(); }); - t.start(); - assertFalse(sendInterrupted.get()); - t.interrupt(); - latch.await(); - assertTrue(sendInterrupted.get()); + exec.shutdownNow(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); } @Test public void testBlockingSendWithTimeout() throws Exception { final QueueChannel channel = new QueueChannel(1); - boolean result1 = channel.send(new GenericMessage("test-1")); + boolean result1 = channel.send(new GenericMessage<>("test-1")); assertTrue(result1); - final AtomicBoolean sendInterrupted = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); - Thread t = new Thread(() -> { - channel.send(new GenericMessage("test-2"), 10000); - sendInterrupted.set(true); + ExecutorService exec = Executors.newSingleThreadExecutor(); + exec.execute(() -> { + channel.send(new GenericMessage<>("test-2"), 10000); latch.countDown(); }); - t.start(); - assertFalse(sendInterrupted.get()); - t.interrupt(); - latch.await(); - assertTrue(sendInterrupted.get()); + exec.shutdownNow(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); } @Test public void testClear() { QueueChannel channel = new QueueChannel(2); - GenericMessage message1 = new GenericMessage("test1"); - GenericMessage message2 = new GenericMessage("test2"); - GenericMessage message3 = new GenericMessage("test3"); + GenericMessage message1 = new GenericMessage<>("test1"); + GenericMessage message2 = new GenericMessage<>("test2"); + GenericMessage message3 = new GenericMessage<>("test3"); assertTrue(channel.send(message1)); assertTrue(channel.send(message2)); assertFalse(channel.send(message3, 0)); @@ -217,11 +282,11 @@ public class QueueChannelTests { .setExpirationDate(future).build(); assertTrue(channel.send(expiredMessage, 0)); assertTrue(channel.send(unexpiredMessage, 0)); - assertFalse(channel.send(new GenericMessage("atCapacity"), 0)); + assertFalse(channel.send(new GenericMessage<>("atCapacity"), 0)); List> purgedMessages = channel.purge(new UnexpiredMessageSelector()); assertNotNull(purgedMessages); assertEquals(1, purgedMessages.size()); - assertTrue(channel.send(new GenericMessage("roomAvailable"), 0)); + assertTrue(channel.send(new GenericMessage<>("roomAvailable"), 0)); } @Rule @@ -340,4 +405,5 @@ public class QueueChannelTests { assertTrue(latch4.await(1000, TimeUnit.MILLISECONDS)); } */ + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java index b6eb9507b9..2837756ee5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.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. @@ -19,15 +19,20 @@ package org.springframework.integration.store; 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 static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.spy; import java.util.HashSet; import java.util.Set; import java.util.concurrent.CompletionService; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorCompletionService; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -38,20 +43,56 @@ import org.springframework.messaging.support.GenericMessage; /** * @author Dave Syer + * @author Gary Russell * @since 2.0 */ public class MessageGroupQueueTests { - static final Log logger = LogFactory.getLog(MessageGroupQueueTests.class); + private static final Log logger = LogFactory.getLog(MessageGroupQueueTests.class); @Test public void testPutAndPoll() throws Exception { MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO"); - queue.put(new GenericMessage("foo")); + queue.put(new GenericMessage<>("foo")); Message result = queue.poll(100, TimeUnit.MILLISECONDS); assertNotNull(result); } + @Test + public void testPollTimeout() throws Exception { + MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO"); + Message result = queue.poll(1, TimeUnit.MILLISECONDS); + assertNull(result); + } + + @Test + public void testPollEmpty() throws Exception { + MessageGroupQueue queue = spy(new MessageGroupQueue(new SimpleMessageStore(), "FOO")); + CountDownLatch latch1 = new CountDownLatch(1); + AtomicBoolean first = new AtomicBoolean(true); + willAnswer(i -> { + latch1.countDown(); + return first.getAndSet(false) ? null : i.callRealMethod(); + }).given(queue).doPoll(); + ExecutorService exec = Executors.newSingleThreadExecutor(); + CountDownLatch latch2 = new CountDownLatch(1); + exec.execute(() -> { + try { + Message result = queue.poll(100, TimeUnit.MILLISECONDS); + if (result != null) { + latch2.countDown(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + queue.put(new GenericMessage<>("foo")); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + exec.shutdownNow(); + } + @Test public void testSize() throws Exception { MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO"); @@ -65,9 +106,9 @@ public class MessageGroupQueueTests { public void testCapacityAfterExpiry() throws Exception { SimpleMessageStore messageGroupStore = new SimpleMessageStore(); MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO", 2); - queue.put(new GenericMessage("foo")); + queue.put(new GenericMessage<>("foo")); assertEquals(1, queue.remainingCapacity()); - queue.put(new GenericMessage("bar")); + queue.put(new GenericMessage<>("bar")); assertEquals(0, queue.remainingCapacity()); Message result = queue.poll(100, TimeUnit.MILLISECONDS); assertNotNull(result); @@ -78,21 +119,21 @@ public class MessageGroupQueueTests { public void testCapacityExceeded() throws Exception { SimpleMessageStore messageGroupStore = new SimpleMessageStore(); MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO", 1); - queue.put(new GenericMessage("foo")); - assertFalse(queue.offer(new GenericMessage("bar"), 100, TimeUnit.MILLISECONDS)); + queue.put(new GenericMessage<>("foo")); + assertFalse(queue.offer(new GenericMessage<>("bar"), 100, TimeUnit.MILLISECONDS)); } @Test public void testPutAndTake() throws Exception { MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), "FOO"); - queue.put(new GenericMessage("foo")); + queue.put(new GenericMessage<>("foo")); Message result = queue.take(); assertNotNull(result); } @Test public void testConcurrentAccess() throws Exception { - doTestConcurrentAccess(50, 20, new HashSet()); + doTestConcurrentAccess(50, 20, new HashSet<>()); } @Test @@ -101,11 +142,10 @@ public class MessageGroupQueueTests { } private void doTestConcurrentAccess(int concurrency, final int maxPerTask, final Set set) throws Exception { - SimpleMessageStore messageGroupStore = new SimpleMessageStore(); final MessageGroupQueue queue = new MessageGroupQueue(messageGroupStore, "FOO"); ExecutorService executorService = Executors.newCachedThreadPool(); - CompletionService completionService = new ExecutorCompletionService(executorService); + CompletionService completionService = new ExecutorCompletionService<>(executorService); for (int i = 0; i < concurrency; i++) { @@ -114,7 +154,7 @@ public class MessageGroupQueueTests { completionService.submit(() -> { boolean result = true; for (int j = 0; j < maxPerTask; j++) { - result &= queue.add(new GenericMessage("count=" + big + ":" + j)); + result &= queue.add(new GenericMessage<>("count=" + big + ":" + j)); if (!result) { logger.warn("Failed to add"); } @@ -158,7 +198,6 @@ public class MessageGroupQueueTests { assertEquals(Integer.MAX_VALUE, queue.remainingCapacity()); executorService.shutdown(); - } } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.java index 1442dafcb2..da8f52887d 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/channel/OracleChannelMessageStoreQueryProvider.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. @@ -27,6 +27,8 @@ import org.springframework.jdbc.core.JdbcTemplate; * * @author Gunnar Hillert * @author Artem Bilan + * @author Gary Russell + * * @since 2.2 */ public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessageStoreQueryProvider { @@ -53,7 +55,8 @@ public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessa @Override public String getPriorityPollFromGroupExcludeIdsQuery() { - return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " + + return "SELECT /*+ INDEX(%PREFIX%CHANNEL_MESSAGE %PREFIX%CHANNEL_MSG_PRIORITY_IDX) */ " + + "%PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " + "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " + "and %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID not in (:message_ids) " + "order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED"; @@ -61,7 +64,8 @@ public class OracleChannelMessageStoreQueryProvider extends AbstractChannelMessa @Override public String getPriorityPollFromGroupQuery() { - return "SELECT %PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " + + return "SELECT /*+ INDEX(%PREFIX%CHANNEL_MESSAGE %PREFIX%CHANNEL_MSG_PRIORITY_IDX) */ " + + "%PREFIX%CHANNEL_MESSAGE.MESSAGE_ID, %PREFIX%CHANNEL_MESSAGE.MESSAGE_BYTES from %PREFIX%CHANNEL_MESSAGE " + "where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key and %PREFIX%CHANNEL_MESSAGE.REGION = :region " + "order by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCE FOR UPDATE SKIP LOCKED"; } diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql index 477d808db6..eabfd5a9eb 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE BIGINT NOT NULL , MESSAGE_BYTES BLOB, REGION VARCHAR(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ); -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql index 0a69f37da4..b095010a37 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1), MESSAGE_BYTES BLOB, REGION VARCHAR(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ); -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql index bbe50da7b3..e2c15e92cd 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE BIGINT NOT NULL , MESSAGE_BYTES LONGVARBINARY, REGION VARCHAR(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ); -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql index 1b9f443709..05dc4a79e6 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE BIGINT NOT NULL , MESSAGE_BYTES LONGVARBINARY, REGION VARCHAR(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ); -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql index a6f3587375..47c13591ca 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE BIGINT NOT NULL AUTO_INCREMENT UNIQUE, MESSAGE_BYTES BLOB, REGION VARCHAR(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ) ENGINE=InnoDB; -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql index eda21929a7..b25c387ac4 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE NUMBER(19,0) NOT NULL , MESSAGE_BYTES BLOB, REGION VARCHAR2(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ); -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql index c11e03cb71..fe17bde63c 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE BIGINT NOT NULL DEFAULT nextval('INT_MESSAGE_SEQ'), MESSAGE_BYTES BYTEA, REGION VARCHAR(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ); -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql index 6d5bfff80e..9012fe12ea 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE BIGINT NOT NULL , MESSAGE_BYTES IMAGE, REGION VARCHAR(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ); -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql index 5863509f79..f920c4a837 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql @@ -46,11 +46,12 @@ CREATE TABLE INT_CHANNEL_MESSAGE ( MESSAGE_SEQUENCE BIGINT NOT NULL , MESSAGE_BYTES IMAGE, REGION VARCHAR(100) NOT NULL, - constraint INT_CHANNEL_MESSAGE_PK primary key (GROUP_KEY, MESSAGE_ID, REGION) + constraint INT_CHANNEL_MESSAGE_PK primary key (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) ) LOCK DATAROWS; -CREATE INDEX INT_CHANNEL_MSG_DATE_IDX ON INT_CHANNEL_MESSAGE (CREATED_DATE, MESSAGE_SEQUENCE); -CREATE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); +CREATE INDEX INT_CHANNEL_MSG_DELETE_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_ID); +-- This is only needed if the message group store property 'priorityEnabled' is true +-- CREATE UNIQUE INDEX INT_CHANNEL_MSG_PRIORITY_IDX ON INT_CHANNEL_MESSAGE (REGION, GROUP_KEY, MESSAGE_PRIORITY DESC, CREATED_DATE, MESSAGE_SEQUENCE); CREATE TABLE INT_METADATA_STORE ( diff --git a/src/reference/asciidoc/jdbc.adoc b/src/reference/asciidoc/jdbc.adoc index 233e1ee3ed..f8c258eb19 100644 --- a/src/reference/asciidoc/jdbc.adoc +++ b/src/reference/asciidoc/jdbc.adoc @@ -153,8 +153,8 @@ If not explicitly set, the attribute defaults to _0_. [[jdbc-outbound-channel-adapter]] === Outbound Channel Adapter -The outbound Channel Adapter is the inverse of the inbound: its role is to handle a message and use it to execute a SQL query. -The message payload and headers are available by default as input parameters to the query, for instance: +The outbound channel adapter is the inverse of the inbound: its role is to handle a message and use it to execute a SQL query. +By default, the message payload and headers are available as input parameters to the query, as the following example shows: [source,xml] ---- @@ -269,7 +269,7 @@ For example: [source,xml] ---- ---- @@ -327,6 +327,12 @@ Spring Integration provides 2 JDBC specific Message Store implementations. The first one, is the `JdbcMessageStore` which is suitable to be used in conjunction with _Aggregators_ and the _Claim-Check_ pattern. While it can be used for backing _Message Channels_ as well, you may want to consider using the `JdbcChannelMessageStore` implementation instead, as it provides a more targeted and scalable implementation. +IMPORTANT: Starting with versions 5.0.11, 5.1.2, the indexes for the `JdbcChannelMessageStore` have been optimized. +If you have large message groups in such a store, you may wish to alter the indexes. +Furthermore, the index for `PriorityChannel` is commented out because it is not needed unless you are using such channels backed by JDBC. + +NOTE: When using the `OracleChannelMessageStoreQueryProvider`, the priority channel index **must** be added because it is included in a hint in the query. + ==== Initializing the Database Before starting to use JDBC Message Store components, it is important to provision target data base with the appropriate objects.