* Introduce PostgresChannelMessageTableSubscriber.Subscription contract

* Implement a `PostgresSubscribableChannel`

Author: artembilan

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

@@ -50,7 +50,7 @@ public final class PostgresChannelMessageTableSubscriber implements SmartLifecyc
 	@Nullable
 	private volatile PgConnection connection;
 
-	private final Map<String, Set<PostgresChannelMessageTableSubscription>> subscriptions = new ConcurrentHashMap<>();
+	private final Map<String, Set<Subscription>> subscriptions = new ConcurrentHashMap<>();
 
 	public PostgresChannelMessageTableSubscriber(PgConnectionSupplier connectionSupplier) {
 		this(connectionSupplier, JdbcChannelMessageStore.DEFAULT_TABLE_PREFIX);
@@ -73,13 +73,13 @@ public SubscribableChannel toSubscribableChannel(
 		return PostgresChannelMessageTableSubscription.asSubscribableChannel(this, messageStore, groupId);
 	}
 
-	boolean subscribe(PostgresChannelMessageTableSubscription subscription) {
-		Set<PostgresChannelMessageTableSubscription> subscriptions = this.subscriptions.computeIfAbsent(subscription.getRegion() + " " + getKey(subscription.getGroupId()), ignored -> ConcurrentHashMap.newKeySet());
+	boolean subscribe(Subscription subscription) {
+		Set<Subscription> subscriptions = this.subscriptions.computeIfAbsent(subscription.getRegion() + " " + getKey(subscription.getGroupId()), ignored -> ConcurrentHashMap.newKeySet());
 		return subscriptions.add(subscription);
 	}
 
-	boolean unsubscribe(PostgresChannelMessageTableSubscription subscription) {
-		Set<PostgresChannelMessageTableSubscription> subscriptions = this.subscriptions.get(subscription.getRegion() + " " + getKey(subscription.getGroupId()));
+	boolean unsubscribe(Subscription subscription) {
+		Set<Subscription> subscriptions = this.subscriptions.get(subscription.getRegion() + " " + getKey(subscription.getGroupId()));
 		return subscriptions != null && subscriptions.remove(subscription);
 	}
 
@@ -127,12 +127,12 @@ public synchronized void start() {
 								if (notifications != null) {
 									for (PGNotification notification : notifications) {
 										String parameter = notification.getParameter();
-										Set<PostgresChannelMessageTableSubscription> subscriptions = this.subscriptions.get(parameter);
+										Set<Subscription> subscriptions = this.subscriptions.get(parameter);
 										if (subscriptions == null) {
 											continue;
 										}
-										for (PostgresChannelMessageTableSubscription subscription : subscriptions) {
-											subscription.onPossibleUpdate();
+										for (Subscription subscription : subscriptions) {
+											subscription.notifyUpdate();
 										}
 									}
 								}
@@ -195,4 +195,15 @@ public boolean isRunning() {
 	private String getKey(Object input) {
 		return input == null ? null : UUIDConverter.getUUID(input).toString();
 	}
+
+	public interface Subscription {
+
+		void notifyUpdate();
+
+		String getRegion();
+
+		Object getGroupId();
+
+	}
+
 }

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresSubscribableChannel.java

@@ -0,0 +1,85 @@
+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;
+
+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());
+
+	public PostgresSubscribableChannel(JdbcChannelMessageStore jdbcChannelMessageStore,
+			Object groupId, PostgresChannelMessageTableSubscriber messageTableSubscriber) {
+
+		this.jdbcChannelMessageStore = jdbcChannelMessageStore;
+		this.groupId = groupId;
+		this.messageTableSubscriber = messageTableSubscriber;
+	}
+
+	public void setDispatcherExecutor(Executor executor) {
+		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);
+			notifyUpdate();
+		}
+		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;
+	}
+
+}