Add replication mode flag (v3) (#124)

Author: osaxma

lib/postgres_v3_experimental.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:io';
 
 import 'package:collection/collection.dart';
+import 'package:postgres/src/replication.dart';
 import 'package:stream_channel/stream_channel.dart';
 
 import 'src/v3/connection.dart';
@@ -359,11 +360,26 @@ final class PgSessionSettings {
   /// For an example, see `example/v3/transformer.dart`.
   final StreamChannelTransformer<BaseMessage, BaseMessage>? transformer;
 
+  /// The replication mode for connecting in streaming replication mode.
+  ///
+  /// The default value is [ReplicationMode.none]. But when the value is set to 
+  /// [ReplicationMode.physical] or [ReplicationMode.logical], the connection 
+  /// will be established in replication mode.
+  ///
+  /// Please note, while in replication mode, only the Simple Query Protcol can
+  /// be used to execute queries. 
+  ///
+  /// For more info, see [Streaming Replication Protocol]
+  ///
+  /// [Streaming Replication Protocol]: https://www.postgresql.org/docs/current/protocol-replication.html
+  final ReplicationMode replicationMode;
+
   PgSessionSettings({
     this.connectTimeout,
     this.timeZone,
     this.onBadSslCertificate,
     this.transformer,
+    this.replicationMode = ReplicationMode.none
   });
 }
 

lib/src/v3/connection.dart

@@ -8,6 +8,7 @@ import 'package:collection/collection.dart';
 import 'package:pool/pool.dart';
 import 'package:postgres/postgres_v3_experimental.dart';
 import 'package:postgres/src/query.dart';
+import 'package:postgres/src/replication.dart';
 import 'package:stream_channel/stream_channel.dart';
 
 import '../auth/auth.dart';
@@ -37,6 +38,8 @@ class _ResolvedSettings {
   final String timeZone;
   //final Encoding encoding;
 
+  final ReplicationMode replicationMode;
+
   final StreamChannelTransformer<BaseMessage, BaseMessage>? transformer;
 
   _ResolvedSettings(
@@ -49,7 +52,8 @@ class _ResolvedSettings {
         //queryTimeout = settings?.connectTimeout ?? const Duration(minutes: 5),
         timeZone = settings?.timeZone ?? 'UTC',
         // encoding = settings?.encoding ?? utf8,
-        transformer = settings?.transformer;
+        transformer = settings?.transformer,
+        replicationMode = settings?.replicationMode ?? ReplicationMode.none;
 
   bool onBadSslCertificate(X509Certificate certificate) {
     return settings?.onBadSslCertificate?.call(certificate) ?? false;
@@ -304,7 +308,7 @@ class PgConnectionImplementation extends _PgSessionBase
         _settings.endpoint.database,
         _settings.timeZone,
         username: _settings.username,
-        // todo: Replication
+        replication: _settings.replicationMode,
       ));
 
       return result._done.future;