Build has been canceled!

#19748

Committed 21 Mar 2025 10:19PM UTC coverage: 88.578% (-0.03%) from 88.604%

Build # #19748

Build Type

push

github

Committed by

web-flow

Commit Message

xds: add support for custom per-target credentials on the transport (#11951)

Run Details

34604 of 39066 relevant lines covered (88.58%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.49

/../xds/src/main/java/io/grpc/xds/GrpcXdsTransportFactory.java

/*
 * Copyright 2024 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.xds;

import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import io.grpc.CallCredentials;
import io.grpc.CallOptions;
import io.grpc.ChannelCredentials;
import io.grpc.ClientCall;
import io.grpc.Context;
import io.grpc.Grpc;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.xds.client.Bootstrapper;
import io.grpc.xds.client.XdsTransportFactory;
import java.util.concurrent.TimeUnit;

final class GrpcXdsTransportFactory implements XdsTransportFactory {

  private final CallCredentials callCredentials;

  GrpcXdsTransportFactory(CallCredentials callCredentials) {
    this.callCredentials = callCredentials;
  }

  @Override
  public XdsTransport create(Bootstrapper.ServerInfo serverInfo) {
    return new GrpcXdsTransport(serverInfo, callCredentials);
  }

  @VisibleForTesting
  public XdsTransport createForTest(ManagedChannel channel) {
    return new GrpcXdsTransport(channel, callCredentials);
  }

  @VisibleForTesting
  static class GrpcXdsTransport implements XdsTransport {

    private final ManagedChannel channel;
    private final CallCredentials callCredentials;

    public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo) {
      this(serverInfo, null);
    }

    @VisibleForTesting
    public GrpcXdsTransport(ManagedChannel channel) {
      this(channel, null);
    }

    public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo, CallCredentials callCredentials) {
      String target = serverInfo.target();
      ChannelCredentials channelCredentials = (ChannelCredentials) serverInfo.implSpecificConfig();
      this.channel = Grpc.newChannelBuilder(target, channelCredentials)
          .keepAliveTime(5, TimeUnit.MINUTES)
          .build();
      this.callCredentials = callCredentials;
    }

    @VisibleForTesting
    public GrpcXdsTransport(ManagedChannel channel, CallCredentials callCredentials) {
      this.channel = checkNotNull(channel, "channel");
      this.callCredentials = callCredentials;
    }

    @Override
    public <ReqT, RespT> StreamingCall<ReqT, RespT> createStreamingCall(
        String fullMethodName,
        MethodDescriptor.Marshaller<ReqT> reqMarshaller,
        MethodDescriptor.Marshaller<RespT> respMarshaller) {
      Context prevContext = Context.ROOT.attach();
      try {
        return new XdsStreamingCall<>(
            fullMethodName, reqMarshaller, respMarshaller, callCredentials);
      } finally {
        Context.ROOT.detach(prevContext);
      }

    }

    @Override
    public void shutdown() {
      channel.shutdown();
    }

    private class XdsStreamingCall<ReqT, RespT> implements
        XdsTransportFactory.StreamingCall<ReqT, RespT> {

      private final ClientCall<ReqT, RespT> call;

      public XdsStreamingCall(
          String methodName,
          MethodDescriptor.Marshaller<ReqT> reqMarshaller,
          MethodDescriptor.Marshaller<RespT> respMarshaller,
          CallCredentials callCredentials) {
        this.call =
            channel.newCall(
                MethodDescriptor.<ReqT, RespT>newBuilder()
                    .setFullMethodName(methodName)
                    .setType(MethodDescriptor.MethodType.BIDI_STREAMING)
                    .setRequestMarshaller(reqMarshaller)
                    .setResponseMarshaller(respMarshaller)
                    .build(),
                CallOptions.DEFAULT.withCallCredentials(
                    callCredentials)); // TODO(zivy): support waitForReady
      }

      @Override
      public void start(EventHandler<RespT> eventHandler) {
        call.start(new EventHandlerToCallListenerAdapter<>(eventHandler), new Metadata());
        call.request(1);
      }

      @Override
      public void sendMessage(ReqT message) {
        call.sendMessage(message);
      }

      @Override
      public void startRecvMessage() {
        call.request(1);
      }

      @Override
      public void sendError(Exception e) {
        call.cancel("Cancelled by XdsClientImpl", e);
      }

      @Override
      public boolean isReady() {
        return call.isReady();
      }
    }
  }

  private static class EventHandlerToCallListenerAdapter<T> extends ClientCall.Listener<T> {
    private final EventHandler<T> handler;

    EventHandlerToCallListenerAdapter(EventHandler<T> eventHandler) {
      this.handler = checkNotNull(eventHandler, "eventHandler");
    }

    @Override
    public void onHeaders(Metadata headers) {}

    @Override
    public void onMessage(T message) {
      handler.onRecvMessage(message);
    }

    @Override
    public void onClose(Status status, Metadata trailers) {
      handler.onStatusReceived(status);
    }

    @Override
    public void onReady() {
      handler.onReady();
    }
  }
}

1	/*
2	* Copyright 2024 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.xds;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20
21	import com.google.common.annotations.VisibleForTesting;
22	import io.grpc.CallCredentials;
23	import io.grpc.CallOptions;
24	import io.grpc.ChannelCredentials;
25	import io.grpc.ClientCall;
26	import io.grpc.Context;
27	import io.grpc.Grpc;
28	import io.grpc.ManagedChannel;
29	import io.grpc.Metadata;
30	import io.grpc.MethodDescriptor;
31	import io.grpc.Status;
32	import io.grpc.xds.client.Bootstrapper;
33	import io.grpc.xds.client.XdsTransportFactory;
34	import java.util.concurrent.TimeUnit;
35
36	final class GrpcXdsTransportFactory implements XdsTransportFactory {
37
38	private final CallCredentials callCredentials;
39
40	GrpcXdsTransportFactory(CallCredentials callCredentials) {	1✔
41	this.callCredentials = callCredentials;	1✔
42	}	1✔
43
44	@Override
45	public XdsTransport create(Bootstrapper.ServerInfo serverInfo) {
46	return new GrpcXdsTransport(serverInfo, callCredentials);	1✔
47	}
48
49	@VisibleForTesting
50	public XdsTransport createForTest(ManagedChannel channel) {
51	return new GrpcXdsTransport(channel, callCredentials);	1✔
52	}
53
54	@VisibleForTesting
55	static class GrpcXdsTransport implements XdsTransport {
56
57	private final ManagedChannel channel;
58	private final CallCredentials callCredentials;
59
60	public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo) {
61	this(serverInfo, null);	×
62	}	×
63
64	@VisibleForTesting
65	public GrpcXdsTransport(ManagedChannel channel) {
66	this(channel, null);	1✔
67	}	1✔
68
69	public GrpcXdsTransport(Bootstrapper.ServerInfo serverInfo, CallCredentials callCredentials) {	1✔
70	String target = serverInfo.target();	1✔
71	ChannelCredentials channelCredentials = (ChannelCredentials) serverInfo.implSpecificConfig();	1✔
72	this.channel = Grpc.newChannelBuilder(target, channelCredentials)	1✔
73	.keepAliveTime(5, TimeUnit.MINUTES)	1✔
74	.build();	1✔
75	this.callCredentials = callCredentials;	1✔
76	}	1✔
77
78	@VisibleForTesting
79	public GrpcXdsTransport(ManagedChannel channel, CallCredentials callCredentials) {	1✔
80	this.channel = checkNotNull(channel, "channel");	1✔
81	this.callCredentials = callCredentials;	1✔
82	}	1✔
83
84	@Override
85	public <ReqT, RespT> StreamingCall<ReqT, RespT> createStreamingCall(
86	String fullMethodName,
87	MethodDescriptor.Marshaller<ReqT> reqMarshaller,
88	MethodDescriptor.Marshaller<RespT> respMarshaller) {
89	Context prevContext = Context.ROOT.attach();	1✔
90	try {
91	return new XdsStreamingCall<>(	1✔
92	fullMethodName, reqMarshaller, respMarshaller, callCredentials);
93	} finally {
94	Context.ROOT.detach(prevContext);	1✔
95	}
96
97	}
98
99	@Override
100	public void shutdown() {
101	channel.shutdown();	1✔
102	}	1✔
103
104	private class XdsStreamingCall<ReqT, RespT> implements
105	XdsTransportFactory.StreamingCall<ReqT, RespT> {
106
107	private final ClientCall<ReqT, RespT> call;
108
109	public XdsStreamingCall(
110	String methodName,
111	MethodDescriptor.Marshaller<ReqT> reqMarshaller,
112	MethodDescriptor.Marshaller<RespT> respMarshaller,
113	CallCredentials callCredentials) {	1✔
114	this.call =	1✔
115	channel.newCall(	1✔
116	MethodDescriptor.<ReqT, RespT>newBuilder()	1✔
117	.setFullMethodName(methodName)	1✔
118	.setType(MethodDescriptor.MethodType.BIDI_STREAMING)	1✔
119	.setRequestMarshaller(reqMarshaller)	1✔
120	.setResponseMarshaller(respMarshaller)	1✔
121	.build(),	1✔
122	CallOptions.DEFAULT.withCallCredentials(	1✔
123	callCredentials)); // TODO(zivy): support waitForReady
124	}	1✔
125
126	@Override
127	public void start(EventHandler<RespT> eventHandler) {
128	call.start(new EventHandlerToCallListenerAdapter<>(eventHandler), new Metadata());	1✔
129	call.request(1);	1✔
130	}	1✔
131
132	@Override
133	public void sendMessage(ReqT message) {
134	call.sendMessage(message);	1✔
135	}	1✔
136
137	@Override
138	public void startRecvMessage() {
139	call.request(1);	1✔
140	}	1✔
141
142	@Override
143	public void sendError(Exception e) {
144	call.cancel("Cancelled by XdsClientImpl", e);	1✔
145	}	1✔
146
147	@Override
148	public boolean isReady() {
149	return call.isReady();	1✔
150	}
151	}
152	}
153
154	private static class EventHandlerToCallListenerAdapter<T> extends ClientCall.Listener<T> {
155	private final EventHandler<T> handler;
156
157	EventHandlerToCallListenerAdapter(EventHandler<T> eventHandler) {	1✔
158	this.handler = checkNotNull(eventHandler, "eventHandler");	1✔
159	}	1✔
160
161	@Override
162	public void onHeaders(Metadata headers) {}	1✔
163
164	@Override
165	public void onMessage(T message) {
166	handler.onRecvMessage(message);	1✔
167	}	1✔
168
169	@Override
170	public void onClose(Status status, Metadata trailers) {
171	handler.onStatusReceived(status);	1✔
172	}	1✔
173
174	@Override
175	public void onReady() {
176	handler.onReady();	1✔
177	}	1✔
178	}
179	}

grpc / grpc-java / #19748

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous