import org.jboss.netty.channel.socket.DatagramChannel; //导入依赖的package包/类
@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);
}
 

import org.jboss.netty.channel.socket.DatagramChannel; //导入依赖的package包/类
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
    LOG.warn("Caught exception during channel read", e);

    if (ctx.getChannel() != null && !(ctx.getChannel() instanceof DatagramChannel)) {
        LOG.warn("Closing channel of IngestHandler");
        ctx.getChannel().close();
    }
}
 

import org.jboss.netty.channel.socket.DatagramChannel; //导入依赖的package包/类
/**
   * Shutdown.
   */
  public void shutdown()
  {
  	log.info("Closing channels");
for (DatagramChannel channel : channels) {
	channel.close();
}
log.info("Executor shutdown");
      executorService.shutdown();     
  }
 

import org.jboss.netty.channel.socket.DatagramChannel; //导入依赖的package包/类
private void closeChannel(Channel channel) {
    if (!(channel instanceof DatagramChannel)) {
        channel.close();
    }
}
 

import org.jboss.netty.channel.socket.DatagramChannel; //导入依赖的package包/类
private Position decodeGprmc(
        DeviceSession deviceSession, String sentence, SocketAddress remoteAddress, Channel channel) {

    if (ack && channel != null && !(channel instanceof DatagramChannel)) {
        channel.write("OK1\r\n");
    }

    Parser parser = new Parser(PATTERN_GPRMC, sentence);
    if (!parser.matches()) {
        return null;
    }

    Position position = new Position();
    position.setProtocol(getProtocolName());

    if (deviceSession != null) {
        position.setDeviceId(deviceSession.getDeviceId());
    }

    DateBuilder dateBuilder = new DateBuilder()
            .setTime(parser.nextInt(0), parser.nextInt(0), parser.nextInt(0));

    position.setValid(parser.next().equals("A"));
    position.setLatitude(parser.nextCoordinate());
    position.setLongitude(parser.nextCoordinate());
    position.setSpeed(parser.nextDouble(0));
    position.setCourse(parser.nextDouble(0));

    dateBuilder.setDateReverse(parser.nextInt(0), parser.nextInt(0), parser.nextInt(0));
    position.setTime(dateBuilder.getDate());

    if (parser.hasNext(5)) {
        position.set(Position.KEY_SATELLITES, parser.nextInt());

        deviceSession = getDeviceSession(channel, remoteAddress, parser.next());
        if (deviceSession == null) {
            return null;
        }
        position.setDeviceId(deviceSession.getDeviceId());

        position.set(Position.KEY_IGNITION, parser.hasNext() && parser.next().equals("1"));
        position.set(Position.KEY_FUEL_LEVEL, parser.nextInt(0));
        position.set(Position.KEY_BATTERY, parser.nextInt());
    }

    if (parser.hasNext()) {
        String[] parameters = parser.next().split(",");
        for (int i = 1; i < parameters.length; i++) {
            position.set(Position.PREFIX_IO + i, parameters[i]);
        }
    }

    if (deviceSession != null) {
        return position;
    } else {
        this.position = position; // save position
        return null;
    }
}
 

import org.jboss.netty.channel.socket.DatagramChannel; //导入依赖的package包/类
protected void startServerBootstrap() throws Exception {
    // create non-shared worker pool
    int count = configuration.getWorkerCount() > 0 ? configuration.getWorkerCount() : NettyHelper.DEFAULT_IO_THREADS;
    workerPool = new NioDatagramWorkerPool(Executors.newCachedThreadPool(), count);

    datagramChannelFactory = new NioDatagramChannelFactory(workerPool);

    connectionlessBootstrap = new ConnectionlessBootstrap(datagramChannelFactory);
    connectionlessBootstrap.setOption("child.keepAlive", configuration.isKeepAlive());
    connectionlessBootstrap.setOption("child.tcpNoDelay", configuration.isTcpNoDelay());
    connectionlessBootstrap.setOption("reuseAddress", configuration.isReuseAddress());
    connectionlessBootstrap.setOption("child.reuseAddress", configuration.isReuseAddress());
    connectionlessBootstrap.setOption("child.connectTimeoutMillis", configuration.getConnectTimeout());
    connectionlessBootstrap.setOption("child.broadcast", configuration.isBroadcast());
    connectionlessBootstrap.setOption("sendBufferSize", configuration.getSendBufferSize());
    connectionlessBootstrap.setOption("receiveBufferSize", configuration.getReceiveBufferSize());
    // only set this if user has specified
    if (configuration.getReceiveBufferSizePredictor() > 0) {
        connectionlessBootstrap.setOption("receiveBufferSizePredictorFactory",
                new FixedReceiveBufferSizePredictorFactory(configuration.getReceiveBufferSizePredictor()));
    }
    if (configuration.getBacklog() > 0) {
        connectionlessBootstrap.setOption("backlog", configuration.getBacklog());
    }

    // set any additional netty options
    if (configuration.getOptions() != null) {
        for (Map.Entry<String, Object> entry : configuration.getOptions().entrySet()) {
            connectionlessBootstrap.setOption(entry.getKey(), entry.getValue());
        }
    }

    LOG.debug("Created ConnectionlessBootstrap {} with options: {}", connectionlessBootstrap, connectionlessBootstrap.getOptions());

    // set the pipeline factory, which creates the pipeline for each newly created channels
    connectionlessBootstrap.setPipelineFactory(pipelineFactory);

    InetSocketAddress hostAddress = new InetSocketAddress(configuration.getHost(), configuration.getPort());
    IpV4Subnet multicastSubnet = new IpV4Subnet(MULTICAST_SUBNET);

    if (multicastSubnet.contains(configuration.getHost())) {
        datagramChannel = (DatagramChannel)connectionlessBootstrap.bind(hostAddress);
        String networkInterface = configuration.getNetworkInterface() == null ? LOOPBACK_INTERFACE : configuration.getNetworkInterface();
        multicastNetworkInterface = NetworkInterface.getByName(networkInterface);
        ObjectHelper.notNull(multicastNetworkInterface, "No network interface found for '" + networkInterface + "'.");
        LOG.info("ConnectionlessBootstrap joining {}:{} using network interface: {}", new Object[]{configuration.getHost(), configuration.getPort(), multicastNetworkInterface.getName()});
        datagramChannel.joinGroup(hostAddress, multicastNetworkInterface).syncUninterruptibly();
        allChannels.add(datagramChannel);
    } else {
        LOG.info("ConnectionlessBootstrap binding to {}:{}", configuration.getHost(), configuration.getPort());
        channel = connectionlessBootstrap.bind(hostAddress);
        allChannels.add(channel);
    }
}