GH-3872: Add PostgresSubscribableChannel implementation
Fixes https://github.com/spring-projects/spring-integration/issues/3872 Adds an implementation for a Postres-compatible notification listener for a `JdbcChannelMessageStore`. * Introduce `PostgresChannelMessageTableSubscriber.Subscription` contract * Implement a `PostgresSubscribableChannel` * Add Javadoc and fix code formatting issues. * Handle temporary connection loss. * Add tests for `PostgresChannelMessageTableSubscriber` * Replace NOTIFY command with pg_notify function call. * Code style clean up * Fix some Javadocs * Use `DataSourceInitializer` in the `PostgresChannelMessageTableSubscriberTests` to populate scripts * Use Java text block for DB scripts
This commit is contained in:
Rafael Winterhalter
committed by
Artem Bilan
Artem Bilan
parent
695e1563ed
commit
14d85977d7
@@ -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'
|
||||
}
|
||||
|
||||
@@ -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}.
|
||||
* <p/>
|
||||
* The supplied connection must <b>not</b> 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;
|
||||
|
||||
}
|
||||
@@ -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' <i>LISTEN</i>/<i>NOTIFY</i> 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.
|
||||
* <p/>
|
||||
* 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.
|
||||
* <p/>
|
||||
* 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 <i>schema-postgresql.sql</i> 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<String, Set<Subscription>> 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<Subscription> 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<Subscription> 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<Subscription> 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();
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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.
|
||||
* <p/>
|
||||
* 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 <i>schema-postgresql.sql</i> 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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
/**
|
||||
* Provides a message channel-specific JDBC API.
|
||||
*/
|
||||
package org.springframework.integration.jdbc.channel;
|
||||
@@ -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
|
||||
|
||||
@@ -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();*/
|
||||
|
||||
@@ -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<Object> 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<Object> 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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user