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`.

Author: kannanjgithub

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

@@ -240,27 +240,9 @@ public void onNext(StreamingOutputCallRequest request) {
               .asRuntimeException());
           return;
         }
-        if (whetherSendClientSocketAddressInResponse(request)) {
-          responseObserver.onNext(
-              StreamingOutputCallResponse.newBuilder()
-                  .setPeerSocketAddress(PEER_ADDRESS_CONTEXT_KEY.get().toString())
-                  .build());
-          return;
-        }
         dispatcher.enqueue(toChunkQueue(request));
       }
 
-      private boolean whetherSendClientSocketAddressInResponse(StreamingOutputCallRequest request) {
-        Iterator<ResponseParameters> responseParametersIterator =
-            request.getResponseParametersList().iterator();
-        while (responseParametersIterator.hasNext()) {
-          if (responseParametersIterator.next().getFillPeerSocketAddress().getValue()) {
-            return true;
-          }
-        }
-        return false;
-      }
-
       @Override
       public void onCompleted() {
         if (oobTestLocked) {
@@ -466,7 +448,12 @@ public Queue<Chunk> toChunkQueue(StreamingOutputCallRequest request) {
     Queue<Chunk> chunkQueue = new ArrayDeque<>();
     int offset = 0;
     for (ResponseParameters params : request.getResponseParametersList()) {
-      chunkQueue.add(new Chunk(params.getIntervalUs(), offset, params.getSize()));
+      String peerSocketAddress = null;
+      if (params.getFillPeerSocketAddress().getValue()) {
+        SocketAddress peerAddress = PEER_ADDRESS_CONTEXT_KEY.get();
+        peerSocketAddress = peerAddress != null ? peerAddress.toString() : "";
+      }
+      chunkQueue.add(new Chunk(params.getIntervalUs(), offset, params.getSize(), peerSocketAddress));
 
       // Increment the offset past this chunk. Buffer need to be circular.
       offset = (offset + params.getSize()) % compressableBuffer.size();
@@ -484,11 +471,17 @@ private class Chunk {
     private final int delayMicroseconds;
     private final int offset;
     private final int length;
+    private final String peerSocketAddress;
 
     public Chunk(int delayMicroseconds, int offset, int length) {
+      this(delayMicroseconds, offset, length, null);
+    }
+
+    public Chunk(int delayMicroseconds, int offset, int length, String peerSocketAddress) {
       this.delayMicroseconds = delayMicroseconds;
       this.offset = offset;
       this.length = length;
+      this.peerSocketAddress = peerSocketAddress;
     }
 
     /**
@@ -497,10 +490,15 @@ public Chunk(int delayMicroseconds, int offset, int length) {
     private StreamingOutputCallResponse toResponse() {
       StreamingOutputCallResponse.Builder responseBuilder =
           StreamingOutputCallResponse.newBuilder();
-      ByteString payload = generatePayload(compressableBuffer, offset, length);
-      responseBuilder.setPayload(
-          Payload.newBuilder()
-              .setBody(payload));
+      if (length > 0) {
+        ByteString payload = generatePayload(compressableBuffer, offset, length);
+        responseBuilder.setPayload(
+            Payload.newBuilder()
+                .setBody(payload));
+      }
+      if (peerSocketAddress != null) {
+        responseBuilder.setPeerSocketAddress(peerSocketAddress);
+      }
       return responseBuilder.build();
     }
   }