Using transformers to send messages to the server in V3

[osaxma]

This post is intended to discuss the transformer/ API (i.e. PgSessionSettings.transformer). The goal is to share how the API is being used as well as to get feedback if there are better approaches.

Some Context:
In Streaming Replication mode, once a connection is successfully established with the server, the server will start streaming data to the client. While the majority of the messages would be just data (i.e. changes) that don't require a response, other messages require a response from the client (i.e. PrimaryKeepAliveMessage).

Example Usage:
While updating the postgresql-dart-replication-example to support version 3, this is how I ended up utilizing the transformer API to send and receive server messages:

/// A "Channel" that can send and receive messages to and from the Server.
class _ServerChannel {
  final controller = StreamController<ServerMessage>.broadcast();

  Stream<ServerMessage> get messages => controller.stream;

  EventSink<BaseMessage>? _cachedSink;

  void addMessage(BaseMessage message) {
    if (_cachedSink == null) {
      throw Exception('Cannot send message because sink is not available');
    }
    _cachedSink!.add(message);
  }

  late final transformer = StreamChannelTransformer<BaseMessage, BaseMessage>(
    // to listen to server messages
    StreamTransformer.fromHandlers(
      handleData: (data, sink) {
        if (!controller.isClosed) {
          controller.add(data as ServerMessage);
        }
        // let the message continue its journey
        sink.add(data);
      },
    ),
    // this is intended to listen to client messages
    // but we capture the sink and cache it so we can send messages
    // to the server when needed.
    StreamSinkTransformer.fromHandlers(
      handleData: (data, sink) {
        if (_cachedSink != sink) {
          _cachedSink = sink;
        }
        // let the message continue its journey
        sink.add(data);
      },
      handleDone: (_) {
        _cachedSink = null;
      },
      handleError: (_, __, ___) {
        _cachedSink = null;
        // handle the error ...
      },
    ),
  );
}

_full example in the following file of the v3 example branch:
https://github.com/osaxma/postgresql-dart-replication-example/blob/main/example/listen_v3.dart

Discussion:
While listening to server messages is an appropriate use of the API, caching the sink feels a hacky solution since this does not seem how the API is supposed to be used (i.e. as messages interceptor).

For comparison, in V2, PostgreSQLConnection.addMessage was an available straightforward method to send messages to the server. Similarly, PostgreSQLConnection.messages was the way to listen to server messages.

While I don't mind using the approach above, caching the sink just didn't feel right. Maybe I am just overthinking. Anyway, I thought it's better to bring this up now than later.

Cheers

cc @simolus3 @isoos

[simolus3]

What I like about the stream transformer API is that it's very flexible, you can essentially implement any "man-in-the-middle" transformation on the connection to the postgres server.

Similarly, PostgreSQLConnection.messages was the way to listen to server messages.

But on that stream, there is no way to "intercept" messages (so that they're not seen by a PgConnection at all), right? For safety, I think it's reasonable that PgConnection now shuts down the connection when it receives an unexpected message since that's likely a bug in the client. (actually we're not doing that at the moment) When replication is enabled with addMessage and message, the client would also see the unexpected messages. By implementing them in a transformer, the client doesn't care about that at all.

caching the sink just didn't feel right. Maybe I am just overthinking

If you only need to react to incoming messages from the server by either forwarding them to the client or by responding directly, how about this:

typedef InterceptedMessage = ({bool forward, BaseMessage? response});
typedef ChannelInterceptor = InterceptedMessage Function(
    BaseMessage fromServer);

extension on StreamChannel<BaseMessage> {
  StreamChannel<BaseMessage> intercept(ChannelInterceptor interceptor) {
    return changeStream((oldStream) async* {
      await for (final originalMessage in oldStream) {
        final intercepted = interceptor(originalMessage);

        if (intercepted.response case final BaseMessage response) {
          // Respond directly to server
          sink.add(response);
        }

        if (intercepted.forward) {
          yield originalMessage;
        }
      }
    });
  }
}

Since the sink isn't changed, it doesn't need to be cached either. More complex transformations could likely be implemented as well by making the sink a final field that is set to channel.sink before any transformation.

[simolus3]

I've just seen the full example, sorry for not checking that directly. Still, I think a general transformer for that can be easier to write:

class ExposeConnection
    implements StreamChannelTransformer<BaseMessage, BaseMessage> {
  final Completer<StreamChannel<BaseMessage>> _completer = Completer();

  Future<StreamChannel<BaseMessage>> get channel => _completer.future;

  @override
  StreamChannel<BaseMessage> bind(StreamChannel<BaseMessage> channel) {
    final broadcast =
        channel.changeStream((stream) => stream.asBroadcastStream());
    _completer.complete(broadcast);
    return broadcast;
  }
}

void main(List<String> arguments) async {
  final expose = ExposeConnection();

  final conn = await PgConnection.open(
    PgEndpoint(...),
    sessionSettings: PgSessionSettings(
      replicationMode: ReplicationMode.logical,
      queryMode: QueryMode.simple,
      onBadSslCertificate: (cert) => true,
      transformer: expose,
      encoding: utf8,
    ),
  );

  final channel = await expose.channel;

  late LSN clientXLogPos;
  final messagesSub = channel.stream.listen((msg) {

To send a message to the server, you'd do channel.sink.add.

[osaxma]

Hi @simolus3

I actually like the transformer API -- I just didn't know how to use it as eloquently as you presented here.

I just plugged the sample code you provided in the comment -- it's much neater now.

Thanks a lot 🙏🏽