code: private CouchbaseResponse handleBucketStreamingResponse(final ChannelHandlerContext ctx,
          final HttpResponse header) {
          SocketAddress addr = ctx.channel().remoteAddress();
          String host = addr instanceof InetSocketAddress ? ((InetSocketAddress) addr).getAddress().getHostAddress()
              : addr.toString();
          ResponseStatus status = ResponseStatusConverter.fromHttp(header.getStatus().code());
  
          Observable<String> scheduledObservable = null;
          if (status.isSuccess()) {
              streamingConfigObservable = BehaviorSubject.create();
              scheduledObservable = streamingConfigObservable.onBackpressureBuffer().observeOn(env().scheduler());
          }
          return new BucketStreamingResponse(
              scheduledObservable,
              host,
              status,
              currentRequest()
          );
      }
docs: Decodes a {@link BucketStreamingResponse}.
  
  @param ctx the handler context.
  @param header the received header.
  @return a initialized {@link CouchbaseResponse}.