lang/java/ipc/src/main/java/org/apache/avro/ipc/Transceiver.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.avro.ipc;

import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.avro.Protocol;

/** Base transport class used by {@link Requestor}. */
public abstract class Transceiver implements Closeable {
  private final ReentrantLock channelLock = new ReentrantLock();

  public abstract String getRemoteName() throws IOException;

  /**
   * Acquires an exclusive lock on the transceiver's channel.
   */
  public void lockChannel() {
    channelLock.lock();
  }

  /**
   * Releases the lock on the transceiver's channel if held by the calling thread.
   */
  public void unlockChannel() {
    if (channelLock.isHeldByCurrentThread()) {
      channelLock.unlock();
    }
  }

  /** Called by {@link Requestor#request(String,Object)} for two-way messages.
   * By default calls {@link #writeBuffers(List)} followed by
   * {@link #readBuffers()}. */
  public List<ByteBuffer> transceive(List<ByteBuffer> request)
    throws IOException {
    lockChannel();
    try {
      writeBuffers(request);
      return readBuffers();
    } finally {
      unlockChannel();
    }
  }

  /**
   * Called by {@link Requestor#request(String,Object,Callback)} for two-way messages using callbacks.
   */
  public void transceive(List<ByteBuffer> request, Callback<List<ByteBuffer>> callback)
    throws IOException {
    // The default implementation works synchronously
    try {
      List<ByteBuffer> response = transceive(request);
      callback.handleResult(response);
    } catch (IOException e) {
      callback.handleError(e);
    }
  }

  /** Called by the default definition of {@link #transceive(List)}.*/
  public abstract List<ByteBuffer> readBuffers() throws IOException;

  /** Called by {@link Requestor#request(String,Object)} for one-way messages.*/
  public abstract void writeBuffers(List<ByteBuffer> buffers)
    throws IOException;

  /** True if a handshake has been completed for this connection.  Used to
   * determine whether a handshake need be completed prior to a one-way
   * message.  Requests and responses are always prefixed by handshakes, but
   * one-way messages.  If the first request sent over a connection is one-way,
   * then a handshake-only response is returned.  Subsequent one-way messages
   * over the connection will have no response data sent.  Returns false by
   * default. */
  public boolean isConnected() { return false; }

  /** Called with the remote protocol when a handshake has been completed.
   * After this has been called and while a connection is maintained, {@link
   * #isConnected()} should return true and #getRemote() should return this
   * protocol.  Does nothing by default. */
  public void setRemote(Protocol protocol) {}

  /** Returns the protocol passed to {@link #setRemote(Protocol)}.  Throws
   * IllegalStateException by default. */
  public Protocol getRemote() {
    throw new IllegalStateException("Not connected.");
  }

  public void close() throws IOException {}
}