Java 类org.jboss.netty.channel.socket.oio.OioDatagramChannelFactory 实例源码

@Override
public void start() {
  // setup Netty server
  ConnectionlessBootstrap serverBootstrap = new ConnectionlessBootstrap(
      new OioDatagramChannelFactory(Executors.newCachedThreadPool()));
  final syslogHandler handler = new syslogHandler();
  handler.setFormater(formaterProp);
  handler.setKeepFields(keepFields);
  serverBootstrap.setOption("receiveBufferSizePredictorFactory",
      new AdaptiveReceiveBufferSizePredictorFactory(DEFAULT_MIN_SIZE,
          DEFAULT_INITIAL_SIZE, maxsize));
  serverBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    @Override
    public ChannelPipeline getPipeline() {
      return Channels.pipeline(handler);
    }
  });

  if (host == null) {
    nettyChannel = serverBootstrap.bind(new InetSocketAddress(port));
  } else {
    nettyChannel = serverBootstrap.bind(new InetSocketAddress(host, port));
  }

  sourceCounter.start();
  super.start();
}

@Override
public void connect(Configuration conf) throws IOException {
  // Can't be NiO with Netty today => not implemented in Netty.
  DatagramChannelFactory f = new OioDatagramChannelFactory(service);

  ConnectionlessBootstrap b = new ConnectionlessBootstrap(f);
  b.setPipeline(Channels.pipeline(
      new ProtobufDecoder(ClusterStatusProtos.ClusterStatus.getDefaultInstance()),
      new ClusterStatusHandler()));

  String mcAddress = conf.get(HConstants.STATUS_MULTICAST_ADDRESS,
      HConstants.DEFAULT_STATUS_MULTICAST_ADDRESS);
  String bindAddress = conf.get(HConstants.STATUS_MULTICAST_BIND_ADDRESS,
    HConstants.DEFAULT_STATUS_MULTICAST_BIND_ADDRESS);
  int port = conf.getInt(HConstants.STATUS_MULTICAST_PORT,
      HConstants.DEFAULT_STATUS_MULTICAST_PORT);

  channel = (DatagramChannel) b.bind(new InetSocketAddress(bindAddress, port));

  channel.getConfig().setReuseAddress(true);

  InetAddress ina;
  try {
    ina = InetAddress.getByName(mcAddress);
  } catch (UnknownHostException e) {
    throw new IOException("Can't connect to " + mcAddress, e);
  }
  channel.joinGroup(ina);
}

@Override
public void connect(Configuration conf) throws IOException {
  // Can't be NiO with Netty today => not implemented in Netty.
  DatagramChannelFactory f = new OioDatagramChannelFactory(service);

  ConnectionlessBootstrap b = new ConnectionlessBootstrap(f);
  b.setPipeline(Channels.pipeline(
      new ProtobufDecoder(ClusterStatusProtos.ClusterStatus.getDefaultInstance()),
      new ClusterStatusHandler()));

  String mcAddress = conf.get(HConstants.STATUS_MULTICAST_ADDRESS,
      HConstants.DEFAULT_STATUS_MULTICAST_ADDRESS);
  String bindAddress = conf.get(HConstants.STATUS_MULTICAST_BIND_ADDRESS,
    HConstants.DEFAULT_STATUS_MULTICAST_BIND_ADDRESS);
  int port = conf.getInt(HConstants.STATUS_MULTICAST_PORT,
      HConstants.DEFAULT_STATUS_MULTICAST_PORT);

  channel = (DatagramChannel) b.bind(new InetSocketAddress(bindAddress, port));

  channel.getConfig().setReuseAddress(true);

  InetAddress ina;
  try {
    ina = InetAddress.getByName(mcAddress);
  } catch (UnknownHostException e) {
    throw new IOException("Can't connect to " + mcAddress, e);
  }
  channel.joinGroup(ina);
}

/**
   * Start sending DHCPv6 SOLICITs.
   */
  public void start()
  {
    DatagramChannelFactory factory = 
        new OioDatagramChannelFactory(Executors.newCachedThreadPool());

    server = new InetSocketAddress(serverAddr, serverPort);
    client = new InetSocketAddress(clientPort);

ChannelPipeline pipeline = Channels.pipeline();
      pipeline.addLast("logger", new LoggingHandler());
      pipeline.addLast("encoder", new DhcpV6ChannelEncoder());
      pipeline.addLast("decoder", new DhcpV6ChannelDecoder(client, false));
      pipeline.addLast("executor", new ExecutionHandler(
            new OrderedMemoryAwareThreadPoolExecutor(16, 1048576, 1048576)));
      pipeline.addLast("handler", this);

      channel = factory.newChannel(pipeline);
channel.getConfig().setNetworkInterface(mcastNetIf);
    channel.bind(client);

    for (int i=1; i<=numRequests; i++) {
        executor.execute(new ClientMachine(i));
    }

    synchronized (syncDone) {
        long ms = timeout * 1000;
        try {
            log.info("Waiting total of " + timeout + " milliseconds for completion");
            syncDone.wait(ms);
        }
        catch (InterruptedException ex) {
            log.error("Interrupted", ex);
        }
}

log.info("Complete: solicitsSent=" + solicitsSent +
        " advertisementsReceived=" + advertisementsReceived +
        " requestsSent=" + requestsSent +
        " requestRepliesReceived=" + requestRepliesReceived +
        " releasesSent=" + releasesSent +
        " releaseRepliesReceived=" + releaseRepliesReceived +
        " elapsedTime=" + (endTime - startTime) + "ms");

    log.info("Shutting down executor...");
    executor.shutdownNow();
    log.info("Closing channel...");
    channel.close();
    log.info("Done.");
    if ((solicitsSent.get() == advertisementsReceived.get()) &&
            (requestsSent.get() == requestRepliesReceived.get()) &&
            (releasesSent.get() == releaseRepliesReceived.get())) {

        System.exit(0);
    }
    else {
        System.exit(1);
    }
  }