feat: Add support for Publisher Compression.

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/PublisherClient.Settings.cs

@@ -50,6 +50,21 @@ public sealed class Settings
         /// </summary>
         public TimeSpan? DisposeTimeout { get; set; }
 
+        /// <summary>
+        /// Enables publish message compression. If set to <c>true</c>, messages will be compressed before being sent to the server
+        /// by the <see cref="PublisherClient"/>.
+        /// </summary>
+        public bool EnableCompression { get; set; }
+
+        /// <summary>
+        /// Specifies the threshold for the number of bytes in a message batch before compression is applied.
+        /// This property comes into play only when <see cref="EnableCompression"/> is set to <c>true</c>.
+        /// If the number of bytes in a batch is less than this value, compression will not be applied even
+        /// if <see cref="EnableCompression"/> is <c>true</c>.
+        /// If <c>null</c>, defaults to <see cref="DefaultCompressionBytesThreshold"/>.
+        /// </summary>
+        public long? CompressionBytesThreshold { get; set; }
+
         /// <summary>
         /// Create a new instance.
         /// </summary>
@@ -61,6 +76,8 @@ internal Settings(Settings other)
             Scheduler = other.Scheduler;
             EnableMessageOrdering = other.EnableMessageOrdering;
             DisposeTimeout = other.DisposeTimeout;
+            EnableCompression = other.EnableCompression;
+            CompressionBytesThreshold = other.CompressionBytesThreshold;
         }
 
         internal void Validate()

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/PublisherClient.cs

@@ -53,6 +53,11 @@ public abstract partial class PublisherClient : IAsyncDisposable
     /// </summary>
     public static TimeSpan DefaultDisposeTimeout { get; } = TimeSpan.FromSeconds(5);
 
+    /// <summary>
+    /// The default <see cref="Settings.CompressionBytesThreshold"/> of 240 bytes for the publisher message compression.
+    /// </summary>
+    public static long DefaultCompressionBytesThreshold { get; } = 240;
+
     /// <summary>
     /// The associated <see cref="TopicName"/>. 
     /// </summary>

apis/Google.Cloud.PubSub.V1/Google.Cloud.PubSub.V1/PublisherClientImpl.cs

@@ -35,6 +35,11 @@ namespace Google.Cloud.PubSub.V1;
 /// </summary>
 public sealed class PublisherClientImpl : PublisherClient
 {
+    private const string CompressionHeaderKey = "grpc-internal-encoding-request";
+    private const string CompressionHeaderValue = "gzip";
+
+    private static readonly CallSettings s_compressionCallSettings = CallSettings.FromHeader(CompressionHeaderKey, CompressionHeaderValue);
+
     /// <summary>
     /// A batch of messages that all have the same ordering-key (or no ordering-key), and will be
     /// sent to the server in a single network call.
@@ -133,6 +138,8 @@ internal PublisherClientImpl(TopicName topicName, IEnumerable<PublisherServiceAp
         _taskHelper = GaxPreconditions.CheckNotNull(taskHelper, nameof(taskHelper));
         _enableMessageOrdering = settings.EnableMessageOrdering;
         _disposeTimeout = settings.DisposeTimeout ?? DefaultDisposeTimeout;
+        _enableCompression = settings.EnableCompression;
+        _compressionBytesThreshold = settings.CompressionBytesThreshold ?? DefaultCompressionBytesThreshold;
 
         // Initialise batching settings. Use ApiMax settings for components not given.
         var batchingSettings = settings.BatchingSettings ?? DefaultBatchingSettings;
@@ -156,6 +163,8 @@ internal PublisherClientImpl(TopicName topicName, IEnumerable<PublisherServiceAp
     private readonly Func<Task> _shutdown;
     private readonly bool _enableMessageOrdering;
     private readonly TimeSpan _disposeTimeout;
+    private readonly bool _enableCompression;
+    private readonly long _compressionBytesThreshold;
 
     // Batching settings
     private readonly long _batchElementCountThreshold;
@@ -446,8 +455,16 @@ private void TriggerSend()
 
         async Task Send()
         {
+            var callSettings = CallSettings.FromCancellationToken(_hardStopCts.Token);
+
+            // If compression is enabled, and the batch size is greater than or equal to the threshold, set the compression header.
+            if (_enableCompression && batch.ByteCount >= _compressionBytesThreshold)
+            {
+                callSettings = callSettings.MergedWith(s_compressionCallSettings);
+            }
+
             // Perform the RPC to server, catching exceptions.
-            var publishTask = client.PublishAsync(TopicName, batch.Messages, CallSettings.FromCancellationToken(_hardStopCts.Token));
+            var publishTask = client.PublishAsync(TopicName, batch.Messages, callSettings);
             var response = await _taskHelper.ConfigureAwaitHideErrors(() => publishTask, null);
             Action postLockAction;
             lock (_lock)