Start work on async listen

[fabianfett]

Motivation

We want to translate all our APIs to async.

Changes

• Add listen(_: String) -> PostgresNotificationSequence function

Result

Async listen is possible.

[pepicrft]

What's the state of this @fabianfett?

[fabianfett]

Hi @pepicrft, thanks for reaching out. I haven't touched this for quite a while and I'd be happy for someone else to drive this forward. Are you interested?

If that's the case I will write down my thoughts on this, to make getting started easier.

[pepicrft]

Hi @pepicrft, thanks for reaching out. I haven't touched this for quite a while and I'd be happy for someone else to drive this forward. Are you interested?

Very interested indeed. Would you brain-dumping everything in this issue?

[codecov-commenter]

Codecov Report

Merging #264 (e4e26c6) into main (4fd297d) will increase coverage by 3.00%.
The diff coverage is 66.01%.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #264      +/-   ##
==========================================
+ Coverage   43.88%   46.88%   +3.00%     
==========================================
  Files         107      110       +3     
  Lines        8596     8988     +392     
==========================================
+ Hits         3772     4214     +442     
+ Misses       4824     4774      -50     

Files Changed	Coverage Δ
Sources/PostgresNIO/New/PSQLTask.swift	33.33% <0.00%> (-2.67%)	⬇️
...rces/PostgresNIO/New/PostgresFrontendMessage.swift	53.93% <ø> (+8.98%)	⬆️
Sources/PostgresNIO/Postgres+PSQLCompat.swift	0.00% <0.00%> (ø)
Sources/PostgresNIO/New/PSQLError.swift	73.83% <25.00%> (-1.45%)	⬇️
...nection State Machine/ConnectionStateMachine.swift	56.48% <33.33%> (+0.15%)	⬆️
...es/PostgresNIO/Connection/PostgresConnection.swift	35.55% <38.18%> (+23.09%)	⬆️
.../Connection State Machine/ListenStateMachine.swift	68.45% <68.45%> (ø)
...urces/PostgresNIO/New/PostgresChannelHandler.swift	71.48% <73.23%> (+9.20%)	⬆️
Sources/PostgresNIO/New/NotificationListener.swift	73.95% <73.95%> (ø)
...PostgresNIO/New/PostgresNotificationSequence.swift	100.00% <100.00%> (ø)

... and 2 files with indirect coverage changes

[fabianfett]

Hi @pepicrft,

I sat down last night and tried to move this forward so that you can have an idea of where I want this to go. Please be well aware that the code is totally undertested and that likely only the happy path works. I'd really appreciate help here. However if you say this is already "too" done, and you are not interested in helping out anymore that is understandable. In that case you have a very bar bones implementation that you might be able to use.

But first let me discuss design goals:

1. We want to support listening to the same channel multiple times
2. We want to cancel listening on the connection (by telling the server UNLISTEN) when all listeners have cancelled
3. Users shall not be required to send LISTEN and UNLISTEN commands themselves to start the listening
4. If a connection dies all listeners shall be closed by throwing an error

Next let me explain, what types I've added and how all of this works together:

1. In PostgresConnection we now have an internalListenID atomic. We use this to generate unique ids for NotificationListeners. This is important to support correct cancellation.
2. The PostgresConnection now communicates with the PostgresChannelHandler using a HandlerTask. The HandlerTask is a superset of the existing PSQLTask and adds startListening and cancelListening
3. All state handling code that is required for Listening has moved from PostgresConnection into the PostgresChannelHandler
4. Inside the PostgresChannelHandler we use a ListenStateMachine to handle all state handling of listening to a given channel
5. The ListenStateMachine has a sub statemachine for each channel called ChannelState
6. The PostgresChannelHandler translates actions from the ListenStateMachine into tasks that need to be performed on the connection
   • makeStartListeningQuery
   • startListenCompleted
   • makeStopListeningQuery
   • forwardNotificationToListeners
7. The NotificationListener is a stateful reference type that represents the different states a single Listen command can be in. It is synchronized by using the connections EventLoop as synchronization primitive.
   • It supports the new API .streamInitialized, .streamListening
   • As well as the old existing interface .closure
8. We return an PostgresNotificationSequence which is a wrapper around an AsyncThrowingStream. We use the wrapper approach here to ensure that we can eventually back PostgresNotificationSequence with another AsyncSequence, without needing to change public API.

Feel free to use the commit in this PR and add further commits (by opening a new PR), that finalize the implementation and add tests. For an example on how to test a state machine, check out the ConnectionStateMachineTests in PostgresNIOTests.

How does all of this sound to you @pepicrft?

Small note: I'll be on vacation starting tomorrow until July 3. For this reason I might be slow to respond.

[pepicrft]

How does all of this sound to you @pepicrft?

Sounds perfect and thanks a lot for the thorough update on the state of things. I won't be able to dedicate myself full-time to it, but I hope to make some progress before you are back from vacation 😊.
I'll open a separate PR that we can sync on when you are back.

[pepicrft]

@fabianfett I got side-tracked with some work on Tuist and I won't be able to work on this 😞. I'll be following the progress made here though.

[gwynne]

A few minor nits, otherwise lgtm

[gwynne]

Should this be @Sendable?

[fabianfett]

API break checker reports a false positive:

1 breaking change detected in PostgresNIO:
  💔 API breakage: func PostgresConnection.addListener(channel:handler:) is now with @preconcurrency