3333import io .grpc .Status ;
3434import io .grpc .xds .client .Bootstrapper ;
3535import io .grpc .xds .client .XdsTransportFactory ;
36+ import java .util .Map ;
37+ import java .util .concurrent .ConcurrentHashMap ;
3638import java .util .concurrent .TimeUnit ;
3739
40+ /**
41+ * A factory for creating gRPC-based transports for xDS communication.
42+ *
43+ * <p>WARNING: This class reuses channels when possible, based on the provided {@link
44+ * Bootstrapper.ServerInfo} with important considerations. The {@link Bootstrapper.ServerInfo}
45+ * includes {@link ChannelCredentials}, which is compared by reference equality. This means every
46+ * {@link Bootstrapper.BootstrapInfo} would have non-equal copies of {@link
47+ * Bootstrapper.ServerInfo}, even if they all represent the same xDS server configuration. For gRPC
48+ * name resolution with the {@code xds} and {@code google-c2p} scheme, this transport sharing works
49+ * as expected as it internally reuses a single {@link Bootstrapper.BootstrapInfo} instance.
50+ * Otherwise, new transports would be created for each {@link Bootstrapper.ServerInfo} despite them
51+ * possibly representing the same xDS server configuration and defeating the purpose of transport
52+ * sharing.
53+ */
3854final class GrpcXdsTransportFactory implements XdsTransportFactory {
3955
4056 private final CallCredentials callCredentials ;
57+ <<<<<<< child -channel -plugin
4158 private final ChannelConfigurer channelConfigurer ;
59+ =======
60+ // The map of xDS server info to its corresponding gRPC xDS transport.
61+ // This enables reusing and sharing the same underlying gRPC channel.
62+ //
63+ // NOTE: ConcurrentHashMap is used as a per-entry lock and all reads and writes must be a mutation
64+ // via the ConcurrentHashMap APIs to acquire the per-entry lock in order to ensure thread safety
65+ // for reference counting of each GrpcXdsTransport instance.
66+ private static final Map <Bootstrapper .ServerInfo , GrpcXdsTransport > xdsServerInfoToTransportMap =
67+ new ConcurrentHashMap <>();
68+ >>>>>>> master
4269
4370
4471 GrpcXdsTransportFactory (CallCredentials callCredentials ,
@@ -49,27 +76,42 @@ final class GrpcXdsTransportFactory implements XdsTransportFactory {
4976
5077 @ Override
5178 public XdsTransport create (Bootstrapper .ServerInfo serverInfo ) {
79+ <<<<<<< child -channel -plugin
5280 return new GrpcXdsTransport (serverInfo , callCredentials , channelConfigurer );
81+ =======
82+ return xdsServerInfoToTransportMap .compute (
83+ serverInfo ,
84+ (info , transport ) -> {
85+ if (transport == null ) {
86+ transport = new GrpcXdsTransport (serverInfo , callCredentials );
87+ }
88+ ++transport .refCount ;
89+ return transport ;
90+ });
91+ >>>>>>> master
5392 }
5493
5594 @ VisibleForTesting
5695 public XdsTransport createForTest (ManagedChannel channel ) {
57- return new GrpcXdsTransport (channel , callCredentials );
96+ return new GrpcXdsTransport (channel , callCredentials , null );
5897 }
5998
6099 @ VisibleForTesting
61100 static class GrpcXdsTransport implements XdsTransport {
62101
63102 private final ManagedChannel channel ;
64103 private final CallCredentials callCredentials ;
104+ private final Bootstrapper .ServerInfo serverInfo ;
105+ // Must only be accessed via the ConcurrentHashMap APIs which act as the locking methods.
106+ private int refCount = 0 ;
65107
66108 public GrpcXdsTransport (Bootstrapper .ServerInfo serverInfo ) {
67109 this (serverInfo , null );
68110 }
69111
70112 @ VisibleForTesting
71113 public GrpcXdsTransport (ManagedChannel channel ) {
72- this (channel , null );
114+ this (channel , null , null );
73115 }
74116
75117 public GrpcXdsTransport (Bootstrapper .ServerInfo serverInfo , CallCredentials callCredentials ) {
@@ -79,6 +121,7 @@ public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo, CallCredentials call
79121 .keepAliveTime (5 , TimeUnit .MINUTES )
80122 .build ();
81123 this .callCredentials = callCredentials ;
124+ this .serverInfo = serverInfo ;
82125 }
83126
84127 public GrpcXdsTransport (Bootstrapper .ServerInfo serverInfo ,
@@ -96,9 +139,13 @@ public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo,
96139 }
97140
98141 @ VisibleForTesting
99- public GrpcXdsTransport (ManagedChannel channel , CallCredentials callCredentials ) {
142+ public GrpcXdsTransport (
143+ ManagedChannel channel ,
144+ CallCredentials callCredentials ,
145+ Bootstrapper .ServerInfo serverInfo ) {
100146 this .channel = checkNotNull (channel , "channel" );
101147 this .callCredentials = callCredentials ;
148+ this .serverInfo = serverInfo ;
102149 }
103150
104151 @ Override
@@ -118,7 +165,19 @@ public <ReqT, RespT> StreamingCall<ReqT, RespT> createStreamingCall(
118165
119166 @ Override
120167 public void shutdown () {
121- channel .shutdown ();
168+ if (serverInfo == null ) {
169+ channel .shutdown ();
170+ return ;
171+ }
172+ xdsServerInfoToTransportMap .computeIfPresent (
173+ serverInfo ,
174+ (info , transport ) -> {
175+ if (--transport .refCount == 0 ) { // Prefix decrement and return the updated value.
176+ transport .channel .shutdown ();
177+ return null ; // Remove mapping.
178+ }
179+ return transport ;
180+ });
122181 }
123182
124183 private class XdsStreamingCall <ReqT , RespT > implements
0 commit comments