Method sendRpc

src/RegionClient.java:982–1067

Sends an RPC out to the wire, or queues it if we're disconnected. IMPORTANT : Make sure you've got a reference to the Deferred of this RPC (HBaseRpc#getDeferred) before you call this method. Otherwise there's a race condition if the RPC completes before you get a chance to call {@

(HBaseRpc rpc)

   * return that Deferred or attach a callback to it).
   */
  void sendRpc(HBaseRpc rpc) {
    if (chan != null) {
      // Now {@link GetRequest} is also a {@link BatchableRpc}, we don't want to retry 
      // the get request once it fails.
      if (rpc instanceof BatchableRpc 
          && !(rpc instanceof GetRequest)
          && (server_version >= SERVER_VERSION_092_OR_ABOVE  // Before 0.92,
              || rpc instanceof PutRequest)) {  // we could only batch "put".
        final BatchableRpc edit = (BatchableRpc) rpc;
        if (edit.canBuffer() && hbase_client.getFlushInterval() > 0) {
          bufferEdit(edit);
          return;
        }
        addSingleEditCallbacks(edit);
      } else if (rpc instanceof MultiAction) {
        // Transform single-edit multi-put into single-put.
        final MultiAction batch = (MultiAction) rpc;
        if (batch.size() == 1) {
          rpc = multiActionToSingleAction(batch);
        } else {
          hbase_client.num_multi_rpcs.increment();
        }
      }
      final ChannelBuffer serialized = encode(rpc);
      if (serialized == null) {  // Error during encoding.
        return;  // Stop here.  RPC has been failed already.
      }
      final Channel chan = this.chan;  // Volatile read.
      if (chan != null) {  // Double check if we disconnected during encode().
        // if our channel isn't able to write, we want to properly queue and
        // retry the RPC later or fail it immediately so we don't fill up the
        // channel's buffer.
        if (check_write_status && !chan.isWritable()) {
          rpc.callback(new PleaseThrottleException("Region client [" + this + 
              " ] channel is not writeable.", null, rpc, rpc.getDeferred()));
          removeRpc(rpc, false);
          writes_blocked.incrementAndGet();
          return;
        }
        
        rpc.enqueueTimeout(this);
        Channels.write(chan, serialized);
        rpcs_sent.incrementAndGet();
        return;
      }  // else: continue to the "we're disconnected" code path below.
    }

    boolean tryagain = false;
    boolean dead;  // Shadows this.dead;
    synchronized (this) {
      dead = this.dead;
      // Check if we got connected while entering this synchronized block.
      if (chan != null) {
        tryagain = true;
      } else if (!dead) {
        if (pending_rpcs == null) {
          pending_rpcs = new ArrayList<HBaseRpc>();
        }