re-enable NoCompress flag behavior and check Compressed Flag byte on server

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

@@ -17,7 +17,7 @@
 
 import * as zlib from 'zlib';
 
-import { Call, WriteObject } from './call-stream';
+import { Call, WriteObject, WriteFlags } from './call-stream';
 import { Channel } from './channel';
 import { BaseFilter, Filter, FilterFactory } from './filter';
 import { Metadata, MetadataValue } from './metadata';
@@ -238,10 +238,10 @@ 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 = !(this.sendCompression instanceof IdentityHandler);
-      // resolvedMessage.flags === undefined
-      //   ? false
-      //   : (resolvedMessage.flags & WriteFlags.NoCompress) === 0;
+    const compress =
+      resolvedMessage.flags === undefined
+        ? !(this.sendCompression instanceof IdentityHandler)
+        : (resolvedMessage.flags & WriteFlags.NoCompress) === 0;
     return {
       message: await this.sendCompression.writeMessage(
         resolvedMessage.message,

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

@@ -561,7 +561,9 @@ export class Http2ServerCallStream<
 
           this.emit('receiveMessage');
 
-          const decompressedMessage = await this.getDecompressedMessage(requestBytes, encoding);
+          const compressed = requestBytes.readUInt8(0) === 1;
+          const compressedMessageEncoding = compressed ? encoding : undefined;
+          const decompressedMessage = await this.getDecompressedMessage(requestBytes, compressedMessageEncoding);
 
           // Encountered an error with decompression; it'll already have been propogated back
           // Just return early
@@ -748,7 +750,9 @@ export class Http2ServerCallStream<
         }
         this.emit('receiveMessage');
 
-        const decompressedMessage = await this.getDecompressedMessage(message, encoding);
+        const compressed = message.readUInt8(0) === 1;
+        const compressedMessageEncoding = compressed ? encoding : undefined;
+        const decompressedMessage = await this.getDecompressedMessage(message, compressedMessageEncoding);
 
         // Encountered an error with decompression; it'll already have been propogated back
         // Just return early

packages/grpc-js/test/test-server.ts

@@ -849,5 +849,31 @@ describe('Compressed requests', () => {
         })
       });
     });
+    
+    it('Should not compress requests when the NoCompress write flag is used', done => {
+      const bidiStream = client.bidiStream();
+      let timesRequested = 0;
+      let timesResponded = 0;
+
+      bidiStream.on('data', () => {
+        timesResponded += 1;
+      });
+
+      bidiStream.on('error', (err) => {
+        assert.ifError(err);
+        done();
+      });
+
+      bidiStream.on('end', () => {
+        assert.equal(timesResponded, timesRequested);
+        done();
+      });
+
+      bidiStream._write({ message: 'foo' }, '2', (err: any) => {
+        assert.ifError(err);
+        timesRequested += 1;
+        setTimeout(() => bidiStream.end(), 10);
+      });
+    });
   });
 });