code: public void stream(String streamId, StreamCallback callback) {
      StdChannelListener listener = new StdChannelListener(streamId) {
        @Override
        void handleFailure(String errorMsg, Throwable cause) throws Exception {
          callback.onFailure(streamId, new IOException(errorMsg, cause));
        }
      };
      if (logger.isDebugEnabled()) {
        logger.debug("Sending stream request for {} to {}", streamId, getRemoteAddress(channel));
      }
  
      // Need to synchronize here so that the callback is added to the queue and the RPC is
      // written to the socket atomically, so that callbacks are called in the right order
      // when responses arrive.
      synchronized (this) {
        handler.addStreamCallback(streamId, callback);
        channel.writeAndFlush(new StreamRequest(streamId)).addListener(listener);
      }
    }
docs: Request to stream the data with the given stream ID from the remote end.
  
  @param streamId The stream to fetch.
  @param callback Object to call with the stream data.