bi-directional stream gRPC with RxJava

MyServiceImpl.java

public class MyServiceImpl extends DeviceServiceGrpc.DeviceServiceImplBase {

  @Override
  public StreamObserver<GetLocationRequest> getLocations(StreamObserver<GetLocationResponse> responseObserver) {
    RequestBridge<GetLocationRequest> request = new RequestBridge<>();
    ResponseBridge response = new ResponseBridge((ServerCallStreamObserver) responseObserver);
    request.map(this::doGetLocation)
           .subscribe(response);
    return request;
  }
}

RequestBridge.java

public class RequestBridge<T> extends Flowable<T> implements StreamObserver<T> {

  private final FlowableProcessor<T> processor = UnicastProcessor.create();

  @Override
  public void onNext(T value) {
    processor.onNext(value);
  }

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

  @Override
  public void onCompleted() {
    processor.onComplete();
  }

  @Override
  protected void subscribeActual(Subscriber<? super T> s) {
    processor.subscribe(s);
  }
}

ResponseBridge.java

@RequiredArgsConstructor
public class ResponseBridge<T> implements Subscriber<T> {

  private final ServerCallStreamObserver<T> observer;
  private Subscription subscription;

  @Override
  public void onSubscribe(Subscription s) {
    subscription = s;
    observer.setOnCancelHandler(s::cancel);
    observer.setOnReadyHandler(() -> s.request(1));
  }

  @Override
  public void onNext(T t) {
    observer.onNext(t);
    if(observer.isReady()) {
      subscription.request(1);
    }
  }

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

  @Override
  public void onComplete() {
    observer.onCompleted();
  }
}