propagate compressorOptions in stream and rpc_util

Author: Sebas03446

rpc_util.go

@@ -152,6 +152,7 @@ func (d *gzipDecompressor) Type() string {
 // callInfo contains all related configuration and information about an RPC.
 type callInfo struct {
 	compressorName        string
+	compressorOptions     []any
 	failFast              bool
 	maxReceiveMessageSize *int
 	maxSendMessageSize    *int
@@ -452,7 +453,7 @@ func (o PerRPCCredsCallOption) after(*callInfo, *csAttempt) {}
 // Notice: This API is EXPERIMENTAL and may be changed or removed in a
 // later release.
 func UseCompressor(name string, compressorOptions ...any) CallOption {
-	return CompressorCallOption{CompressorType: name}
+	return CompressorCallOption{CompressorType: name, CompressorOptions: compressorOptions}
 }
 
 // CompressorCallOption is a CallOption that indicates the compressor to use.
@@ -462,11 +463,13 @@ func UseCompressor(name string, compressorOptions ...any) CallOption {
 // Notice: This type is EXPERIMENTAL and may be changed or removed in a
 // later release.
 type CompressorCallOption struct {
-	CompressorType string
+	CompressorType    string
+	CompressorOptions []any
 }
 
 func (o CompressorCallOption) before(c *callInfo) error {
 	c.compressorName = o.CompressorType
+	c.compressorOptions = o.CompressorOptions
 	return nil
 }
 func (o CompressorCallOption) after(*callInfo, *csAttempt) {}
@@ -735,7 +738,7 @@ func encode(c baseCodec, msg any) (mem.BufferSlice, error) {
 // indicating no compression was done.
 //
 // TODO(dfawley): eliminate cp parameter by wrapping Compressor in an encoding.Compressor.
-func compress(in mem.BufferSlice, cp Compressor, compressor encoding.Compressor, pool mem.BufferPool) (mem.BufferSlice, payloadFormat, error) {
+func compress(in mem.BufferSlice, cp Compressor, compressor encoding.Compressor, pool mem.BufferPool, compressorOptions ...any) (mem.BufferSlice, payloadFormat, error) {
 	if (compressor == nil && cp == nil) || in.Len() == 0 {
 		return nil, compressionNone, nil
 	}
@@ -746,7 +749,7 @@ func compress(in mem.BufferSlice, cp Compressor, compressor encoding.Compressor,
 		return status.Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error())
 	}
 	if compressor != nil {
-		z, err := compressor.Compress(w)
+		z, err := compressor.Compress(w, compressorOptions...)
 		if err != nil {
 			return nil, 0, wrapErr(err)
 		}

stream.go

@@ -910,7 +910,7 @@ func (cs *clientStream) SendMsg(m any) (err error) {
 	}
 
 	// load hdr, payload, data
-	hdr, data, payload, pf, err := prepareMsg(m, cs.codec, cs.compressorV0, cs.compressorV1, cs.cc.dopts.copts.BufferPool)
+	hdr, data, payload, pf, err := prepareMsg(m, cs.codec, cs.compressorV0, cs.compressorV1, cs.cc.dopts.copts.BufferPool, cs.callInfo.compressorOptions)
 	if err != nil {
 		return err
 	}
@@ -1417,7 +1417,7 @@ func (as *addrConnStream) SendMsg(m any) (err error) {
 	}
 
 	// load hdr, payload, data
-	hdr, data, payload, pf, err := prepareMsg(m, as.codec, as.sendCompressorV0, as.sendCompressorV1, as.ac.dopts.copts.BufferPool)
+	hdr, data, payload, pf, err := prepareMsg(m, as.codec, as.sendCompressorV0, as.sendCompressorV1, as.ac.dopts.copts.BufferPool, as.callInfo.compressorOptions)
 	if err != nil {
 		return err
 	}
@@ -1814,7 +1814,7 @@ func MethodFromServerStream(stream ServerStream) (string, bool) {
 // compression was made and therefore whether the payload needs to be freed in
 // addition to the returned data. Freeing the payload if the returned boolean is
 // false can lead to undefined behavior.
-func prepareMsg(m any, codec baseCodec, cp Compressor, comp encoding.Compressor, pool mem.BufferPool) (hdr []byte, data, payload mem.BufferSlice, pf payloadFormat, err error) {
+func prepareMsg(m any, codec baseCodec, cp Compressor, comp encoding.Compressor, pool mem.BufferPool, compressorOptions ...any) (hdr []byte, data, payload mem.BufferSlice, pf payloadFormat, err error) {
 	if preparedMsg, ok := m.(*PreparedMsg); ok {
 		return preparedMsg.hdr, preparedMsg.encodedData, preparedMsg.payload, preparedMsg.pf, nil
 	}
@@ -1824,7 +1824,7 @@ func prepareMsg(m any, codec baseCodec, cp Compressor, comp encoding.Compressor,
 	if err != nil {
 		return nil, nil, nil, 0, err
 	}
-	compData, pf, err := compress(data, cp, comp, pool)
+	compData, pf, err := compress(data, cp, comp, pool, compressorOptions...)
 	if err != nil {
 		data.Free()
 		return nil, nil, nil, 0, err