code: protected void connect() throws RpcException {
          if (_state.equals(State.CONNECTED)) {
              return;
          }
  
          final ChannelFuture oldChannelFuture = _channelFuture;
  
          if (LOG.isDebugEnabled()) {
              String logPrefix = _usePrivilegedPort ? "usePrivilegedPort " : "";
              LOG.debug("{}connecting to {}", logPrefix, getRemoteAddress());
          }
          _state = State.CONNECTING;
  
          if (_usePrivilegedPort) {
              _channel = bindToPrivilegedPort();
              _channelFuture = _channel.connect(getRemoteAddress());
          } else {
              _channelFuture = _clientBootstrap.connect();
              _channel = _channelFuture.getChannel();
          }
  
          NioSocketChannelConfig cfg = (NioSocketChannelConfig) _channel.getConfig();
          cfg.setWriteBufferHighWaterMark(MAX_SENDING_QUEUE_SIZE);
  
          _channelFuture.addListener(new ChannelFutureListener() {
              /* (non-Javadoc)
               * @see org.jboss.netty.channel.ChannelFutureListener#operationComplete(org.jboss.netty.channel.ChannelFuture)
               */
              public void operationComplete(ChannelFuture future) {
                  if (_channelFuture.isSuccess()) {
                      _state = State.CONNECTED;
                      oldChannelFuture.setSuccess();
                  } else {
                      _state = State.DISCONNECTED;
                      oldChannelFuture.cancel();
                  }
  
              }
          });
  
      }
docs: If there is no current connection, start a new tcp connection asynchronously.
  
  @throws RpcException