Repository: grpc/grpc-java
Issue: InProcessTransport doesn't call onReady

The in-process transport supports flow control and supports `isReady()`, but it never calls `onReady()`. It seems to be just an oversight/bug. Since the in-process transport connects immediately, `onReady()` should probably be called on the client immediately in `newStream()`.

Locking will be a little interesting since for a single `request()` both client and server listeners may need to be called (because numMessages can be > 1). It looks like `{client,server}Requested()` could maybe return a boolean for whether `{client,server}Requested > 0 && {client,server}Requested <= numMessages`, which would imply `onReady()` should be called.


Changed files:
core/src/main/java/io/grpc/inprocess/InProcessTransport.java

diff --git a/core/src/main/java/io/grpc/inprocess/InProcessTransport.java b/core/src/main/java/io/grpc/inprocess/InProcessTransport.java
index 3498173d6..f9ecf01ac 100644
--- a/core/src/main/java/io/grpc/inprocess/InProcessTransport.java
+++ b/core/src/main/java/io/grpc/inprocess/InProcessTransport.java
@@ -231,14 +231,27 @@ class InProcessTransport implements ServerTransport, ClientTransport {
 
       @Override
       public void request(int numMessages) {
-        clientStream.serverRequested(numMessages);
+        boolean onReady = clientStream.serverRequested(numMessages);
+        if (onReady) {
+          synchronized (this) {
+            if (!closed) {
+              clientStreamListener.onReady();
+            }
+          }
+        }
       }
 
       // This method is the only reason we have to synchronize field accesses.
-      private synchronized void clientRequested(int numMessages) {
+      /**
+       * Client requested more messages.
+       *
+       * @return whether onReady should be called on the server
+       */
+      private synchronized boolean clientRequested(int numMessages) {
         if (closed) {
-          return;
+          return false;
         }
+        boolean previouslyReady = clientRequested > 0;
         clientRequested += numMessages;
         while (clientRequested > 0 && !clientReceiveQueue.isEmpty()) {
           clientRequested--;
@@ -246,12 +259,14 @@ class InProcessTransport implements ServerTransport, ClientTransport {
         }
         // Attempt being reentrant-safe
         if (closed) {
-          return;
+          return false;
         }
         if (clientReceiveQueue.isEmpty() && clientNotifyStatus != null) {
           closed = true;
           clientStreamListener.closed(clientNotifyStatus, clientNotifyTrailers);
         }
+        boolean nowReady = clientRequested > 0;
+        return !previouslyReady && nowReady;
       }
 
       private void clientCancelled(Status status) {
@@ -366,14 +381,27 @@ class InProcessTransport implements ServerTransport, ClientTransport {
 
       @Override
       public void request(int numMessages) {
-        serverStream.clientRequested(numMessages);
+        boolean onReady = serverStream.clientRequested(numMessages);
+        if (onReady) {
+          synchronized (this) {
+            if (!closed) {
+              serverStreamListener.onReady();
+            }
+          }
+        }
       }
 
       // This method is the only reason we have to synchronize field accesses.
-      private synchronized void serverRequested(int numMessages) {
+      /**
+       * Client requested more messages.
+       *
+       * @return whether onReady should be called on the server
+       */
+      private synchronized boolean serverRequested(int numMessages) {
         if (closed) {
-          return;
+          return false;
         }
+        boolean previouslyReady = serverRequested > 0;
         serverRequested += numMessages;
         while (serverRequested > 0 && !serverReceiveQueue.isEmpty()) {
           serverRequested--;
@@ -383,6 +411,8 @@ class InProcessTransport implements ServerTransport, ClientTransport {
           serverNotifyHalfClose = false;
           serverStreamListener.halfClosed();
         }
+        boolean nowReady = serverRequested > 0;
+        return !previouslyReady && nowReady;
       }
 
       private void serverClosed(Status status) {