Adds a listener for Postgres-based channel message stores. #3884

raphw · 2022-09-08T08:36:21Z

Addresses #3872.

This listener allows for push notifications upon the arrival of new messages for a given group id and region.

raphw · 2022-09-08T08:46:45Z

I added this as a draft for creating a better environment for discussing this. I addressed your last comments, however not the point about the DataSource as it will most likely not work. If one used

dataSource.getConnection().unwrap(PgConnection.class)

for getting a connection, the DataSource would likely be a pooled one. These pools normally expect a connection to return to the pool after a given timeout, otherwise, they close the physical connection and report a connection leak. This would abort the listener, and repeat the process as a new connection was checked out. Therefore, a connection must normally be read directly from the driver, often via Driver.getConnection, that's why the interface is added.

artembilan · 2022-09-08T15:33:46Z

the DataSource would likely be a pooled one

So, you mean that the logic like this dataSource.getConnection().unwrap(PgConnection.class) is not not always correct because we cannot be sure what type of DataSource is provided.
If that is a case, then we must not tempt the fate and avoid problems as much as possible.
Therefore and approach with a Supplier<PgConnection> is the way to go - and we leave a choice up to end-user. And here already we can show a sample as this dataSource.getConnection().unwrap(PgConnection.class).

...on-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PgConnectionSupplier.java

artembilan · 2022-09-08T16:08:46Z

...va/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscription.java

+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+
+public final class PostgresChannelMessageTableSubscription {


I don't think that I can accept this abstraction:

It is probably better to have a JdbcChannelMessageStore as a ctor for the subscriber.

Have getter for it over there as well

The Postgres message channel should expect the subscriber as its ctor and take that store from there.
This way we would avoid the problem when subscriber and store look into different DBs or tables.

The subscription should be an inner interface of the subscriber to avoid classes tangle.

This way a region would be just an internal of the subscriber since both subscriber and channel would rely on the same store, therefore on the same region.

I don't believe in external Runnable onPossibleUpdate: it is really has to be an onUpdate() impl details. That's why I'm asking about a SubscribableChannel impl together with the mentioned subscription contract.

Somehow "subscriber" and "subscription" words don't work for me here: they are clashing with Reactive Streams. Not sure yet in other names though...

Not sure if I follow, you suggest to make the JdbcChannelMessageStore a constructor argument to the subscriber? Then it would no longer be possible to reuse the subscriber for different regions. This would require multiple connections to block for each region, would it not? Ideally, I wanted to avoid this.

Yeah... Sorry, I forgot that JdbcChannelMessageStore is restricted to specific region.
We will have to document it properly (and in JavaDocs as well), that JdbcChannelMessageStore on the channel we want to subscribe from and PgConnectionSupplier must belong to the same table.

Yeah, I think this will need some extensive javadoc, but I wanted to get the API right before I started on writing anything there. Due to the sharing requirement, the API will get slightly more complicated than it otherwise would have been, but I think it's worth that performance-wise.

raphw · 2022-09-08T20:07:01Z

the DataSource would likely be a pooled one

So, you mean that the logic like this dataSource.getConnection().unwrap(PgConnection.class) is not not always correct because we cannot be sure what type of DataSource is provided. If that is a case, then we must not tempt the fate and avoid problems as much as possible. Therefore and approach with a Supplier<PgConnection> is the way to go - and we leave a choice up to end-user. And here already we can show a sample as this dataSource.getConnection().unwrap(PgConnection.class).

For example Spring Boot would always auto-configure a data source where you could not use it without constant reconnects. And I have rarely seen unpooled data sources being used. I think that's the right approach.

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

...tegration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql

artembilan · 2022-09-09T14:19:38Z

...va/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscription.java

+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+
+public final class PostgresChannelMessageTableSubscription {


Yeah... Sorry, I forgot that JdbcChannelMessageStore is restricted to specific region.
We will have to document it properly (and in JavaDocs as well), that JdbcChannelMessageStore on the channel we want to subscribe from and PgConnectionSupplier must belong to the same table.

artembilan · 2022-09-09T14:23:26Z

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

+		}
+		// Avoid that stop/start sequence reactivates previously stopped thread.
+		Object current = new Object();
+		identity = current;


You just avoid this via making those start() and stop() as synchronized.

That would not work as the method returns before the executor thread might have ended. If the executor's thread dangles around, it could pick up a subsequent start. By switching the identity, it is certain that this additional listener should no longer listen or produce events, even if it takes the thread some time to complete.

We probably need to wait in the stop() until that thread is done with its loop before returning from the method. Something like CountDownLatch

I changed that to use a latch now.

artembilan · 2022-09-12T17:26:55Z

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

+		}
+		// Avoid that stop/start sequence reactivates previously stopped thread.
+		Object current = new Object();
+		identity = current;


We probably need to wait in the stop() until that thread is done with its loop before returning from the method. Something like CountDownLatch

artembilan · 2022-09-12T17:30:42Z

...tegration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql

+-- RETURNS TRIGGER AS
+-- $BODY$
+-- BEGIN
+--     EXECUTE 'NOTIFY int_channel_message_notify, ''' ||  QUOTE_LITERAL(NEW.REGION || NEW.GROUP_KEY) || '''';


Why there is no some readable concatenator between region and group?
What if I have them like this: myregion and mygroup ?
So, with this I'll end up as myregionmygroup. something what is not end-user friendly 😢

I thought that, since it's not exposed to the user, it does not matter much, but it's probably nice for debugging purposes. I added a space now.

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

...n-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java

...tegration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql

artembilan · 2022-09-13T14:14:21Z

...va/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscription.java

+		this.onPossibleUpdate.run();
+	}
+
+	public static PostgresChannelMessageTableSubscription ofMessageStore(


I'm still against this solution.
I'd like to see a subscription interface and some SubscribableChannel implementation.

I redid it to yield a SubscribableChannel. Of course, the implementation could be changed to only allow for this abstraction, but this would take away the possibility to being informed that there are new messages without polling them off the queue.

artembilan · 2022-09-13T19:15:18Z

...va/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscription.java

+		return result;
+	}
+
+	public static SubscribableChannel asSubscribableChannel(


Why do you still insist on this factory?
This is not Spring-friendly and I doubt anyone is going to use a subscription outside of the channel.
The PostgresSubscribableChannel must be a top-level API and no any extra classes for some onPossibleUpdate .
and I doubt in this onPossibleUpdate() name yet. Why just not notify() ?
What is wrong with extracting contract Subscription interface and implement it on the channel?
The PostgresChannelMessageTableSubscriber must be a ctor argument for that channel.
Also dispatching must be handled by the MessageDispatcher abstraction.

My intuition is to expose the most primitive abstraction on the base level, which is receiving a notification whenever a message was added to the group. If you wanted to use this information to read a message from a worker thread or directly, should be up to the user. I made a few changes now to expose a different API that always goes via a SubscribableChannel, if you'd prefer not introducing more API. I think this would satisfy the most common use.

artembilan

One more not forgot to say:

No reason to squash your commits all the time: we are going to do that on merge.
Meanwhile, during review and feedback process, it is better to keep commits history.
This way we know what was before we started and concluded some discussion.

artembilan

If you are not going to implement a MessageChannel as a first-class citizen for Spring Integration functionality and not going to add tests, I cannot accept the PR as it is right now and take it over for original expectations to follow up. We have some other priorities at the moment, so it may take some time until we incorporate this feature.

Thank you for understanding!

artembilan · 2022-09-13T20:14:57Z

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

@@ -0,0 +1,198 @@
+/*
+ * Copyright 2002-2022 the original author or authors.


This is a new class, so only the current year applies.

artembilan · 2022-09-13T20:16:00Z

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

+import java.sql.Statement;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.*;


No asterisk imports, please.
And also this is wrong order.
You simply can run :spring-integration-jdbc:check Gradle to see what is wrong with Checkstyle we require.

artembilan · 2022-09-13T20:53:27Z

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

+		return subscriptions.add(subscription);
+	}
+
+	boolean unsubscribe(PostgresChannelMessageTableSubscription subscription) {


I think these two methods must be public

artembilan · 2022-09-13T21:04:25Z

...va/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscription.java

+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.util.Assert;
+
+final class PostgresChannelMessageTableSubscription {


I still don't see this class as a Spring Integration structure.
Your original concern was exactly about a MessageChannel to be ready for notification from the PostgreSQL and there is no a MessageChannel as a top-level Spring Integration API in your PR.
It is fully not clean why you resist to follow a simple Subscription contract interface and a MessageChannel.
Plus I said before that without an inner interface in the subscriber there is going to be a class tangle, which you exactly have just introduced with that asSubscribableChannel() factory method.

There is also no tests yet for this PR.
We have in the jdbc module a MySqlContainerTest to test against MySQL container in Docker.
Probably we can have something similar for Postgres to be sure in the functionality you introduce.

I still don't understand what API you are looking for. Before that, I would not want to spend time on cleaning up the implementation or adding tests as this would require additional effort for every API suggestion. Once we find out that we're on the same page regarding the API, it's easy to add tests and clean up, that's why I set this PR as a draft.

I made the subscription class package-private to indicate that it's not the API I would go for but rather expose a factory for a SubscribableChannel. We have not discussed ChannelDispatcher before, only subscribable channel, would you rather have it as a factory for such a dispatcher?

What do you mean by "inner interface"? That the PostgresChannelMessageTableSubscriber should implement one? There's no interface in Spring integration that fits the necessary API, so I would need to create one. This goes back to what we discussed that it needs to be shared across regions and group ids as it fully occupies a physical Postgres connection.

artembilan · 2022-09-14T19:26:52Z

@raphw ,

my idea about an implementation and API is like this: artembilan@92de4ca

raphw · 2022-09-16T13:36:41Z

I have changed the implementation, cleaned up and added javadoc and a test.

The test is still failing, somehow the notification does never arrive at the listener. Is the setup so that it wraps everything in a separate transaction? It is really strange that the listen does not trigger here.

raphw · 2022-09-16T13:48:20Z

As an example: a very similar implementation using Testcontainers that works: https://github.com/Skatteetaten/skatteprosessen-task/blob/main/task-processor-jdbc/src/main/java/no/skatteetaten/fastsetting/formueinntekt/felles/task/processor/jdbc/PostgresProcessorNotifier.java

raphw · 2022-09-19T18:34:08Z

For some reason, Postgres does not like the inline concatenation when NOTIFY is used, but accepts it via the pg_notify function. The integration test is now passing and I would consider this a working solution.

artembilan

This is good progress.
This probably won't make it into today's release, but definitely next milestone.
Thank you!

...on-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PgConnectionSupplier.java

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

artembilan · 2022-09-20T16:55:15Z

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

+	 * @return {@code true} if the subscription was not already added.
+	 */
+	public boolean subscribe(Subscription subscription) {
+		Set<Subscription> subscriptions = this.subscriptions.computeIfAbsent(subscription.getRegion() + " " + getKey(subscription.getGroupId()), ignored -> ConcurrentHashMap.newKeySet());


We prefer code like length not more than 120 symbols.

...java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java

.../org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTest.java

...n-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresContainerTest.java

...ringframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTest-context.xml

.../org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTest.java

artembilan · 2022-09-20T17:11:30Z

.../org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTest.java

+					.isTrue();
+		}
+		finally {
+			postgresChannelMessageTableSubscriber.stop();


After making this as a bean, the ApplicationContext will take care about its lifecycle

I wanted to control the life cycle explicitly since one test checks a registration prior to starting the life cycle and one while the subscriber is already running.

.../org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTest.java

raphw · 2022-09-20T17:57:59Z

I addressed all comments but am left with not having the subscriber being registered as a bean to allow for two tests where subscribers are registered in a started and in a stopped condition.

artembilan

Good.
Would you mind to rebase your branch to the latest main and I'll it locally for final review.
Perhaps it may make it into today's release after all 😃 .

...n-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresContainerTest.java

… for a JdbcChannelMessageStore.

…ract * Implement a `PostgresSubscribableChannel`

raphw · 2022-09-20T18:37:16Z

I rebased on master.

artembilan · 2022-09-20T20:21:03Z

Merged as 14d8597.

Unfortunately I don't have time today to add docs for this new feature, but we are OK to go with then in the next milestone.

@raphw ,

thank you very much for the contribution; looking forward for more! For example those docs for this new feature 😄

raphw · 2022-09-20T21:20:26Z

Of course, thank you for your feedback along the way.

As for the documentation: #3890

raphw force-pushed the postgres-notification-listener branch from 41c6066 to c1a723f Compare September 8, 2022 08:41

raphw marked this pull request as draft September 8, 2022 08:41

artembilan requested changes Sep 8, 2022

View reviewed changes

artembilan requested changes Sep 9, 2022

View reviewed changes

raphw force-pushed the postgres-notification-listener branch from c1a723f to 0197628 Compare September 12, 2022 17:14

artembilan requested changes Sep 12, 2022

View reviewed changes

raphw force-pushed the postgres-notification-listener branch 10 times, most recently from 8f2b3db to 5d39c70 Compare September 12, 2022 21:49

artembilan requested changes Sep 13, 2022

View reviewed changes

raphw force-pushed the postgres-notification-listener branch 3 times, most recently from f0c3a32 to a6396ba Compare September 13, 2022 16:00

artembilan requested changes Sep 13, 2022

View reviewed changes

raphw force-pushed the postgres-notification-listener branch from a6396ba to ab121ed Compare September 13, 2022 20:11

artembilan requested changes Sep 13, 2022

View reviewed changes

raphw marked this pull request as ready for review September 19, 2022 18:34

raphw requested a review from artembilan September 19, 2022 18:34

artembilan requested changes Sep 20, 2022

View reviewed changes

raphw requested a review from artembilan September 20, 2022 17:58

artembilan requested changes Sep 20, 2022

View reviewed changes

...n-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresContainerTest.java Show resolved Hide resolved

raphw and others added 14 commits September 20, 2022 20:26

Adds an implementation for a Postres-compatible notification listener…

8148b49

… for a JdbcChannelMessageStore.

* Introduce PostgresChannelMessageTableSubscriber.Subscription cont…

0101d08

…ract * Implement a `PostgresSubscribableChannel`

Add Javadoc and fix code formatting issues.

2eef402

Handle temporary connection loss.

dce0651

Add tests for PostgresChannelMessageTableSubscriber.

3e4fd9f

Adjust tests.

3064435

Remove unused imports.

8f9ba92

Fix import order.

4b201b5

Remove unused transaction manager.

72d5fe2

Replace NOTIFY command with pg_notify function call.

848126b

Fix copyright year and add author tags.

f7ea100

Reformat and javdoc refinements.

7b3f635

Rewrite test to use Java configuration.

46dd2d3

Change name of test class.

a03d082

raphw force-pushed the postgres-notification-listener branch from ef54d8d to a03d082 Compare September 20, 2022 18:33

raphw requested a review from artembilan September 20, 2022 18:37

artembilan closed this Sep 20, 2022

joshiste mentioned this pull request Feb 2, 2023

Use postgres listen/notify for publish and subscribe. #4004

Open

		@@ -0,0 +1,198 @@
		/*
		* Copyright 2002-2022 the original author or authors.

Adds a listener for Postgres-based channel message stores. #3884

Adds a listener for Postgres-based channel message stores. #3884

Uh oh!

Conversation

raphw commented Sep 8, 2022

Uh oh!

raphw commented Sep 8, 2022

Uh oh!

artembilan commented Sep 8, 2022

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

raphw commented Sep 8, 2022

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

artembilan commented Sep 14, 2022

Uh oh!

raphw commented Sep 16, 2022

Uh oh!

raphw commented Sep 16, 2022

Uh oh!

raphw commented Sep 19, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

raphw commented Sep 19, 2022 •

edited

Loading