code: private void completeResponse(final CouchbaseResponse response,
          final Subject<CouchbaseResponse, CouchbaseResponse> observable) {
          // Noone is listening anymore, handle tracing and/or orphan reporting
          // depending on if enabled or not.
          CouchbaseRequest request = response.request();
          if (request != null && !request.isActive()) {
              if (env().operationTracingEnabled() && request.span() != null) {
                  Scope scope = env().tracer().scopeManager()
                          .activate(request.span(), true);
                  scope.span().setBaggageItem("couchbase.orphan", "true");
                  scope.close();
              }
              if (env().orphanResponseReportingEnabled()) {
                  env().orphanResponseReporter().report(response);
              }
          }
  
          try {
              observable.onNext(response);
              observable.onCompleted();
          } catch (Exception ex) {
              LOGGER.warn("Caught exception while onNext on observable", ex);
              observable.onError(ex);
          }
      }
docs: Fulfill and complete the response observable.
  
  When called directly, this method completes on the event loop, but it can also be used in a callback (see
  {@link #scheduleDirect(CoreScheduler, CouchbaseResponse, Subject)} for example.