import org.apache.avro.ipc.Responder; //导入依赖的package包/类
public static Server startTestFlumeServer(int port) {
  Responder responder = new SpecificResponder(AvroSourceProtocol.class,
          new OKAvroHandler());
  Server server = new NettyServer(responder,
            new InetSocketAddress("127.0.0.1", port));

  server.start();
  LOG.info("Server started on test flume server hostname: localhost, port: " + port);

  try {

    Thread.sleep(1000L);

  } catch (InterruptedException ex) {
    LOG.error("Thread interrupted. Exception follows.", ex);
    Thread.currentThread().interrupt();
  }

  return server;
}
 

import org.apache.avro.ipc.Responder; //导入依赖的package包/类
@Override
public void start() {
  logger.info("Starting {}...", this);

  Responder responder = new SpecificResponder(AvroSourceProtocol.class, this);

  NioServerSocketChannelFactory socketChannelFactory = initSocketChannelFactory();

  ChannelPipelineFactory pipelineFactory = initChannelPipelineFactory();

  server = new NettyServer(responder, new InetSocketAddress(bindAddress, port),
        socketChannelFactory, pipelineFactory, null);

  connectionCountUpdater = Executors.newSingleThreadScheduledExecutor();
  server.start();
  sourceCounter.start();
  super.start();
  final NettyServer srv = (NettyServer)server;
  connectionCountUpdater.scheduleWithFixedDelay(new Runnable() {

    @Override
    public void run() {
      sourceCounter.setOpenConnectionCount(
              Long.valueOf(srv.getNumActiveConnections()));
    }
  }, 0, 60, TimeUnit.SECONDS);

  logger.info("Avro source {} started.", getName());
}
 

import org.apache.avro.ipc.Responder; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  headers = Maps.newHashMap();
  headers.put("key1", "value1");
  body = "body".getBytes(Charsets.UTF_8);

  int port = findFreePort();
  eventCollector = new EventCollector();
  Responder responder = new SpecificResponder(AvroSourceProtocol.class,
      eventCollector);
  nettyServer = new NettyServer(responder,
            new InetSocketAddress(HOSTNAME, port));
  nettyServer.start();

  // give the server a second to start
  Thread.sleep(1000L);

  properties = Maps.newHashMap();
  properties.put("channel.type", "memory");
  properties.put("channel.capacity", "200");
  properties.put("sinks", "sink1 sink2");
  properties.put("sink1.type", "avro");
  properties.put("sink2.type", "avro");
  properties.put("sink1.hostname", HOSTNAME);
  properties.put("sink1.port", String.valueOf(port));
  properties.put("sink2.hostname", HOSTNAME);
  properties.put("sink2.port", String.valueOf(port));
  properties.put("processor.type", "load_balance");

  agent = new EmbeddedAgent("test-" + serialNumber.incrementAndGet());
}
 

import org.apache.avro.ipc.Responder; //导入依赖的package包/类
private static Object getRequest(Responder responder, Message message, byte[] data)
    throws IOException {
  Schema schema = message.getRequest();
  if (ArrayUtils.isEmpty(data)) {
    // The method takes no argument; use empty data.
    data = "{}".getBytes(Constants.UTF8);
  }
  JsonNode node = Json.parse(new ByteArrayInputStream(data));
  node = AvroHelper.convertFromSimpleRecord(schema, node);
  return responder.readRequest(message.getRequest(), message.getRequest(),
      DecoderFactory.get().jsonDecoder(schema, node.toString()));
}
 

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

import org.apache.avro.ipc.Responder; //导入依赖的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.Responder; //导入依赖的package包/类
public EventCollector(final int port) {
    final Responder responder = new SpecificResponder(AvroSourceProtocol.class, this);
    nettyServer = new NettyServer(responder, new InetSocketAddress(HOSTNAME, port));
    nettyServer.start();
}
 

import org.apache.avro.ipc.Responder; //导入依赖的package包/类
public EventCollector(final int port) {
  this.eventQueue = new LinkedBlockingQueue<AvroFlumeEvent>();
  final Responder responder = new SpecificResponder(AvroSourceProtocol.class, this);
  nettyServer = new NettyServer(responder, new InetSocketAddress(HOSTNAME, port));
  nettyServer.start();
}
 

import org.apache.avro.ipc.Responder; //导入依赖的package包/类
public EventCollector(final int port) {
    final Responder responder = new SpecificResponder(AvroSourceProtocol.class, this);
    System.out.println("Collector listening on port " + port);
    nettyServer = new NettyServer(responder, new InetSocketAddress(HOSTNAME, port));
    nettyServer.start();
}