code: @Override
      public void onEvent(final ResponseEvent event, long sequence, boolean endOfBatch) throws Exception {
          try {
              CouchbaseMessage message = event.getMessage();
              if (message instanceof SignalConfigReload) {
                  configurationProvider.signalOutdated();
              } else if (message instanceof CouchbaseResponse) {
                  final CouchbaseResponse response = (CouchbaseResponse) message;
                  ResponseStatus status = response.status();
                  if (status == ResponseStatus.RETRY) {
                      retry(event, true);
                  } else {
                      final Scheduler.Worker worker = environment.scheduler().createWorker();
                      final Subject<CouchbaseResponse, CouchbaseResponse> obs = event.getObservable();
                      worker.schedule(new Action0() {
                          @Override
                          public void call() {
                              try {
                                  obs.onNext(response);
                                  obs.onCompleted();
                              } catch(Exception ex) {
                                  obs.onError(ex);
                              } finally {
                                  worker.unsubscribe();
                              }
                          }
                      });
                  }
              } else if (message instanceof CouchbaseRequest) {
                  retry(event, false);
              } else {
                  throw new IllegalStateException("Got message type I do not understand: " + message);
              }
          } finally {
             event.setMessage(null);
             event.setObservable(null);
          }
      }
docs: Handles {@link ResponseEvent}s that come into the response RingBuffer.
  
  Hey I just mapped you,
  And this is crazy,
  But here's my data
  so subscribe me maybe.
  
  It's hard to block right,
  at you baby,
  But here's my data ,
  so subscribe me maybe.