import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
/**
 * 进行必要的业务处理
 * 
 * @param transceiver
 * @throws IOException
 */
private void bussinessDeal(Transceiver transceiver) throws IOException {
    // 2.获取协议
    Protocol protocol = Protocol.parse(this.getClass().getResourceAsStream("/Members.avpr"));
    // 3.根据协议和通讯构造请求对象
    GenericRequestor requestor = new GenericRequestor(protocol, transceiver);
    // 4.根据schema获取messages主节点内容
    GenericRecord loginGr = new GenericData.Record(protocol.getMessages().get("login").getRequest());
    // 5.在根据协议里面获取request中的schema
    GenericRecord mGr = new GenericData.Record(protocol.getType("Members"));
    // 6.设置request中的请求数据
    mGr.put("userName", "rita");
    mGr.put("userPwd", "123456");
    // 7、把二级内容加入到一级message的主节点中
    loginGr.put("m", mGr);
    // 8.设置完毕后，请求方法，正式发送访问请求信息，并得到响应内容
    Object retObj = requestor.request("login", loginGr);
    // 9.进行解析操作
    GenericRecord upGr = (GenericRecord) retObj;
    System.out.println(upGr.get("msg"));
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
/**
 * 采用Netty方式建立和服务端的连接
 * 
 * @throws IOException
 */
public void MemberNettyRPCDynBuilderClient() throws IOException {
    // 1.建立和服务端的Netty通讯
    Transceiver transceiver = new NettyTransceiver(new InetSocketAddress("127.0.0.1", 60090));
    // 2.进行必要的业务处理
    bussinessDeal(transceiver);
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
/**
 * Java工具生成协议代码方式：java -jar E:\avro\avro-tools-1.7.7.jar compile protocol
 * E:\avro\Members.avpr E:\avro 功能和动态调用方式一致
 * 
 * @throws InterruptedException
 * @throws IOException
 */
public void MemberNettyRPCToolsBuilderClient() throws InterruptedException, IOException {
    // 1.和服务端建立通讯
    Transceiver transceiver = new NettyTransceiver(new InetSocketAddress("192.168.1.116", 60090));
    // 2.获取客户端对象
    MemberIFace memberIFace = SpecificRequestor.getClient(MemberIFace.class, transceiver);
    // 3.进行数据设置
    Members members = new Members();
    members.setUserName("rita");
    members.setUserPwd("123456");
    // 开始调用登录方法
    Retmsg retmsg = memberIFace.login(members);
    System.out.println("Recive Msg:" + retmsg.getMsg());
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
/**
 * 采用Netty方式建立和服务端的连接
 * 
 * @throws IOException
 */
public void testClient() throws IOException {
    // 1.建立和服务端的Netty通讯
    Transceiver transceiver = new NettyTransceiver(new InetSocketAddress("127.0.0.1", 60090));
    // 2.进行必要的业务处理
    bussinessDeal(transceiver);
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
@Test
public void testRequest() throws InterruptedException, IOException {
  boolean bound = false;
  int i;

  for (i = 0; i < 100 && !bound; i++) {
    try {
      Context context = new Context();

      context.put("port", String.valueOf(selectedPort = 41414 + i));
      context.put("host", "0.0.0.0");

      Configurables.configure(source, context);

      source.start();
      bound = true;
    } catch (ChannelException e) {
      // Assume port in use, try another one
    }
  }

  Assert
      .assertTrue("Reached start or error", LifecycleController.waitForOneOf(
          source, LifecycleState.START_OR_ERROR));
  Assert.assertEquals("Server is started", LifecycleState.START,
      source.getLifecycleState());

  // setup a requester, to send a flume OG event
  URL url = new URL("http", "0.0.0.0", selectedPort, "/");
  Transceiver http = new HttpTransceiver(url);
  FlumeOGEventAvroServer client = SpecificRequestor.getClient(
      FlumeOGEventAvroServer.class, http);

  AvroFlumeOGEvent avroEvent = AvroFlumeOGEvent.newBuilder().setHost("foo")
      .setPriority(Priority.INFO).setNanos(0).setTimestamp(1)
      .setFields(new HashMap<CharSequence, ByteBuffer>())
      .setBody(ByteBuffer.wrap("foo".getBytes())).build();

  client.append(avroEvent);

  // check if the even has arrived in the channel through OG avro source
  Transaction transaction = channel.getTransaction();
  transaction.begin();

  Event event = channel.take();
  Assert.assertNotNull(event);
  Assert.assertEquals("Channel contained our event", "foo",
      new String(event.getBody()));
  transaction.commit();
  transaction.close();

  source.stop();

  Assert.assertTrue("Reached stop or error",
      LifecycleController.waitForOneOf(source, LifecycleState.STOP_OR_ERROR));
  Assert.assertEquals("Server is stopped", LifecycleState.STOP,
      source.getLifecycleState());
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
/**
 * 采用HTTP方式建立和服务端的连接
 * 
 * @throws IOException
 */
public void MemberHttpRPCDynBuilderClient() throws IOException {
    // 1.建立和服务端的http通讯
    Transceiver transceiver = new HttpTransceiver(new URL("http://127.0.0.1:60090"));
    bussinessDeal(transceiver);
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
public AvroReflectRequestor(Class<?> iface, Transceiver transceiver) throws IOException {
    super(iface, transceiver);
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
@Override
public Transceiver createTransceiver() throws Exception {
    AvroConfiguration configuration = getEndpoint().getConfiguration();
    return transceiver = new NettyTransceiver(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
public AvroSpecificRequestor(Protocol protocol, Transceiver transceiver) throws IOException {
    super(protocol, transceiver);
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
@Override
public Transceiver createTransceiver() throws Exception {
    return new HttpTransceiver(new URL(URISupport.normalizeUri(getEndpoint().getEndpointUri())));
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
protected Protocol handshake(Decoder in, Encoder out, Transceiver connection) throws IOException {
  return AvroPlugin.getInstance().getProtocolVersionResolver().resolve(this, in, out, connection);
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
public Protocol resolve(Responder responder, Decoder in, Encoder out, Transceiver connection)
throws IOException;
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
@Override
public Protocol resolve(Responder responder, Decoder in, Encoder out, Transceiver connection)
    throws IOException {
  Protocol serverProtocol = responder.getLocal();
  byte[] serverMD5 = serverProtocol.getMD5();
  String namespace = serverProtocol.getNamespace();
  String name = serverProtocol.getName();
  HandshakeRequest request = HANDSHAKE_READER.read(null, in);
  MD5 clientHash = request.getClientHash();
  HandshakeResponse response = new HandshakeResponse();
  Protocol protocol = null;
  if (clientHash == null) {
    LOG.error("Client protocol MD5 is missing from request (namespace=" + namespace + ", name=" +
        name + ")");
  } else {
    byte[] clientMD5 = clientHash.bytes();
    if (Arrays.equals(clientMD5, serverMD5)) {
      protocol = serverProtocol;
      response.setMatch(HandshakeMatch.BOTH);
    } else {
      String clientMD5String = DatatypeConverter.printHexBinary(clientMD5);
      protocol = getProtocol(namespace, name, clientMD5String);
      if (protocol == null) {
        try {
          protocol = AvroD2Helper.readProtocolFromZk(
              AvroD2Plugin.getInstance().getZooKeeper(), namespace, name, clientMD5String);
          response.setMatch(HandshakeMatch.CLIENT);
          setProtocol(namespace, name, clientMD5String, protocol);
        } catch (InterruptedException | KeeperException e) {
          LOG.error("Unable to read schema from ZooKeeper for protocol (namespace=" + namespace +
              ", name=" + name + ", MD5=" + clientMD5String + ")", e);
        }
      } else {
        response.setMatch(HandshakeMatch.CLIENT);
      }
    }
  }

  if (protocol == null) {
    response.setMatch(HandshakeMatch.NONE);
  }

  if (response.getMatch() != HandshakeMatch.BOTH) {
    // Do not return the server protocol, because it is already in ZooKeeper.
    // See AvroD2ResponseProcessor.
    // response.setServerProtocol(serverProtocol.toString());
    response.setServerHash(new MD5(serverMD5));
  }

  RPCContext context = new RPCContext();
  context.setHandshakeRequest(request);
  context.setHandshakeResponse(response);
  HANDSHAKE_WRITER.write(response, out);

  return protocol;
}
 

import org.apache.avro.ipc.Transceiver; //导入依赖的package包/类
public abstract Transceiver createTransceiver() throws Exception;