Implement compression

Signed-off-by: Sri Krishna <[email protected]>

packages/connect/lib/connect.dart

@@ -18,6 +18,7 @@ export 'src/abort.dart';
 export 'src/client.dart';
 export 'src/code.dart';
 export 'src/codec.dart';
+export 'src/compression.dart';
 export 'src/exception.dart';
 export 'src/headers.dart';
 export 'src/http.dart';

packages/connect/lib/io.dart

@@ -12,4 +12,5 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+export 'src/gzip.dart';
 export 'src/io.dart';

packages/connect/lib/src/compression.dart

@@ -0,0 +1,22 @@
+// Copyright 2024 The Connect Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import 'dart:convert';
+import 'dart:typed_data';
+
+/// Compression is specialized [Codec] for a certain compression algorithm.
+abstract interface class Compression implements Codec<Uint8List, Uint8List> {
+  /// Name of the compression algorithm.
+  abstract final String name;
+}

packages/connect/lib/src/connect/header.dart

@@ -14,6 +14,7 @@
 
 import '../abort.dart';
 import '../codec.dart';
+import '../compression.dart';
 import '../headers.dart';
 import '../spec.dart';
 import 'version.dart';
@@ -22,12 +23,18 @@ const headerContentType = "content-type";
 const headerContentLength = "content-length";
 const headerTimeout = "connect-timeout-ms";
 const headerProtocolVersion = "connect-protocol-version";
+const headerUnaryEncoding = "content-encoding";
+const headerStreamEncoding = "connect-content-encoding";
+const headerUnaryAcceptEncoding = "accept-encoding";
+const headerStreamAcceptEncoding = "connect-accept-encoding";
 
 Headers requestHeader(
   Codec codec,
   StreamType streamType,
   Headers? userProvidedHeaders,
   AbortSignal? signal,
+  Compression? sendCompression,
+  List<Compression> acceptCompressions,
 ) {
   final header = userProvidedHeaders == null
       ? Headers()
@@ -41,5 +48,16 @@ Headers requestHeader(
       ? 'application/${codec.name}'
       : 'application/connect+${codec.name}';
   // TODO: User agent headers.
+  if (sendCompression != null) {
+    header[streamType == StreamType.unary
+        ? headerUnaryEncoding
+        : headerStreamEncoding] = sendCompression.name;
+  }
+  if (acceptCompressions.isNotEmpty) {
+    header[streamType == StreamType.unary
+            ? headerUnaryAcceptEncoding
+            : headerStreamAcceptEncoding] =
+        acceptCompressions.map((c) => c.name).join(",");
+  }
   return header;
 }

packages/connect/lib/src/connect/protocol.dart

@@ -14,6 +14,7 @@
 
 import '../code.dart';
 import '../codec.dart';
+import '../compression.dart';
 import '../exception.dart';
 import '../headers.dart';
 import '../http.dart';
@@ -42,13 +43,17 @@ final class Protocol implements base.Protocol {
   Headers requestHeaders<I, O>(
     Spec<I, O> spec,
     Codec codec,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
     CallOptions? options,
   ) {
     return requestHeader(
       codec,
       spec.streamType,
       options?.headers,
       options?.signal,
+      sendCompression,
+      acceptCompressions,
     );
   }
 
@@ -57,6 +62,8 @@ final class Protocol implements base.Protocol {
     UnaryRequest<I, O> req,
     Codec codec,
     HttpClient httpClient,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
   ) async {
     final HttpRequest hReq;
     if (useHttpGet &&
@@ -73,20 +80,28 @@ final class Protocol implements base.Protocol {
         req.signal,
       );
     } else {
+      final body = codec.encode(req.message);
       hReq = HttpRequest(
         req.url,
         "POST",
         req.headers,
-        Stream.fromIterable([codec.encode(req.message)]),
+        Stream.fromIterable([sendCompression?.encode(body) ?? body]),
         req.signal,
       );
     }
     final res = await httpClient(hReq);
-    final unaryError = res.validate(req.spec.streamType, codec);
+    final (:compression, :unaryError) = res.validate(
+      req.spec.streamType,
+      codec,
+      acceptCompressions,
+    );
     final (:headers, :trailers) = res.header.demux();
-    final body = await res.body.toBytes(
+    var body = await res.body.toBytes(
       int.tryParse(headers[headerContentLength] ?? ''),
     );
+    if (compression != null) {
+      body = compression.decode(body);
+    }
     if (unaryError != null) {
       throw errorFromJsonBytes(
         body,
@@ -107,6 +122,8 @@ final class Protocol implements base.Protocol {
     StreamRequest<I, O> req,
     Codec codec,
     HttpClient httpClient,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
   ) async {
     final res = await httpClient(
       HttpRequest(
@@ -117,14 +134,15 @@ final class Protocol implements base.Protocol {
         req.signal,
       ),
     );
-    res.validate(req.spec.streamType, codec);
+    final (:compression, unaryError: _) =
+        res.validate(req.spec.streamType, codec, acceptCompressions);
     final trailer = Headers();
     return StreamResponse(
       req.spec,
       res.header,
       res.body
           .splitEnvelope()
-          .decompress()
+          .decompress(compression)
           .parse(
             codec,
             req.spec.outputFactory,

packages/connect/lib/src/connect/response.dart

@@ -14,9 +14,11 @@
 
 import '../code.dart';
 import '../codec.dart';
+import '../compression.dart';
 import '../exception.dart';
 import '../headers.dart';
 import '../http.dart';
+import '../protocol/compression.dart';
 import '../spec.dart';
 import 'header.dart';
 import 'http_status.dart';
@@ -28,10 +30,17 @@ extension ResponseValidation on HttpResponse {
   /// For unary RPCs with an HTTP error status, this returns an error
   /// derived from the HTTP status instead of throwing it, giving an
   /// implementation a chance to parse a Connect error from the wire.
-  ConnectException? validate(
+  ({ConnectException? unaryError, Compression? compression}) validate(
     StreamType streamType,
     Codec codec,
+    List<Compression> acceptCompressions,
   ) {
+    final compression = findCompression(
+      acceptCompressions,
+      streamType == StreamType.unary
+          ? headerUnaryEncoding
+          : headerStreamEncoding,
+    );
     final contentType = header[headerContentType] ?? '';
     if (status != 200) {
       final statusErr = ConnectException(
@@ -41,7 +50,7 @@ extension ResponseValidation on HttpResponse {
       if (streamType == StreamType.unary &&
           contentType.startsWith("application/json")) {
         // Unary JSON response
-        return statusErr;
+        return (unaryError: statusErr, compression: compression);
       }
       throw statusErr;
     }
@@ -58,6 +67,6 @@ extension ResponseValidation on HttpResponse {
         metadata: header,
       );
     }
-    return null;
+    return (unaryError: null, compression: compression);
   }
 }

packages/connect/lib/src/connect/transport.dart

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 import "../codec.dart";
+import "../compression.dart";
 import "../http.dart";
 import "../interceptor.dart";
 import "../protocol/transport.dart";
@@ -26,11 +27,15 @@ final class Transport extends ProtocolTransport {
     required HttpClient httpClient,
     List<Interceptor>? interceptors,
     bool useHttpGet = false,
+    Compression? sendCompression,
+    List<Compression>? acceptCompressions,
   }) : super(
           baseUrl,
           codec,
           Protocol(useHttpGet),
           httpClient,
           interceptors ?? [],
+          sendCompression,
+          acceptCompressions ?? [],
         );
 }

packages/connect/lib/src/fake.dart

@@ -17,6 +17,7 @@ import 'dart:typed_data';
 
 import 'abort.dart';
 import 'codec.dart';
+import 'compression.dart';
 import 'headers.dart';
 import 'http.dart';
 import 'interceptor.dart';
@@ -155,6 +156,8 @@ final class _FakeTransport extends ProtocolTransport {
           _FakeProtocol(handlers),
           (req) => throw UnimplementedError(),
           interceptors ?? [],
+          null,
+          [],
         );
 }
 
@@ -167,6 +170,8 @@ final class _FakeProtocol implements Protocol {
   Headers requestHeaders<I, O>(
     Spec<I, O> spec,
     Codec codec,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
     CallOptions? options,
   ) {
     final headers = Headers();
@@ -181,6 +186,8 @@ final class _FakeProtocol implements Protocol {
     UnaryRequest<I, O> req,
     Codec codec,
     HttpClient client,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
   ) async {
     final handler = handlers[req.spec];
     if (handler == null) {
@@ -211,6 +218,8 @@ final class _FakeProtocol implements Protocol {
     StreamRequest<I, O> req,
     Codec codec,
     HttpClient client,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
   ) async {
     final handler = handlers[req.spec];
     if (handler == null) {

packages/connect/lib/src/grpc/protocol.dart

@@ -14,6 +14,7 @@
 
 import '../code.dart';
 import '../codec.dart';
+import '../compression.dart';
 import '../exception.dart';
 import '../headers.dart';
 import '../http.dart';
@@ -32,37 +33,54 @@ import 'trailer.dart';
 
 final class Protocol implements base.Protocol {
   final StatusParser statusParser;
+  final Compression? sendCompression;
 
-  const Protocol(this.statusParser);
+  const Protocol(this.statusParser, this.sendCompression);
 
   @override
   Headers requestHeaders<I, O>(
     Spec<I, O> spec,
     Codec codec,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
     CallOptions? options,
   ) {
-    return requestHeader(codec, options?.headers, options?.signal);
+    return requestHeader(
+      codec,
+      options?.headers,
+      options?.signal,
+      sendCompression,
+      acceptCompressions,
+    );
   }
 
   @override
   Future<UnaryResponse<I, O>> unary<I extends Object, O extends Object>(
     UnaryRequest<I, O> req,
     Codec codec,
     HttpClient httpClient,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
   ) async {
     final res = await httpClient(
       HttpRequest(
         req.url,
         "POST",
         req.headers,
-        Stream.fromIterable([encodeEnvelope(0, codec.encode(req.message))]),
+        Stream.fromIterable([req.message])
+            .serialize(codec)
+            .compress(sendCompression)
+            .joinEnvelope(),
         req.signal,
       ),
     );
-    final (foundStatus: _, :headerError) = res.validate(statusParser);
+    final (foundStatus: _, :headerError, :compression) = res.validate(
+      statusParser,
+      acceptCompressions,
+    );
     final message = await res.body
         .splitEnvelope()
-        .decompress()
+        .decompress(compression)
         .parse(codec, req.spec.outputFactory)
         .tryReadingSingleMessage();
     res.trailer.validateTrailer(res.header, statusParser);
@@ -98,17 +116,22 @@ final class Protocol implements base.Protocol {
     StreamRequest<I, O> req,
     Codec codec,
     HttpClient httpClient,
+    Compression? sendCompression,
+    List<Compression> acceptCompressions,
   ) async {
     final res = await httpClient(
       HttpRequest(
         req.url,
         "POST",
         req.headers,
-        req.message.serialize(codec).joinEnvelope(),
+        req.message.serialize(codec).compress(sendCompression).joinEnvelope(),
         req.signal,
       ),
     );
-    final (:foundStatus, :headerError) = res.validate(statusParser);
+    final (:foundStatus, :headerError, :compression) = res.validate(
+      statusParser,
+      acceptCompressions,
+    );
     if (headerError != null) {
       // Trailers only response.
       throw headerError;
@@ -118,7 +141,7 @@ final class Protocol implements base.Protocol {
       res.header,
       res.body
           .splitEnvelope()
-          .decompress()
+          .decompress(compression)
           .parse(codec, req.spec.outputFactory)
           .onDone(() {
         if (!foundStatus) {