diff --git a/build.gradle b/build.gradle index 7dde91ff65..516e780429 100644 --- a/build.gradle +++ b/build.gradle @@ -712,6 +712,7 @@ project('spring-integration-jdbc') { dependencies { api project(':spring-integration-core') api 'org.springframework:spring-jdbc' + optionalApi "org.postgresql:postgresql:$postgresVersion" testImplementation "com.h2database:h2:$h2Version" testImplementation "org.hsqldb:hsqldb:$hsqldbVersion" @@ -721,6 +722,7 @@ project('spring-integration-jdbc') { testImplementation "mysql:mysql-connector-java:$mysqlVersion" testImplementation "org.apache.commons:commons-dbcp2:$commonsDbcp2Version" testImplementation 'org.testcontainers:mysql' + testImplementation 'org.testcontainers:postgresql' testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind' } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PgConnectionSupplier.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PgConnectionSupplier.java new file mode 100644 index 0000000000..fc12e8d807 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PgConnectionSupplier.java @@ -0,0 +1,54 @@ +/* + * Copyright 2022 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 + * + * https://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.jdbc.channel; + +import java.sql.SQLException; +import java.util.Properties; + +import org.postgresql.jdbc.PgConnection; + +/** + * A connection supplier for a {@link PgConnection} to a Postgres database that is + * to be used for a {@link PostgresChannelMessageTableSubscriber}. + *

+ * The supplied connection must not be read from a shared connection pool, typically + * represented by a {@link javax.sql.DataSource}. If a shared connection pool is used, this + * pool might reclaim a connection that was not closed within a given time frame. This + * becomes a problem as a {@link PostgresChannelMessageTableSubscriber} requires a dedicated + * {@link java.sql.Connection} to receive notifications from the Postgres database. This + * connection needs to remain open over a longer period of time. Typically, a + * {@link PgConnection} should be created directly via + * {@link java.sql.Driver#connect(String, Properties)} and a subsequent call + * to {@link java.sql.Connection#unwrap(Class)}. + * + * @author Rafael Winterhalter + * + * @since 6.0 + * + * @see PostgresChannelMessageTableSubscriber + */ +@FunctionalInterface +public interface PgConnectionSupplier { + + /** + * Supply an open, un-pooled connection to a Postgres database. + * @return A dedicated connection to a Postgres database for listening. + * @throws SQLException If the connection could not be established. + */ + PgConnection get() throws SQLException; + +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java new file mode 100644 index 0000000000..f8f5b29623 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java @@ -0,0 +1,289 @@ +/* + * Copyright 2022 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 + * + * https://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.jdbc.channel; + +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.postgresql.PGNotification; +import org.postgresql.jdbc.PgConnection; + +import org.springframework.context.SmartLifecycle; +import org.springframework.core.log.LogAccessor; +import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; +import org.springframework.integration.util.UUIDConverter; +import org.springframework.lang.Nullable; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; +import org.springframework.util.Assert; + +/** + * A subscriber for new messages being received by a Postgres database via a + * {@link JdbcChannelMessageStore}. This subscriber implementation is using + * Postgres' LISTEN/NOTIFY mechanism to allow for receiving push + * notifications for new messages what functions even if a message is written + * and read from different JVMs or {@link JdbcChannelMessageStore}s. + *

+ * Note that this subscriber requires an unshared {@link PgConnection} which + * remains open for any lifecycle. It is therefore recommended to execute a single + * subscriber for any JVM. For this reason, this subscriber is region-agnostic. + * To listen for messages for a given region and group id, use a + * {@link Subscription} and register it with this subscriber. + *

+ * In order to function, the Postgres database that is used must define a trigger + * for sending notifications upon newly arrived messages. This trigger is defined + * in the schema-postgresql.sql file within this artifact but commented + * out. + * + * @author Rafael Winterhalter + * @author Artem Bilan + * + * @since 6.0 + */ +public final class PostgresChannelMessageTableSubscriber implements SmartLifecycle { + + private static final LogAccessor LOGGER = new LogAccessor(PostgresChannelMessageTableSubscriber.class); + + private final Map> subscriptions = new ConcurrentHashMap<>(); + + private final PgConnectionSupplier connectionSupplier; + + private final String tablePrefix; + + @Nullable + private ExecutorService executor; + + private CountDownLatch latch = new CountDownLatch(0); + + private Future future = CompletableFuture.completedFuture(null); + + @Nullable + private volatile PgConnection connection; + + /** + * Create a new subscriber using the {@link JdbcChannelMessageStore#DEFAULT_TABLE_PREFIX}. + * @param connectionSupplier The connection supplier for the targeted Postgres database. + */ + public PostgresChannelMessageTableSubscriber(PgConnectionSupplier connectionSupplier) { + this(connectionSupplier, JdbcChannelMessageStore.DEFAULT_TABLE_PREFIX); + } + + /** + * Create a new subscriber. + * @param tablePrefix The table prefix of the {@link JdbcChannelMessageStore} to subscribe to. + * @param connectionSupplier The connection supplier for the targeted Postgres database. + */ + public PostgresChannelMessageTableSubscriber(PgConnectionSupplier connectionSupplier, String tablePrefix) { + Assert.notNull(connectionSupplier, "A connectionSupplier must be provided."); + Assert.notNull(tablePrefix, "A table prefix must be set."); + this.connectionSupplier = connectionSupplier; + this.tablePrefix = tablePrefix; + } + + /** + * Define an executor to use for listening for new messages. Note that the Postgres SQL driver implements + * listening for notifications as a blocking operation which will permanently block a thread of this executor + * while running. + * @param executor The executor to use or {@code null} if an executor should be created by this class. + */ + public void setExecutor(@Nullable ExecutorService executor) { + this.executor = executor; + } + + /** + * Add a new subscription to this subscriber. + * @param subscription The subscription to register. + * @return {@code true} if the subscription was not already added. + */ + public boolean subscribe(Subscription subscription) { + String subscriptionKey = subscription.getRegion() + " " + getKey(subscription.getGroupId()); + Set subscriptions = + this.subscriptions.computeIfAbsent(subscriptionKey, __ -> ConcurrentHashMap.newKeySet()); + return subscriptions.add(subscription); + } + + /** + * Remove a previous subscription from this subscriber. + * @param subscription The subscription to remove. + * @return {@code true} if the subscription was previously registered and is now removed. + */ + public boolean unsubscribe(Subscription subscription) { + String subscriptionKey = subscription.getRegion() + " " + getKey(subscription.getGroupId()); + Set subscriptions = this.subscriptions.get(subscriptionKey); + return subscriptions != null && subscriptions.remove(subscription); + } + + @Override + public synchronized void start() { + if (this.latch.getCount() > 0) { + return; + } + ExecutorService executor = this.executor; + if (executor == null) { + CustomizableThreadFactory threadFactory = + new CustomizableThreadFactory("postgres-channel-message-table-subscriber-"); + threadFactory.setDaemon(true); + executor = Executors.newSingleThreadExecutor(threadFactory); + this.executor = executor; + } + this.latch = new CountDownLatch(1); + this.future = executor.submit(() -> { + try { + while (isActive()) { + try { + PgConnection conn = this.connectionSupplier.get(); + try (Statement stmt = conn.createStatement()) { + stmt.execute("LISTEN " + this.tablePrefix.toLowerCase() + "channel_message_notify"); + } + catch (Throwable t) { + try { + conn.close(); + } + catch (Throwable suppressed) { + t.addSuppressed(suppressed); + } + throw t; + } + this.subscriptions.values() + .forEach(subscriptions -> subscriptions.forEach(Subscription::notifyUpdate)); + try { + this.connection = conn; + while (isActive()) { + PGNotification[] notifications = conn.getNotifications(0); + // Unfortunately, there is no good way of interrupting a notification + // poll but by closing its connection. + if (!isActive()) { + return; + } + if (notifications != null) { + for (PGNotification notification : notifications) { + String parameter = notification.getParameter(); + Set subscriptions = this.subscriptions.get(parameter); + if (subscriptions == null) { + continue; + } + for (Subscription subscription : subscriptions) { + subscription.notifyUpdate(); + } + } + } + } + } + finally { + conn.close(); + } + } + catch (Exception e) { + // The getNotifications method does not throw a meaningful message on interruption. + // Therefore, we do not log an error, unless it occurred while active. + if (isActive()) { + LOGGER.error(e, "Failed to poll notifications from Postgres database"); + } + } + catch (Throwable t) { + LOGGER.error(t, "Failed to poll notifications from Postgres database"); + return; + } + } + } + finally { + this.latch.countDown(); + } + }); + } + + private boolean isActive() { + if (Thread.interrupted()) { + Thread.currentThread().interrupt(); + return false; + } + return true; + } + + @Override + public synchronized void stop() { + Future future = this.future; + if (future.isDone()) { + return; + } + future.cancel(true); + PgConnection conn = this.connection; + if (conn != null) { + try { + conn.close(); + } + catch (SQLException ignored) { + } + } + try { + if (!this.latch.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("Failed to stop " + + PostgresChannelMessageTableSubscriber.class.getName()); + } + } + catch (InterruptedException ignored) { + } + } + + @Override + public boolean isRunning() { + return this.latch.getCount() > 0; + } + + private static String getKey(Object input) { + return input == null ? null : UUIDConverter.getUUID(input).toString(); + } + + /** + * A subscription to a {@link PostgresChannelMessageTableSubscriber} for + * receiving push notifications for new messages that are added to + * a {@link JdbcChannelMessageStore}. + */ + public interface Subscription { + + /** + * Indicate that a message was added to the represented region and + * group id. Note that this method might also be invoked if there are + * no new messages to read, for example if another subscription already + * read those messages or if a new messages might have arrived during + * a temporary connection loss. + */ + void notifyUpdate(); + + /** + * Return the region for which this subscription receives notifications. + * @return The relevant region of the {@link JdbcChannelMessageStore}. + */ + String getRegion(); + + /** + * Return the group id for which this subscription receives notifications. + * @return The group id of the {@link PostgresSubscribableChannel}. + */ + Object getGroupId(); + + } + +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java new file mode 100644 index 0000000000..dde6bfcd34 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java @@ -0,0 +1,129 @@ +/* + * Copyright 2022 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 + * + * https://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.jdbc.channel; + +import java.util.concurrent.Executor; + +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.integration.channel.AbstractSubscribableChannel; +import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.UnicastingDispatcher; +import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; + +/** + * An {@link AbstractSubscribableChannel} for receiving push notifications for + * messages send to a group id of a {@link JdbcChannelMessageStore}. Receiving + * such push notifications is only possible if using a Postgres database. + *

+ * In order to function, the Postgres database that is used must define a trigger + * for sending notifications upon newly arrived messages. This trigger is defined + * in the schema-postgresql.sql file within this artifact but commented + * out. + * + * @author Rafael Winterhalter + * @author Artem Bilan + * + * @since 6.0 + */ +public class PostgresSubscribableChannel extends AbstractSubscribableChannel + implements PostgresChannelMessageTableSubscriber.Subscription { + + private final JdbcChannelMessageStore jdbcChannelMessageStore; + + private final Object groupId; + + private final PostgresChannelMessageTableSubscriber messageTableSubscriber; + + private UnicastingDispatcher dispatcher = new UnicastingDispatcher(new SimpleAsyncTaskExecutor()); + + /** + * Create a subscribable channel for a Postgres database. + * @param jdbcChannelMessageStore The message store to use for the relevant region. + * @param groupId The group id that is targeted by the subscription. + * @param messageTableSubscriber The subscriber to use for receiving notifications. + */ + public PostgresSubscribableChannel(JdbcChannelMessageStore jdbcChannelMessageStore, + Object groupId, PostgresChannelMessageTableSubscriber messageTableSubscriber) { + Assert.notNull(jdbcChannelMessageStore, "A jdbcChannelMessageStore must be provided."); + Assert.notNull(groupId, "A groupId must be set."); + Assert.notNull(messageTableSubscriber, "A messageTableSubscriber must be set."); + this.jdbcChannelMessageStore = jdbcChannelMessageStore; + this.groupId = groupId; + this.messageTableSubscriber = messageTableSubscriber; + } + + /** + * Set the executor to use for dispatching newly received messages. + * @param executor The executor to use. + */ + public void setDispatcherExecutor(Executor executor) { + Assert.notNull(executor, "An executor must be provided."); + this.dispatcher = new UnicastingDispatcher(executor); + } + + @Override + public boolean subscribe(MessageHandler handler) { + boolean subscribed = super.subscribe(handler); + if (this.dispatcher.getHandlerCount() == 1) { + this.messageTableSubscriber.subscribe(this); + notifyUpdate(); + } + return subscribed; + } + + @Override + public boolean unsubscribe(MessageHandler handle) { + boolean unsubscribed = super.unsubscribe(handle); + if (this.dispatcher.getHandlerCount() == 0) { + this.messageTableSubscriber.unsubscribe(this); + } + return unsubscribed; + } + + @Override + protected MessageDispatcher getDispatcher() { + return this.dispatcher; + } + + @Override + protected boolean doSend(Message message, long timeout) { + this.jdbcChannelMessageStore.addMessageToGroup(this.groupId, message); + return true; + } + + @Override + public void notifyUpdate() { + Message message; + while ((message = this.jdbcChannelMessageStore.pollMessageFromGroup(this.groupId)) != null) { + this.dispatcher.dispatch(message); + } + } + + @Override + public String getRegion() { + return this.jdbcChannelMessageStore.getRegion(); + } + + @Override + public Object getGroupId() { + return this.groupId; + } + +} diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/package-info.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/package-info.java new file mode 100644 index 0000000000..ea360d5972 --- /dev/null +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides a message channel-specific JDBC API. + */ +package org.springframework.integration.jdbc.channel; diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java index 053b9bb005..736665cb58 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java @@ -273,6 +273,15 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto this.region = region; } + /** + * Returns the current region that was set or {@link #DEFAULT_REGION}, which is the default. + * @return the set region name + * @since 6.0 + */ + public String getRegion() { + return this.region; + } + /** * A converter for serializing messages to byte arrays for storage. * @param serializer The serializer to set 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 25fd2c44be..45894c57ea 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 @@ -58,3 +58,20 @@ CREATE TABLE INT_METADATA_STORE ( REGION VARCHAR(100) NOT NULL, constraint INT_METADATA_STORE_PK primary key (METADATA_KEY, REGION) ); + +-- This is only needed if using PostgresChannelMessageSubscriber + +/*CREATE FUNCTION INT_CHANNEL_MESSAGE_NOTIFY_FCT() +RETURNS TRIGGER AS + $BODY$ + BEGIN + PERFORM pg_notify('int_channel_message_notify', NEW.REGION || ' ' || NEW.GROUP_KEY); + RETURN NEW; + END; + $BODY$ + LANGUAGE PLPGSQL; + + CREATE TRIGGER INT_CHANNEL_MESSAGE_NOTIFY_TRG + AFTER INSERT ON INT_CHANNEL_MESSAGE + FOR EACH ROW + EXECUTE PROCEDURE INT_CHANNEL_MESSAGE_NOTIFY_FCT();*/ diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java new file mode 100644 index 0000000000..1fc5380b1c --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java @@ -0,0 +1,194 @@ +/* + * Copyright 2022 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 + * + * https://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.jdbc.channel; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.sql.DriverManager; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import javax.sql.DataSource; + +import org.apache.commons.dbcp2.BasicDataSource; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.postgresql.jdbc.PgConnection; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; +import org.springframework.integration.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider; +import org.springframework.jdbc.datasource.init.DataSourceInitializer; +import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator; +import org.springframework.jdbc.datasource.init.ScriptUtils; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Rafael Winterhalter + * @author Artem Bilan + * + * @since 6.0 + */ +@SpringJUnitConfig +@DirtiesContext +public class PostgresChannelMessageTableSubscriberTests implements PostgresContainerTest { + + private static final String INTEGRATION_DB_SCRIPTS = """ + CREATE SEQUENCE INT_MESSAGE_SEQ START WITH 1 INCREMENT BY 1 NO CYCLE; + ^^^ END OF SCRIPT ^^^ + + CREATE TABLE INT_CHANNEL_MESSAGE ( + MESSAGE_ID CHAR(36) NOT NULL, + GROUP_KEY CHAR(36) NOT NULL, + CREATED_DATE BIGINT NOT NULL, + MESSAGE_PRIORITY BIGINT, + 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 (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE) + ); + ^^^ END OF SCRIPT ^^^ + + CREATE FUNCTION INT_CHANNEL_MESSAGE_NOTIFY_FCT() + RETURNS TRIGGER AS + $BODY$ + BEGIN + PERFORM pg_notify('int_channel_message_notify', NEW.REGION || ' ' || NEW.GROUP_KEY); + RETURN NEW; + END; + $BODY$ + LANGUAGE PLPGSQL; + ^^^ END OF SCRIPT ^^^ + + CREATE TRIGGER INT_CHANNEL_MESSAGE_NOTIFY_TRG + AFTER INSERT ON INT_CHANNEL_MESSAGE + FOR EACH ROW + EXECUTE PROCEDURE INT_CHANNEL_MESSAGE_NOTIFY_FCT(); + ^^^ END OF SCRIPT ^^^ + """; + + @Autowired + private JdbcChannelMessageStore messageStore; + + private PostgresChannelMessageTableSubscriber postgresChannelMessageTableSubscriber; + + @BeforeEach + void setUp() { + // Not initiated as a bean to allow for registrations prior and post the life cycle + this.postgresChannelMessageTableSubscriber = new PostgresChannelMessageTableSubscriber( + () -> DriverManager.getConnection(POSTGRES_CONTAINER.getJdbcUrl(), + POSTGRES_CONTAINER.getUsername(), + POSTGRES_CONTAINER.getPassword()) + .unwrap(PgConnection.class) + ); + } + + @Test + public void testMessagePollMessagesAddedAfterStart() throws Exception { + CountDownLatch latch = new CountDownLatch(2); + List payloads = new ArrayList<>(); + postgresChannelMessageTableSubscriber.start(); + try { + PostgresSubscribableChannel channel = new PostgresSubscribableChannel(messageStore, + "testMessagePollMessagesAddedAfterStart", + postgresChannelMessageTableSubscriber); + channel.subscribe(message -> { + payloads.add(message.getPayload()); + latch.countDown(); + }); + messageStore.addMessageToGroup("testMessagePollMessagesAddedAfterStart", new GenericMessage<>("1")); + messageStore.addMessageToGroup("testMessagePollMessagesAddedAfterStart", new GenericMessage<>("2")); + assertThat(latch.await(3, TimeUnit.SECONDS)) + .as("Expected Postgres notification within 3 seconds") + .isTrue(); + } + finally { + postgresChannelMessageTableSubscriber.stop(); + } + assertThat(payloads).containsExactly("1", "2"); + } + + @Test + public void testMessagePollMessagesAddedBeforeStart() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(2); + List payloads = new ArrayList<>(); + PostgresSubscribableChannel channel = + new PostgresSubscribableChannel(messageStore, + "testMessagePollMessagesAddedBeforeStart", + postgresChannelMessageTableSubscriber); + channel.subscribe(message -> { + payloads.add(message.getPayload()); + latch.countDown(); + }); + messageStore.addMessageToGroup("testMessagePollMessagesAddedBeforeStart", new GenericMessage<>("1")); + messageStore.addMessageToGroup("testMessagePollMessagesAddedBeforeStart", new GenericMessage<>("2")); + postgresChannelMessageTableSubscriber.start(); + try { + assertThat(latch.await(3, TimeUnit.SECONDS)) + .as("Expected Postgres notification within 3 seconds") + .isTrue(); + } + finally { + postgresChannelMessageTableSubscriber.stop(); + } + assertThat(payloads).containsExactly("1", "2"); + } + + @Configuration + @EnableIntegration + public static class Config { + + @Bean + public DataSource dataSource() { + BasicDataSource dataSource = new BasicDataSource(); + dataSource.setUrl(PostgresContainerTest.getJdbcUrl()); + dataSource.setUsername(PostgresContainerTest.getUsername()); + dataSource.setPassword(PostgresContainerTest.getPassword()); + return dataSource; + } + + @Bean + DataSourceInitializer dataSourceInitializer(DataSource dataSource) { + DataSourceInitializer dataSourceInitializer = new DataSourceInitializer(); + dataSourceInitializer.setDataSource(dataSource); + ResourceDatabasePopulator databasePopulator = + new ResourceDatabasePopulator(new ByteArrayResource(INTEGRATION_DB_SCRIPTS.getBytes())); + databasePopulator.setSeparator(ScriptUtils.EOF_STATEMENT_SEPARATOR); + dataSourceInitializer.setDatabasePopulator( + databasePopulator); + return dataSourceInitializer; + } + + @Bean + public JdbcChannelMessageStore jdbcChannelMessageStore(DataSource dataSource) { + JdbcChannelMessageStore messageStore = new JdbcChannelMessageStore(dataSource); + messageStore.setRegion("PostgresChannelMessageTableSubscriberTest"); + messageStore.setChannelMessageStoreQueryProvider(new PostgresChannelMessageStoreQueryProvider()); + return messageStore; + } + + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresContainerTest.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresContainerTest.java new file mode 100644 index 0000000000..4ae45e8923 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresContainerTest.java @@ -0,0 +1,60 @@ +/* + * Copyright 2022 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 + * + * https://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.jdbc.channel; + +import org.junit.jupiter.api.BeforeAll; +import org.testcontainers.containers.PostgreSQLContainer; +import org.testcontainers.junit.jupiter.Testcontainers; + +/** + * The base contract for JUnit tests based on the container for Postgres. + * The Testcontainers 'reuse' option must be disabled, so, Ryuk container is started + * and will clean all the containers up from this test suite after JVM exit. + * Since the Postgres container instance is shared via static property, it is going to be + * started only once per JVM, therefore the target Docker container is reused automatically. + * + * @author Rafael Winterhalter + * + * @since 6.0 + */ +@Testcontainers(disabledWithoutDocker = true) +public interface PostgresContainerTest { + + PostgreSQLContainer POSTGRES_CONTAINER = new PostgreSQLContainer<>("postgres:11"); + + @BeforeAll + static void startContainer() { + POSTGRES_CONTAINER.start(); + } + + static String getDriverClassName() { + return POSTGRES_CONTAINER.getDriverClassName(); + } + + static String getJdbcUrl() { + return POSTGRES_CONTAINER.getJdbcUrl(); + } + + static String getUsername() { + return POSTGRES_CONTAINER.getUsername(); + } + + static String getPassword() { + return POSTGRES_CONTAINER.getPassword(); + } + +}