code: @Beta
      public Observable<WebSocketFrame> getInput(boolean untilCloseFrame) {
          Observable<WebSocketFrame> rawInput = delegate.getInput();
  
          if (untilCloseFrame) {
              return rawInput.takeUntil(new Func1<WebSocketFrame, Boolean>() {
                  @Override
                  public Boolean call(WebSocketFrame webSocketFrame) {
                      return webSocketFrame instanceof CloseWebSocketFrame;
                  }
              });
          } else {
              return rawInput;
          }
      }
docs: Returns the input stream for this connection. If {@code untilCloseFrame} is {@code true} then the returned stream
  completes after receiving (and emitting) a {@link CloseWebSocketFrame}, otherwise, it completes with an error
  when the underlying channel is closed.
  
  @return The input stream for this connection.