ClusterStatusListener.java 文件源码

@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);
}