changes in rerprocess as well.

Author: kannanjgithub

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

@@ -20,19 +20,22 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
+import io.grpc.Attributes.Builder;
 import io.grpc.CallOptions;
 import io.grpc.ClientStreamTracer;
 import io.grpc.Context;
 import io.grpc.InternalChannelz.SocketStats;
 import io.grpc.InternalLogId;
 import io.grpc.LoadBalancer.PickResult;
 import io.grpc.LoadBalancer.PickSubchannelArgs;
+import io.grpc.LoadBalancer.Subchannel;
 import io.grpc.LoadBalancer.SubchannelPicker;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
 import io.grpc.SynchronizationContext;
 import io.grpc.internal.ClientStreamListener.RpcProgress;
+import io.grpc.internal.InternalSubchannel.SubchannelStatusListener;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -129,6 +132,7 @@ public final ClientStream newStream(
         if (state.shutdownStatus != null) {
           return new FailingClientStream(state.shutdownStatus, tracers);
         }
+        InternalSubchannel internalSubchannel = null;
         if (state.lastPicker != null) {
           PickResult pickResult = state.lastPicker.pickSubchannel(args);
           callOptions = args.getCallOptions();
@@ -149,14 +153,21 @@ public final ClientStream newStream(
               stream.setAuthority(pickResult.getAuthorityOverride());
             }
             return stream;
+          } else {
+            if (args.getCallOptions().isWaitForReady()) {
+              Subchannel subchannel = pickResult.getSubchannel();
+              if (subchannel != null) {
+                internalSubchannel = (InternalSubchannel) subchannel.getInternalSubchannel();
+              }
+            }
           }
         }
         // This picker's conclusion is "buffer".  If there hasn't been a newer picker set (possible
         // race with reprocess()), we will buffer the RPC.  Otherwise, will try with the new picker.
         synchronized (lock) {
           PickerState newerState = pickerState;
           if (state == newerState) {
-            return createPendingStream(args, tracers);
+            return createPendingStream(args, tracers, internalSubchannel);
           }
           state = newerState;
         }
@@ -172,8 +183,8 @@ public final ClientStream newStream(
    */
   @GuardedBy("lock")
   private PendingStream createPendingStream(
-      PickSubchannelArgs args, ClientStreamTracer[] tracers) {
-    PendingStream pendingStream = new PendingStream(args, tracers);
+      PickSubchannelArgs args, ClientStreamTracer[] tracers, InternalSubchannel internalSubchannel) {
+    PendingStream pendingStream = new PendingStream(args, tracers, internalSubchannel);
     pendingStreams.add(pendingStream);
     if (getPendingStreamsCount() == 1) {
       syncContext.executeLater(reportTransportInUse);
@@ -308,7 +319,13 @@ final void reprocess(@Nullable SubchannelPicker picker) {
           executor.execute(runnable);
         }
         toRemove.add(stream);
-      }  // else: stay pending
+      }  else { // else: stay pending
+        Subchannel subchannel = pickResult.getSubchannel();
+        if (subchannel != null) {
+          InternalSubchannel internalSubchannel = (InternalSubchannel) subchannel.getInternalSubchannel();
+          internalSubchannel.addSubchannelStatusListener(stream);
+        }
+      }
     }
 
     synchronized (lock) {
@@ -345,14 +362,18 @@ public InternalLogId getLogId() {
     return logId;
   }
 
-  private class PendingStream extends DelayedStream {
+  private class PendingStream extends DelayedStream implements SubchannelStatusListener {
     private final PickSubchannelArgs args;
     private final Context context = Context.current();
     private final ClientStreamTracer[] tracers;
+    private Status lastTransportStatus;
 
-    private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers) {
+    private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers, InternalSubchannel internalSubchannel) {
       this.args = args;
       this.tracers = tracers;
+      if (internalSubchannel != null) {
+        internalSubchannel.addSubchannelStatusListener(this);
+      }
     }
 
     /** Runnable may be null. */
@@ -406,8 +427,16 @@ public void appendTimeoutInsight(InsightBuilder insight) {
       if (args.getCallOptions().isWaitForReady()) {
         insight.append("wait_for_ready");
       }
+      if (lastTransportStatus != null && !lastTransportStatus.isOk()) {
+        insight.appendKeyValue("Last Transport Status", lastTransportStatus);
+      }
       super.appendTimeoutInsight(insight);
     }
+
+    @Override
+    public void updateState(Status s) {
+      this.lastTransportStatus = s;
+    }
   }
 
   static final class PickerState {

core/src/main/java/io/grpc/internal/InternalSubchannel.java

@@ -81,6 +81,7 @@ final class InternalSubchannel implements InternalInstrumented<ChannelStats>, Tr
   private final boolean reconnectDisabled;
 
   private final List<ClientTransportFilter> transportFilters;
+  private final List<SubchannelStatusListener> subchannelStatusListeners = new ArrayList<>();
 
   /**
    * All field must be mutated in the syncContext.
@@ -160,6 +161,7 @@ protected void handleNotInUse() {
   private Status shutdownReason;
 
   private volatile Attributes connectedAddressAttributes;
+  private Status transportStatus;
 
   InternalSubchannel(LoadBalancer.CreateSubchannelArgs args, String authority, String userAgent,
                      BackoffPolicy.Provider backoffPolicyProvider,
@@ -194,10 +196,18 @@ protected void handleNotInUse() {
     this.reconnectDisabled = args.getOption(LoadBalancer.DISABLE_SUBCHANNEL_RECONNECT_KEY);
   }
 
+  Status getTransportStatus() {
+    return transportStatus;
+  }
+
   ChannelLogger getChannelLogger() {
     return channelLogger;
   }
 
+  void addSubchannelStatusListener(SubchannelStatusListener subchannelStatusListener) {
+    subchannelStatusListeners.add(subchannelStatusListener);
+  }
+
   @Override
   public ClientTransport obtainActiveTransport() {
     ClientTransport savedTransport = activeTransport;
@@ -605,12 +615,16 @@ public void transportInUse(boolean inUse) {
 
     @Override
     public void transportShutdown(final Status s) {
+      transportStatus = s;
       channelLogger.log(
           ChannelLogLevel.INFO, "{0} SHUTDOWN with {1}", transport.getLogId(), printShortStatus(s));
       shutdownInitiated = true;
       syncContext.execute(new Runnable() {
         @Override
         public void run() {
+          for (SubchannelStatusListener subchannelStatusListener: subchannelStatusListeners) {
+            subchannelStatusListener.updateState(s);
+          }
           if (state.getState() == SHUTDOWN) {
             return;
           }
@@ -660,6 +674,10 @@ public void run() {
     }
   }
 
+  interface SubchannelStatusListener {
+    void updateState(Status s);
+  }
+
   // All methods are called in syncContext
   abstract static class Callback {
     /**

examples/build.gradle

@@ -45,7 +45,7 @@ dependencies {
 protobuf {
     protoc { artifact = "com.google.protobuf:protoc:${protocVersion}" }
     plugins {
-        grpc { artifact = "io.grpc:protoc-gen-grpc-java:${grpcVersion}" }
+        grpc { artifact = "io.grpc:protoc-gen-grpc-java:1.70.0" }
     }
     generateProtoTasks {
         all()*.plugins { grpc {} }

examples/src/main/java/io/grpc/examples/helloworld/HelloWorldClient.java

@@ -16,6 +16,8 @@
 
 package io.grpc.examples.helloworld;
 
+import static io.grpc.examples.helloworld.GreeterGrpc.getSayHelloMethod;
+
 import io.grpc.Channel;
 import io.grpc.Grpc;
 import io.grpc.InsecureChannelCredentials;
@@ -48,7 +50,7 @@ public void greet(String name) {
     HelloRequest request = HelloRequest.newBuilder().setName(name).build();
     HelloReply response;
     try {
-      response = blockingStub.sayHello(request);
+      response = blockingStub.withWaitForReady().withDeadline(io.grpc.Deadline.after(10, TimeUnit.SECONDS)).sayHello(request);
     } catch (StatusRuntimeException e) {
       logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
       return;