feat: Support TPC connection sharing across gRPC channels

Closes googleapis#13655

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

@@ -27,9 +27,11 @@ internal static class ClientBuilderChannelHelper
  // Set max metadata size to 4 MB i.e., 4194304 bytes.
  .WithCustomOption("grpc.max_metadata_size", 4194304);
 
- internal static GrpcChannelOptions GetUnlimitedSendReceiveChannelOptions() =>
- // Use a random arg to prevent sub-channel re-use in gRPC, so each channel uses its own connection.
- s_unlimitedSendReceiveChannelOptions.WithCustomOption("sub-channel-separator", Guid.NewGuid().ToString());
+ internal static GrpcChannelOptions GetUnlimitedSendReceiveChannelOptions(bool sharedConnections) =>
+ sharedConnections ?
+ s_unlimitedSendReceiveChannelOptions :
+ // Use a random arg to prevent sub-channel re-use in gRPC, so each channel uses its own connection.
+ s_unlimitedSendReceiveChannelOptions.WithCustomOption("sub-channel-separator", Guid.NewGuid().ToString());
 
  internal static Task DisposeChannelAsync(ChannelBase channel)
  {

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

@@ -56,6 +56,12 @@ public int? ClientCount
  }
  }
 
+ /// <summary>
+ /// Whether the <see cref="PublisherServiceApiClient"/>s may share gRPC connections or not.
+ /// Defaults to false, meaning, each client has its own gRPC connection.
+ /// </summary>
+ public bool ClientsShareConnections { get; set; }
+
  /// <summary>
  /// Additional settings for batching, message ordering etc. Default settings will be used if this is null.
  /// </summary>
@@ -108,7 +114,7 @@ private async Task<PublisherClient> BuildAsyncImpl(CancellationToken cancellatio
  var shutdowns = new Func<Task>[clientCount];
  for (int i = 0; i < clientCount; i++)
  {
- var grpcChannelOptions = ClientBuilderChannelHelper.GetUnlimitedSendReceiveChannelOptions();
+ var grpcChannelOptions = ClientBuilderChannelHelper.GetUnlimitedSendReceiveChannelOptions(ClientsShareConnections);
 
  var builder = new PublisherServiceApiClientBuilder(this, grpcChannelOptions);
  clients[i] = isAsync

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

@@ -56,6 +56,12 @@ public int? ClientCount
  }
  }
 
+ /// <summary>
+ /// Whether the <see cref="PublisherServiceApiClient"/>s may share gRPC connections or not.
+ /// Defaults to false, meaning, each client has its own gRPC connection.
+ /// </summary>
+ public bool ClientsShareConnections { get; set; }
+
  /// <summary>
  /// Additional settings for batching, message ordering etc. Default settings will be used if this is null.
  /// </summary>
@@ -108,7 +114,7 @@ private async Task<SubscriberClient> BuildAsyncImpl(CancellationToken cancellati
  var shutdowns = new Func<Task>[clientCount];
  for (int i = 0; i < clientCount; i++)
  {
- var grpcChannelOptions = ClientBuilderChannelHelper.GetUnlimitedSendReceiveChannelOptions();
+ var grpcChannelOptions = ClientBuilderChannelHelper.GetUnlimitedSendReceiveChannelOptions(ClientsShareConnections);
 
  var builder = new SubscriberServiceApiClientBuilder(this, grpcChannelOptions);
  clients[i] = isAsync