Skip to content

Commit 6a7bf67

Browse files
committed
interop: unify peer socket address response streaming in TestServiceImpl
Refactor `fullDuplexCall` in `TestServiceImpl` to route peer socket address requests through the `toChunkQueue()` pipeline and the `ResponseDispatcher`. This replaces the previous approach of manually branching and bypassing the standard chunk streaming flow. Specifically: - Retrieve the peer's socket address on the request-handling thread inside `toChunkQueue()` to prevent context loss on the background scheduling thread. - Overload the `Chunk` class constructor to accept a nullable `peerSocketAddress` while keeping the default constructor for backward compatibility. - Ensure `Chunk.toResponse()` only populates the response `payload` field if `length > 0`, preserving the exact wire-format for empty/address-only requests. - Remove the manual branch on `whetherSendClientSocketAddressInResponse` and its associated helper method, simplifying `fullDuplexCall`.
1 parent 657c929 commit 6a7bf67

1 file changed

Lines changed: 22 additions & 24 deletions

File tree

interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java

Lines changed: 22 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,6 @@
5151
import java.util.ArrayDeque;
5252
import java.util.Arrays;
5353
import java.util.HashMap;
54-
import java.util.Iterator;
5554
import java.util.List;
5655
import java.util.Map;
5756
import java.util.Queue;
@@ -240,27 +239,9 @@ public void onNext(StreamingOutputCallRequest request) {
240239
.asRuntimeException());
241240
return;
242241
}
243-
if (whetherSendClientSocketAddressInResponse(request)) {
244-
responseObserver.onNext(
245-
StreamingOutputCallResponse.newBuilder()
246-
.setPeerSocketAddress(PEER_ADDRESS_CONTEXT_KEY.get().toString())
247-
.build());
248-
return;
249-
}
250242
dispatcher.enqueue(toChunkQueue(request));
251243
}
252244

253-
private boolean whetherSendClientSocketAddressInResponse(StreamingOutputCallRequest request) {
254-
Iterator<ResponseParameters> responseParametersIterator =
255-
request.getResponseParametersList().iterator();
256-
while (responseParametersIterator.hasNext()) {
257-
if (responseParametersIterator.next().getFillPeerSocketAddress().getValue()) {
258-
return true;
259-
}
260-
}
261-
return false;
262-
}
263-
264245
@Override
265246
public void onCompleted() {
266247
if (oobTestLocked) {
@@ -466,7 +447,13 @@ public Queue<Chunk> toChunkQueue(StreamingOutputCallRequest request) {
466447
Queue<Chunk> chunkQueue = new ArrayDeque<>();
467448
int offset = 0;
468449
for (ResponseParameters params : request.getResponseParametersList()) {
469-
chunkQueue.add(new Chunk(params.getIntervalUs(), offset, params.getSize()));
450+
String peerSocketAddress = null;
451+
if (params.getFillPeerSocketAddress().getValue()) {
452+
SocketAddress peerAddress = PEER_ADDRESS_CONTEXT_KEY.get();
453+
peerSocketAddress = peerAddress != null ? peerAddress.toString() : "";
454+
}
455+
chunkQueue.add(
456+
new Chunk(params.getIntervalUs(), offset, params.getSize(), peerSocketAddress));
470457

471458
// Increment the offset past this chunk. Buffer need to be circular.
472459
offset = (offset + params.getSize()) % compressableBuffer.size();
@@ -484,11 +471,17 @@ private class Chunk {
484471
private final int delayMicroseconds;
485472
private final int offset;
486473
private final int length;
474+
private final String peerSocketAddress;
487475

488476
public Chunk(int delayMicroseconds, int offset, int length) {
477+
this(delayMicroseconds, offset, length, null);
478+
}
479+
480+
public Chunk(int delayMicroseconds, int offset, int length, String peerSocketAddress) {
489481
this.delayMicroseconds = delayMicroseconds;
490482
this.offset = offset;
491483
this.length = length;
484+
this.peerSocketAddress = peerSocketAddress;
492485
}
493486

494487
/**
@@ -497,10 +490,15 @@ public Chunk(int delayMicroseconds, int offset, int length) {
497490
private StreamingOutputCallResponse toResponse() {
498491
StreamingOutputCallResponse.Builder responseBuilder =
499492
StreamingOutputCallResponse.newBuilder();
500-
ByteString payload = generatePayload(compressableBuffer, offset, length);
501-
responseBuilder.setPayload(
502-
Payload.newBuilder()
503-
.setBody(payload));
493+
if (length > 0) {
494+
ByteString payload = generatePayload(compressableBuffer, offset, length);
495+
responseBuilder.setPayload(
496+
Payload.newBuilder()
497+
.setBody(payload));
498+
}
499+
if (peerSocketAddress != null) {
500+
responseBuilder.setPeerSocketAddress(peerSocketAddress);
501+
}
504502
return responseBuilder.build();
505503
}
506504
}

0 commit comments

Comments
 (0)