grpc
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

alts: create handshaker RPC lazily (#7630) 

* alts: create handshaker RPC lazily

* alts: address review comments
This commit is contained in:
Jiangtao Li 2020-11-17 17:36:09 -08:00 committed by GitHub
parent d7a00e6047
commit 24e4d68282
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 17 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
alts/src/main/java/io/grpc/alts/internal/AltsHandshakerStub.java
View File

@ -29,7 +29,8 @@ import java.util.concurrent.atomic.AtomicReference;
/** An interface to the ALTS handshaker service. */ /** An interface to the ALTS handshaker service. */
class AltsHandshakerStub { class AltsHandshakerStub {
private final StreamObserver<HandshakerResp> reader = new Reader(); private final StreamObserver<HandshakerResp> reader = new Reader();
private final StreamObserver<HandshakerReq> writer; private StreamObserver<HandshakerReq> writer;
private final HandshakerServiceStub serviceStub;
private final ArrayBlockingQueue<Optional<HandshakerResp>> responseQueue = private final ArrayBlockingQueue<Optional<HandshakerResp>> responseQueue =
new ArrayBlockingQueue<>(1); new ArrayBlockingQueue<>(1);
private final AtomicReference<String> exceptionMessage = new AtomicReference<>(); private final AtomicReference<String> exceptionMessage = new AtomicReference<>();
@ -37,20 +38,18 @@ class AltsHandshakerStub {
private static final long HANDSHAKE_RPC_DEADLINE_SECS = 20; private static final long HANDSHAKE_RPC_DEADLINE_SECS = 20;
AltsHandshakerStub(HandshakerServiceStub serviceStub) { AltsHandshakerStub(HandshakerServiceStub serviceStub) {
this.writer = this.serviceStub = serviceStub;
serviceStub
.withDeadlineAfter(HANDSHAKE_RPC_DEADLINE_SECS, SECONDS)
.doHandshake(this.reader);
} }
@VisibleForTesting @VisibleForTesting
AltsHandshakerStub() { AltsHandshakerStub() {
writer = null; serviceStub = null;
} }
@VisibleForTesting @VisibleForTesting
AltsHandshakerStub(StreamObserver<HandshakerReq> writer) { AltsHandshakerStub(StreamObserver<HandshakerReq> writer) {
this.writer = writer; this.writer = writer;
serviceStub = null;
} }
@VisibleForTesting @VisibleForTesting
@ -60,6 +59,7 @@ class AltsHandshakerStub {
/** Send a handshaker request and return the handshaker response. */ /** Send a handshaker request and return the handshaker response. */
public HandshakerResp send(HandshakerReq req) throws InterruptedException, IOException { public HandshakerResp send(HandshakerReq req) throws InterruptedException, IOException {
createWriterIfNull();
maybeThrowIoException(); maybeThrowIoException();
if (!responseQueue.isEmpty()) { if (!responseQueue.isEmpty()) {
throw new IOException("Received an unexpected response."); throw new IOException("Received an unexpected response.");
@ -72,6 +72,14 @@ class AltsHandshakerStub {
return result.get(); return result.get();
} }
/** Create a new writer if the writer is null. */
private void createWriterIfNull() {
if (writer == null) {
writer =
serviceStub.withDeadlineAfter(HANDSHAKE_RPC_DEADLINE_SECS, SECONDS).doHandshake(reader);
}
}
/** Throw exception if there is an outstanding exception. */ /** Throw exception if there is an outstanding exception. */
private void maybeThrowIoException() throws IOException { private void maybeThrowIoException() throws IOException {
if (exceptionMessage.get() != null) { if (exceptionMessage.get() != null) {
@ -81,7 +89,9 @@ class AltsHandshakerStub {
/** Close the connection. */ /** Close the connection. */
public void close() { public void close() {
writer.onCompleted(); if (writer != null) {
writer.onCompleted();
}
} }
private class Reader implements StreamObserver<HandshakerResp> { private class Reader implements StreamObserver<HandshakerResp> {