default to identity header everywhere instead of leaving it undefined

Author: b0b3rt

packages/grpc-js/src/compression-filter.ts

@@ -183,7 +183,7 @@ function getCompressionHandler(compressionName: string): CompressionHandler {
 export class CompressionFilter extends BaseFilter implements Filter {
   private sendCompression: CompressionHandler = new IdentityHandler();
   private receiveCompression: CompressionHandler = new IdentityHandler();
-  private defaultCompressionAlgorithm: CompressionAlgorithm | undefined;
+  private currentCompressionAlgorithm: CompressionAlgorithm = 'identity';
 
   constructor(channelOptions: ChannelOptions, private sharedFilterConfig: SharedCompressionFilterConfig) {
     super();
@@ -201,8 +201,8 @@ export class CompressionFilter extends BaseFilter implements Filter {
          *    In that case we only want to use the encoding chosen by the client if the server supports it
          */
         if (!serverSupportedEncodings || serverSupportedEncodings.includes(clientSelectedEncoding)) {
-          this.defaultCompressionAlgorithm = clientSelectedEncoding;
-          this.sendCompression = getCompressionHandler(this.defaultCompressionAlgorithm);
+          this.currentCompressionAlgorithm = clientSelectedEncoding;
+          this.sendCompression = getCompressionHandler(this.currentCompressionAlgorithm);
         }
       } else {
         logging.log(LogVerbosity.ERROR, `Invalid value provided for grpc.default_compression_algorithm option: ${compressionAlgorithmKey}`);
@@ -215,10 +215,11 @@ export class CompressionFilter extends BaseFilter implements Filter {
     headers.set('grpc-accept-encoding', 'identity,deflate,gzip');
     headers.set('accept-encoding', 'identity');
 
-    if (this.defaultCompressionAlgorithm && ['deflate', 'gzip'].includes(this.defaultCompressionAlgorithm)) {
-      headers.set('grpc-encoding', this.defaultCompressionAlgorithm);
-    } else {
+    // No need to send the header if it's "identity" -  behavior is identical; save the bandwidth
+    if (this.currentCompressionAlgorithm === 'identity') {
       headers.remove('grpc-encoding');
+    } else {
+      headers.set('grpc-encoding', this.currentCompressionAlgorithm);
     }
 
     return headers;
@@ -255,10 +256,13 @@ export class CompressionFilter extends BaseFilter implements Filter {
      * and the output is a framed and possibly compressed message. For this
      * reason, this filter should be at the bottom of the filter stack */
     const resolvedMessage: WriteObject = await message;
-    const compress =
-      resolvedMessage.flags === undefined
-        ? !(this.sendCompression instanceof IdentityHandler)
-        : (resolvedMessage.flags & WriteFlags.NoCompress) === 0;
+    let compress: boolean;
+    if (this.sendCompression instanceof IdentityHandler) {
+      compress = false;
+    } else {
+      compress = ((resolvedMessage.flags ?? 0) & WriteFlags.NoCompress) === 0;
+    }
+
     return {
       message: await this.sendCompression.writeMessage(
         resolvedMessage.message,

packages/grpc-js/src/server-call.ts

@@ -138,7 +138,7 @@ export class ServerReadableStreamImpl<RequestType, ResponseType>
     private call: Http2ServerCallStream<RequestType, ResponseType>,
     public metadata: Metadata,
     public deserialize: Deserialize<RequestType>,
-    encoding?: string
+    encoding: string
   ) {
     super({ objectMode: true });
     this.cancelled = false;
@@ -253,7 +253,7 @@ export class ServerDuplexStreamImpl<RequestType, ResponseType>
     public metadata: Metadata,
     public serialize: Serialize<ResponseType>,
     public deserialize: Deserialize<RequestType>,
-    encoding?: string
+    encoding: string
   ) {
     super({ objectMode: true });
     this.cancelled = false;
@@ -442,7 +442,7 @@ export class Http2ServerCallStream<
     return this.cancelled;
   }
 
-  private getDecompressedMessage(message: Buffer, encoding?: string) {
+  private getDecompressedMessage(message: Buffer, encoding: string) {
     switch (encoding) {
       case 'deflate': {
         return new Promise<Buffer | undefined>((resolve, reject) => {
@@ -477,9 +477,18 @@ export class Http2ServerCallStream<
           });
         });
       }
-  
-      default:
+
+      case 'identity': {
         return Promise.resolve(message);
+      }
+  
+      default: {
+        this.sendError({
+          code: Status.UNIMPLEMENTED,
+          details: `Received "grpc-encoding" header "${encoding}", which is not supported`,
+        });
+        return Promise.resolve();
+      }
     }
   }
 
@@ -533,7 +542,7 @@ export class Http2ServerCallStream<
     return metadata;
   }
 
-  receiveUnaryMessage(encoding?: string): Promise<RequestType> {
+  receiveUnaryMessage(encoding: string): Promise<RequestType> {
     return new Promise((resolve, reject) => {
       const stream = this.stream;
       const chunks: Buffer[] = [];
@@ -561,7 +570,7 @@ export class Http2ServerCallStream<
           this.emit('receiveMessage');
 
           const compressed = requestBytes.readUInt8(0) === 1;
-          const compressedMessageEncoding = compressed ? encoding : undefined;
+          const compressedMessageEncoding = compressed ? encoding : 'identity';
           const decompressedMessage = await this.getDecompressedMessage(requestBytes, compressedMessageEncoding);
 
           // Encountered an error with decompression; it'll already have been propogated back
@@ -729,7 +738,7 @@ export class Http2ServerCallStream<
     readable:
       | ServerReadableStream<RequestType, ResponseType>
       | ServerDuplexStream<RequestType, ResponseType>,
-    encoding?: string
+    encoding: string
   ) {
     const decoder = new StreamDecoder();
 
@@ -750,7 +759,7 @@ export class Http2ServerCallStream<
         this.emit('receiveMessage');
 
         const compressed = message.readUInt8(0) === 1;
-        const compressedMessageEncoding = compressed ? encoding : undefined;
+        const compressedMessageEncoding = compressed ? encoding : 'identity';
         const decompressedMessage = await this.getDecompressedMessage(message, compressedMessageEncoding);
 
         // Encountered an error with decompression; it'll already have been propogated back

packages/grpc-js/src/server.ts

@@ -778,7 +778,7 @@ export class Server {
             });
           }
           const metadata = call.receiveMetadata(headers);
-          const encoding = metadata.get('grpc-encoding')[0] as string | undefined;
+          const encoding = (metadata.get('grpc-encoding')[0] as string | undefined) ?? 'identity';
           metadata.remove('grpc-encoding');
 
           switch (handler.type) {
@@ -863,7 +863,7 @@ async function handleUnary<RequestType, ResponseType>(
   call: Http2ServerCallStream<RequestType, ResponseType>,
   handler: UnaryHandler<RequestType, ResponseType>,
   metadata: Metadata,
-  encoding?: string
+  encoding: string
 ): Promise<void> {
   const request = await call.receiveUnaryMessage(encoding);
 
@@ -894,7 +894,7 @@ function handleClientStreaming<RequestType, ResponseType>(
   call: Http2ServerCallStream<RequestType, ResponseType>,
   handler: ClientStreamingHandler<RequestType, ResponseType>,
   metadata: Metadata,
-  encoding?: string
+  encoding: string
 ): void {
   const stream = new ServerReadableStreamImpl<RequestType, ResponseType>(
     call,
@@ -925,7 +925,7 @@ async function handleServerStreaming<RequestType, ResponseType>(
   call: Http2ServerCallStream<RequestType, ResponseType>,
   handler: ServerStreamingHandler<RequestType, ResponseType>,
   metadata: Metadata,
-  encoding?: string
+  encoding: string
 ): Promise<void> {
   const request = await call.receiveUnaryMessage(encoding);
 
@@ -947,7 +947,7 @@ function handleBidiStreaming<RequestType, ResponseType>(
   call: Http2ServerCallStream<RequestType, ResponseType>,
   handler: BidiStreamingHandler<RequestType, ResponseType>,
   metadata: Metadata,
-  encoding?: string
+  encoding: string
 ): void {
   const stream = new ServerDuplexStreamImpl<RequestType, ResponseType>(
     call,