Preparing Type refactor. (#238)

Author: isoos

lib/postgres.dart

@@ -3,11 +3,11 @@ import 'dart:convert';
 
 import 'package:collection/collection.dart';
 import 'package:meta/meta.dart';
-import 'package:postgres/src/types/type_registry.dart';
 import 'package:stream_channel/stream_channel.dart';
 
 import 'src/replication.dart';
 import 'src/types.dart';
+import 'src/types/type_registry.dart';
 import 'src/v3/connection.dart';
 import 'src/v3/protocol.dart';
 import 'src/v3/query_description.dart';
@@ -16,6 +16,7 @@ export 'src/exceptions.dart';
 export 'src/pool/pool_api.dart';
 export 'src/replication.dart';
 export 'src/types.dart';
+export 'src/types/type_registry.dart' show TypeOid, TypeRegistry;
 
 /// A description of a SQL query as interpreted by this package.
 ///

lib/src/messages/client_messages.dart

@@ -172,7 +172,7 @@ class DescribeMessage extends ClientMessage {
 }
 
 class BindMessage extends ClientMessage {
-  final List<TypedValue> _parameters;
+  final List<EncodeOutput?> _parameters;
   final String _portalName;
   final String _statementName;
 
@@ -189,19 +189,12 @@ class BindMessage extends ClientMessage {
     final portalName = buffer.encodeString(_portalName);
     final statementName = buffer.encodeString(_statementName);
 
-    final encodedOutputs = _parameters
-        .map((p) => p.value == null
-            ? null
-            : p.type.encode(
-                EncodeInput(value: p.value!, encoding: buffer.encoding)))
-        .toList();
-    final encodedBytes = encodedOutputs.map((e) {
+    final encodedBytes = _parameters.map((e) {
       if (e == null) return null;
       return e.isBinary ? e.bytes! : buffer.encoding.encode(e.text!);
     }).toList();
-    final binaryCount =
-        encodedOutputs.where((p) => p?.isBinary ?? false).length;
-    final isAllBinary = binaryCount == encodedOutputs.length;
+    final binaryCount = _parameters.where((p) => p?.isBinary ?? false).length;
+    final isAllBinary = binaryCount == _parameters.length;
     final isAllText = binaryCount == 0;
     final inputParameterElementCount =
         isAllBinary || isAllText ? 1 : _parameters.length;
@@ -232,7 +225,7 @@ class BindMessage extends ClientMessage {
     } else {
       // Well, we have some text and some binary, so we have to be explicit about each one
       buffer.writeUint16(_parameters.length);
-      for (final p in encodedOutputs) {
+      for (final p in _parameters) {
         buffer.writeUint16(p != null && p.isBinary
             ? ClientMessageFormat.binary
             : ClientMessageFormat.text);

lib/src/types/text_codec.dart

@@ -8,6 +8,14 @@ class PostgresTextEncoder {
   const PostgresTextEncoder();
 
   String convert(Object input, {bool escapeStrings = true}) {
+    final value = tryConvert(input, escapeStrings: escapeStrings);
+    if (value != null) {
+      return value;
+    }
+    throw PgException("Could not infer type of value '$input'.");
+  }
+
+  String? tryConvert(Object input, {bool escapeStrings = false}) {
     if (input is int) {
       return _encodeNumber(input);
     }
@@ -40,9 +48,7 @@ class PostgresTextEncoder {
       return _encodeList(input);
     }
 
-    // TODO: use custom type encoders
-
-    throw PgException("Could not infer type of value '$input'.");
+    return null;
   }
 
   String _encodeString(String text, bool escapeStrings) {

lib/src/types/type_registry.dart

@@ -1,9 +1,15 @@
+import 'dart:convert';
+import 'dart:typed_data';
+
 import 'package:meta/meta.dart';
 
 import '../types.dart';
 
 import 'generic_type.dart';
 
+typedef TypeEncoderFn = EncodeOutput Function(EncodeInput input);
+typedef TypeDecoderFn = Object? Function(DecodeInput input);
+
 /// See: https://github.com/postgres/postgres/blob/master/src/include/catalog/pg_type.dat
 class TypeOid {
   static const bigInteger = 20;
@@ -110,3 +116,32 @@ class TypeRegistry {
   @internal
   Iterable<Type> get registered => _byTypeOid.values;
 }
+
+extension TypeRegistryExt on TypeRegistry {
+  EncodeOutput? encodeValue(
+    Object? value, {
+    Type? type,
+    required Encoding encoding,
+  }) {
+    if (value == null) return null;
+    return type?.encode(EncodeInput(value: value, encoding: encoding));
+  }
+
+  Object? decodeBytes(
+    Uint8List? bytes, {
+    required int typeOid,
+    required bool isBinary,
+    required Encoding encoding,
+  }) {
+    if (bytes == null) {
+      return null;
+    }
+    final type = resolveOid(typeOid);
+    return type.decode(DecodeInput(
+      bytes: bytes,
+      isBinary: isBinary,
+      encoding: encoding,
+      typeRegistry: this,
+    ));
+  }
+}

lib/src/v2/query.dart

@@ -91,7 +91,8 @@ class Query<T> {
     final messages = [
       ParseMessage(sqlString, statementName: statementName),
       DescribeMessage(statementName: statementName),
-      BindMessage(parameterList, statementName: statementName),
+      BindMessage(_toEncodedOutput(parameterList),
+          statementName: statementName),
       ExecuteMessage(),
       SyncMessage(),
     ];
@@ -104,6 +105,17 @@ class Query<T> {
         .add(_aggregateClientMessages(messages, encoding: connection.encoding));
   }
 
+  late final _typeRegistry = TypeRegistry();
+  List<EncodeOutput?> _toEncodedOutput(List<ParameterValue> values) {
+    return values
+        .map((e) => _typeRegistry.encodeValue(
+              e.value,
+              type: e.type,
+              encoding: connection.encoding,
+            ))
+        .toList();
+  }
+
   void sendCachedQuery(Socket socket, CachedQuery cacheQuery,
       Map<String, dynamic>? substitutionValues) {
     final statementName = cacheQuery.preparedStatementName;
@@ -114,7 +126,8 @@ class Query<T> {
 
     final bytes = _aggregateClientMessages(
       [
-        BindMessage(parameterList, statementName: statementName!),
+        BindMessage(_toEncodedOutput(parameterList),
+            statementName: statementName!),
         ExecuteMessage(),
         SyncMessage()
       ],

lib/src/v3/connection.dart

@@ -7,6 +7,7 @@ import 'package:async/async.dart' as async;
 import 'package:charcode/ascii.dart';
 import 'package:meta/meta.dart';
 import 'package:pool/pool.dart' as pool;
+import 'package:postgres/src/types/type_registry.dart';
 import 'package:stream_channel/stream_channel.dart';
 
 import '../../postgres.dart';
@@ -568,7 +569,13 @@ class _PgResultStreamSubscription
 
       connection._channel.sink.add(AggregatedClientMessage([
         BindMessage(
-          statement.parameters,
+          statement.parameters
+              .map((e) => connection._settings.typeRegistry.encodeValue(
+                    e.value,
+                    type: e.type,
+                    encoding: connection.encoding,
+                  ))
+              .toList(),
           portalName: _portalName,
           statementName: statement.statement._name,
         ),
@@ -674,16 +681,14 @@ class _PgResultStreamSubscription
           for (var i = 0; i < message.values.length; i++) {
             final field = schema.columns[i];
 
-            final type = field.type;
             final input = message.values[i];
-            final value = input == null
-                ? null
-                : type.decode(DecodeInput(
-                    bytes: input,
-                    isBinary: field.isBinaryEncoding,
-                    encoding: session.encoding,
-                    typeRegistry: session._connection._settings.typeRegistry,
-                  ));
+            final value =
+                session._connection._settings.typeRegistry.decodeBytes(
+              input,
+              typeOid: field.typeOid,
+              isBinary: field.isBinaryEncoding,
+              encoding: session.encoding,
+            );
             columnValues.add(value);
           }
 

lib/src/v3/resolved_settings.dart

@@ -1,7 +1,6 @@
 import 'dart:convert';
 
 import 'package:postgres/messages.dart';
-import 'package:postgres/src/types/type_registry.dart';
 import 'package:stream_channel/stream_channel.dart';
 
 import '../../postgres.dart';

test/decode_test.dart

@@ -4,7 +4,6 @@ import 'package:buffer/buffer.dart';
 import 'package:postgres/legacy.dart';
 import 'package:postgres/postgres.dart';
 import 'package:postgres/src/types/binary_codec.dart';
-import 'package:postgres/src/types/type_registry.dart';
 import 'package:test/test.dart';
 
 import 'docker.dart';

test/encoding_test.dart

@@ -6,7 +6,6 @@ import 'package:postgres/legacy.dart';
 import 'package:postgres/postgres.dart';
 import 'package:postgres/src/types/binary_codec.dart';
 import 'package:postgres/src/types/text_codec.dart';
-import 'package:postgres/src/types/type_registry.dart';
 import 'package:test/test.dart';
 
 import 'docker.dart';

test/interpolation_test.dart

@@ -2,7 +2,6 @@ import 'dart:convert';
 
 import 'package:postgres/legacy.dart';
 import 'package:postgres/postgres.dart';
-import 'package:postgres/src/types/type_registry.dart';
 import 'package:postgres/src/v2/query.dart';
 import 'package:test/test.dart';