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 ) ; } }
Fulfill and complete the response observable .