package mirror;

import io.grpc.stub.CallStreamObserver;
import io.grpc.stub.StreamObserver;

/**
 * Provides flow control feedback to the application by blocking on {@code onNext}
 * if the underlying {@link CallStreamObserver} is not ready.
 */
class BlockingStreamObserver<T> implements StreamObserver<T> {

  private final CallStreamObserver<T> delegate;
  private final Object lock = new Object();

  BlockingStreamObserver(CallStreamObserver<T> delegate) {
    this.delegate = delegate;
    this.delegate.setOnReadyHandler(() -> {
      synchronized (lock) {
        lock.notifyAll(); // wake up our thread
      }
    });
  }

  @Override
  public void onNext(T value) {
    synchronized (lock) {
      while (!delegate.isReady()) {
        try {
          lock.wait();
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          return;
        }
      }
    }
    delegate.onNext(value);
  }

  @Override
  public void onError(Throwable t) {
    delegate.onError(t);
  }

  @Override
  public void onCompleted() {
    delegate.onCompleted();
  }
}

@@ -50,16 +50,62 @@
 server;%0A
+import io.grpc.stub.ServerCallStreamObserver;%0A
 import i
@@ -282,16 +282,433 @@
  ready.%0A
+ *%0A * Typically blocking a thread is considered bad form, e.g. you could block your client%0A * UI thread or your server-side request-serving thread, but for Mirror's purposes, we%0A * only write to BlockingStreamObserver from our own dedicated application threads. These%0A * application threads are typically processing a queue, so blocking is actually what we%0A * want to do, as then the work will build up in the queue.%0A
  */%0Aclas
@@ -957,40 +957,35 @@
     
-this.delegate.setOnReadyHandler(
+final Runnable notifyAll = 
 () -
@@ -1081,85 +1081,752 @@
    %7D
-);%0A  %7D%0A%0A  @Override%0A  public void onNext(T value) %7B%0A    synchronized (lock) %7B
+;%0A    this.delegate.setOnReadyHandler(notifyAll);%0A    if (delegate instanceof ServerCallStreamObserver) %7B%0A      ((ServerCallStreamObserver%3CT%3E) delegate).setOnCancelHandler(notifyAll);%0A    %7D%0A  %7D%0A%0A  @Override%0A  public void onNext(T value) %7B%0A    synchronized (lock) %7B%0A      // in theory we could implement ServerCallStreamObserver and expose isCancelled to our client,%0A      // but for current purposes we only need the StreamObserver API, so treat a cancelled observer%0A      // as something we just want to un-block from and return, and we'll trust the rest of our session%0A      // to shutdown accordingly.%0A      if (delegate instanceof ServerCallStreamObserver && ((ServerCallStreamObserver%3CT%3E) delegate).isCancelled()) %7B%0A        return;%0A      %7D
 %0A   

Update BlockingStreamObserver to handle cancelled server-side delegates.