JavaMuller
diff --git a/Collapse file
‎stub/src/main/java/io/grpc/stub/Calls.java‎
Copy file name to clipboardExpand all lines: stub/src/main/java/io/grpc/stub/Calls.java
+11-3Lines changed: 11 additions & 3 deletions b/Collapse file
‎stub/src/main/java/io/grpc/stub/Calls.java‎
Copy file name to clipboardExpand all lines: stub/src/main/java/io/grpc/stub/Calls.java
+11-3Lines changed: 11 additions & 3 deletions
@@ -45,10 +45,11 @@
 
 import java.util.Iterator;
 import java.util.NoSuchElementException;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -148,7 +149,7 @@ public static <ReqT, RespT> void asyncUnaryCall(
   // TODO(lryan): Not clear if we want to use this idiom for 'simple' stubs.
   public static <ReqT, RespT> Iterator<RespT> blockingServerStreamingCall(
       Call<ReqT, RespT> call, ReqT param) {
-    BlockingResponseStream<RespT> result = new BlockingResponseStream<RespT>();
+    BlockingResponseStream<RespT> result = new BlockingResponseStream<RespT>(call);
     asyncServerStreamingCall(call, param, result.listener());
     return result;
   }
@@ -331,11 +332,17 @@ public void onClose(Status status, Metadata.Trailers trailers) {
    */
   // TODO(ejona): determine how to allow Call.cancel() in case of application error.
   private static class BlockingResponseStream<T> implements Iterator<T> {
-    private final LinkedBlockingQueue<Object> buffer = new LinkedBlockingQueue<Object>();
+    // Due to flow control, only needs to hold up to 2 items: 1 for value, 1 for close.
+    private final BlockingQueue<Object> buffer = new ArrayBlockingQueue<Object>(2);
     private final Call.Listener<T> listener = new QueuingListener();
+    private final Call<?, T> call;
     // Only accessed when iterating.
     private Object last;
 
+    private BlockingResponseStream(Call<?, T> call) {
+      this.call = call;
+    }
+
     Call.Listener<T> listener() {
       return listener;
     }
@@ -362,6 +369,7 @@ public T next() {
         throw new NoSuchElementException();
       }
       try {
+        call.request(1);
         @SuppressWarnings("unchecked")
         T tmp = (T) last;
         return tmp;