Java 类java.nio.channels.DatagramChannel 实例源码

public static void main(String[] args) throws IOException
{
    ByteBuffer data = ByteBuffer.wrap("TESTING DATA".getBytes());
    DatagramChannel dgChannel = DatagramChannel.open();

    for(int i = 0; i < targets.length; i++){
        data.rewind();
        SocketAddress sa = new InetSocketAddress(targets[i], port);
        System.out.println("-------------\nDG_Sending data:" +
                           "\n    remaining:" + data.remaining() +
                           "\n     position:" + data.position() +
                           "\n        limit:" + data.limit() +
                           "\n     capacity:" + data.capacity() +
                           " bytes on DG channel to " + sa);
        try {
            int n = dgChannel.send(data, sa);
            System.out.println("DG_Sent " + n + " bytes");
        } catch (IOException e) {
            //This regression test is to check vm crash only, so ioe is OK.
            e.printStackTrace();
        }
    }
    dgChannel.close();
}

@Override
protected DatagramChannel newHandle(SocketAddress localAddress) throws Exception {
    DatagramChannel ch = DatagramChannel.open();

    try {
        if (localAddress != null) {
            ch.socket().bind(localAddress);
        }

        return ch;
    } catch (Exception e) {
        // If we got an exception while binding the datagram,
        // we have to close it otherwise we will loose an handle
        ch.close();
        throw e;
    }
}

public void run () {
    if (mIsConnected) return;
    try {
        mChannel = DatagramChannel.open();
        mChannel.configureBlocking(false);
        mChannel.connect(new InetSocketAddress(mAddress, mPort));

        if (mListenerThread == null) {
            mListenerThread = new ListenerThread();
            mListenerThread.start();
            mActivityHandler.sendMessage(composeMessage(MSG_ON_CONNECT, ""));
            mIsConnected = true;
        }

        Looper.prepare();
        mSendHandler = new Handler();
        Looper.loop();

    } catch (Exception e) {
        mActivityHandler.sendMessage(composeMessage(MSG_ON_CONNECTION_FAIL, e.toString()));
    }
}

/**
     * resolves host, builds socket, returns true if it succeeds
     */
    private boolean checkSocket() {
        if ((channel != null) && channel.isOpen()) { //if(datagramSocket!=null && datagramSocket.isBound()) {
            return true;
        }
        try {
            channel = DatagramChannel.open();
            datagramSocket = channel.socket();
            datagramSocket.setReuseAddress(true);
            // disable timeout so that receive just waits for data forever (until interrupted)
//            datagramSocket.setSoTimeout(TIMEOUT_MS);
//            if (datagramSocket.getSoTimeout() != TIMEOUT_MS) {
//                log.warning("datagram socket read timeout value read=" + datagramSocket.getSoTimeout() + " which is different than timeout value of " + TIMEOUT_MS + " that we tried to set - perhaps timeout is not supported?");
//            }
            SocketAddress address = new InetSocketAddress(getPort());
            datagramSocket.bind(address);
            log.info("bound " + this);
            datagramSocket.setSoTimeout(0); // infinite timeout
            return true;
        } catch (IOException e) {
            log.warning("caught " + e + ", datagramSocket will be constructed later");
            return false;
        }
    }

/**
 * Create a new broadcaster
 *
 * @param address - multicast group address
 * @param srcAddress - address of interface we should use to broadcast.
 * @param port - udp port to use
 * @param ttl - packet ttl
 * @throws IOException
 */
public JdpBroadcaster(InetAddress address, InetAddress srcAddress, int port, int ttl)
        throws IOException, JdpException {
    this.addr = address;
    this.port = port;

    ProtocolFamily family = (address instanceof Inet6Address)
            ? StandardProtocolFamily.INET6 : StandardProtocolFamily.INET;

    channel = DatagramChannel.open(family);
    channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
    channel.setOption(StandardSocketOptions.IP_MULTICAST_TTL, ttl);

    // with srcAddress equal to null, this constructor do exactly the same as
    // if srcAddress is not passed
    if (srcAddress != null) {
        // User requests particular interface to bind to
        NetworkInterface interf = NetworkInterface.getByInetAddress(srcAddress);
        try {
            channel.bind(new InetSocketAddress(srcAddress, 0));
        } catch (UnsupportedAddressTypeException ex) {
            throw new JdpException("Unable to bind to source address");
        }
        channel.setOption(StandardSocketOptions.IP_MULTICAST_IF, interf);
    }
}

public void run() {
    try {
        DatagramChannel dc = DatagramChannel.open();
        ByteBuffer bb = ByteBuffer.allocateDirect(12);
        bb.order(ByteOrder.BIG_ENDIAN);
        bb.putInt(1).putLong(1);
        bb.flip();
        InetAddress address = InetAddress.getLocalHost();
        InetSocketAddress isa = new InetSocketAddress(address, port);
        dc.connect(isa);
        clientISA = dc.getLocalAddress();
        dc.write(bb);
    } catch (Exception ex) {
        e = ex;
    }
}

public static void main(String[] args) throws IOException
{
    ByteBuffer data = ByteBuffer.wrap("TESTING DATA".getBytes());
    DatagramChannel dgChannel = DatagramChannel.open();

    for(int i = 0; i < targets.length; i++){
        data.rewind();
        SocketAddress sa = new InetSocketAddress(targets[i], port);
        System.out.println("-------------\nDG_Sending data:" +
                           "\n    remaining:" + data.remaining() +
                           "\n     position:" + data.position() +
                           "\n        limit:" + data.limit() +
                           "\n     capacity:" + data.capacity() +
                           " bytes on DG channel to " + sa);
        try {
            int n = dgChannel.send(data, sa);
            System.out.println("DG_Sent " + n + " bytes");
        } catch (IOException e) {
            //This regression test is to check vm crash only, so ioe is OK.
            e.printStackTrace();
        }
    }
    dgChannel.close();
}

public static DatagramChannel launchWithDatagramChannel(String className, String options[], String args[])
    throws IOException
{
    DatagramChannel dc = DatagramChannel.open();
    dc.socket().bind(new InetSocketAddress(0));

    int port = dc.socket().getLocalPort();
    launch(className, options, args, Util.getFD(dc));
    dc.close();

    dc = DatagramChannel.open();
    InetAddress address = InetAddress.getLocalHost();
    if (address.isLoopbackAddress()) {
        address = InetAddress.getLoopbackAddress();
    }
    InetSocketAddress isa = new InetSocketAddress(address, port);

    dc.connect(isa);
    return dc;
}

private static void UDPEchoTest() throws IOException {
    DatagramChannel dc = Launcher.launchWithDatagramChannel(ECHO_SERVICE, null);

    String msg = "I was out saving the galaxy when your grandfather was in diapers";

    ByteBuffer bb = ByteBuffer.wrap(msg.getBytes("UTF-8"));
    dc.write(bb);

    // and receive the echo
    byte b[] = new byte[msg.length() + 100];
    DatagramPacket pkt2 = new DatagramPacket(b, b.length);
    dc.socket().setSoTimeout(5000);
    dc.socket().receive(pkt2);

    if (pkt2.getLength() != msg.length()) {
        throw new RuntimeException("Received packet of incorrect length");
    }

    dc.close();
}

public synchronized void start() throws IOException {
    if (running) {
        return;
    }
    // Start channel
    DatagramChannel channel = DatagramChannel.open();
    channel.configureBlocking(true);
    channel.socket().bind(new InetSocketAddress(port));
    running = true;
    // Start server thread
    new ServerThread(channel).start();
    log.info("DNS server started");
}

@Override
protected DatagramChannel open(SocketAddress localAddress) throws Exception {
    final DatagramChannel c = DatagramChannel.open();
    boolean success = false;
    try {
        new NioDatagramSessionConfig(c).setAll(getSessionConfig());
        c.configureBlocking(false);
        c.socket().bind(localAddress);
        c.register(selector, SelectionKey.OP_READ);
        success = true;
    } finally {
        if (!success) {
            close(c);
        }
    }

    return c;
}

@Override
protected boolean isReadable(DatagramChannel handle) {
    SelectionKey key = handle.keyFor(selector);

    if ((key == null) || (!key.isValid())) {
        return false;
    }

    return key.isReadable();
}

@Override
protected boolean isWritable(DatagramChannel handle) {
    SelectionKey key = handle.keyFor(selector);

    if ((key == null) || (!key.isValid())) {
        return false;
    }

    return key.isWritable();
}

@Override
protected SocketAddress localAddress(DatagramChannel handle) throws Exception {
    InetSocketAddress inetSocketAddress = (InetSocketAddress) handle.socket().getLocalSocketAddress();
    InetAddress inetAddress = inetSocketAddress.getAddress();

    if ((inetAddress instanceof Inet6Address) && (((Inet6Address) inetAddress).isIPv4CompatibleAddress())) {
        // Ugly hack to workaround a problem on linux : the ANY address is always converted to IPV6
        // even if the original address was an IPV4 address. We do store the two IPV4 and IPV6
        // ANY address in the map.
        byte[] ipV6Address = ((Inet6Address) inetAddress).getAddress();
        byte[] ipV4Address = new byte[4];

        for (int i = 0; i < 4; i++) {
            ipV4Address[i] = ipV6Address[12 + i];
        }

        InetAddress inet4Adress = Inet4Address.getByAddress(ipV4Address);
        return new InetSocketAddress(inet4Adress, inetSocketAddress.getPort());
    } else {
        return inetSocketAddress;
    }
}

@Override
protected void close(DatagramChannel handle) throws Exception {
    SelectionKey key = handle.keyFor(selector);

    if (key != null) {
        key.cancel();
    }

    handle.disconnect();
    handle.close();
}

/**
 * send a reply-acknowledgement (6,2,3), sends it doing a busy write, the ACK is so small
 * that it should always go to the buffer
 * @param key
 * @param channel
 */
protected void sendAck(SelectionKey key, WritableByteChannel channel, byte[] command, SocketAddress udpaddr) {
    try {

        ByteBuffer buf = ByteBuffer.wrap(command);
        int total = 0;
        if (channel instanceof DatagramChannel) {
            DatagramChannel dchannel = (DatagramChannel)channel;
            //were using a shared channel, document says its thread safe
            //TODO check optimization, one channel per thread?
            while ( total < command.length ) {
                total += dchannel.send(buf, udpaddr);
            }
        } else {
            while ( total < command.length ) {
                total += channel.write(buf);
            }
        }
        if (log.isTraceEnabled()) {
            log.trace("ACK sent to " +
                    ( (channel instanceof SocketChannel) ?
                      ((SocketChannel)channel).socket().getInetAddress() :
                      ((DatagramChannel)channel).socket().getInetAddress()));
        }
    } catch ( java.io.IOException x ) {
        log.warn("Unable to send ACK back through channel, channel disconnected?: "+x.getMessage());
    }
}

protected void bind() throws IOException {
    // allocate an unbound server socket channel
    serverChannel = ServerSocketChannel.open();
    // Get the associated ServerSocket to bind it with
    ServerSocket serverSocket = serverChannel.socket();
    // create a new Selector for use below
    synchronized (Selector.class) {
        // Selector.open() isn't thread safe
        // http://bugs.sun.com/view_bug.do?bug_id=6427854
        // Affects 1.6.0_29, fixed in 1.7.0_01
        this.selector.set(Selector.open());
    }
    // set the port the server channel will listen to
    //serverSocket.bind(new InetSocketAddress(getBind(), getTcpListenPort()));
    bind(serverSocket,getPort(),getAutoBind());
    // set non-blocking mode for the listening socket
    serverChannel.configureBlocking(false);
    // register the ServerSocketChannel with the Selector
    serverChannel.register(this.selector.get(), SelectionKey.OP_ACCEPT);

    //set up the datagram channel
    if (this.getUdpPort()>0) {
        datagramChannel = DatagramChannel.open();
        configureDatagraChannel();
        //bind to the address to avoid security checks
        bindUdp(datagramChannel.socket(),getUdpPort(),getAutoBind());
    }
}

public static void main(String[] args) throws Exception {
    InetAddress lh = InetAddress.getLocalHost();
    SocketAddress remote = new InetSocketAddress(lh, 1234);

    DatagramSocket ds = null;
    DatagramChannel dc = null;
    try {

        ds = new DatagramSocket();
        dc = DatagramChannel.open().bind(new InetSocketAddress(0));
        check(ds, dc);

        ds.connect(remote);
        dc.connect(remote);
        check(ds, dc);

        ds.disconnect();
        dc.disconnect();
        check(ds, dc);

        // repeat tests using socket adapter
        ds.connect(remote);
        dc.socket().connect(remote);
        check(ds, dc);

        ds.disconnect();
        dc.socket().disconnect();
        check(ds, dc);

   } finally {
        if (ds != null) ds.close();
        if (dc != null) dc.close();
   }
}

private void close(DatagramChannel udp) {
    if (udp != null) {
        try {
            udp.close();
        } catch (IOException e) {
            // No-op.
        }
    }
}

@Before
@Override
public void setUp() throws Exception {
    super.setUp();

    StatsdMetricsConfig cfg = new StatsdMetricsConfig()
        .withHost(InetAddress.getLocalHost().getHostAddress())
        .withPort(testPort)
        .withBatchSize(TEST_BATCH_SIZE);

    publisher = new StatsdMetricsPublisher(cfg) {
        @Override
        void doWrite(DatagramChannel udp, List<ByteBuffer> packets) throws IOException {
            if (writeDelegate == null) {
                super.doWrite(udp, packets);
            } else {
                writeDelegate.write(udp, packets, () ->
                    super.doWrite(udp, packets)
                );
            }
        }
    };
}

/**
 * 
 * @param port
 * @param addr
 * @return
 * @throws Exception
 */
public static DatagramChannel createDatagramChannel(int port, InetAddress addr) throws Exception {
    final DatagramChannel channel = DatagramChannel.open();
    if (addr == null) {
        channel.bind(new InetSocketAddress(port));
    }
    else {
        channel.bind(new InetSocketAddress(addr, port));
    }

    return channel;
}

public void start() throws Exception{
    if(antenna != null){
        throw new Exception("antenna is not null, may have run before");
    }
    antenna = DatagramChannel.open();
    antenna.socket().bind(new InetSocketAddress(port));
    System.out.println("udp connector port:"+port);
    //non-blocking
    antenna.configureBlocking(false);
    antenna.socket().setReceiveBufferSize(1024*1024*PropertyUtil.getPropertyInt("CLIENT_UDP_BUFFER_RECEIVE"));
    antenna.socket().setSendBufferSize(1024*1024*PropertyUtil.getPropertyInt("CLIENT_UDP_BUFFER_SEND"));

    this.receiver = new Receiver(antenna);
    this.receiver.init();
    this.sender = new Sender(antenna);
    this.sender.init();

    this.senderThread = new Thread(sender,"AsynUdpConnector-sender");
    this.receiverThread = new Thread(receiver,"AsynUdpConnector-receiver");
    this.receiverThread.start();
    this.senderThread.start();
}

/** Opens or reopens the channel. If the channel is not open, open it. If it is open then just return.
 *
 * @throws IOException
 */
public void open() throws IOException {
    if (isOpen) {
        log.info("already open, not opening");
        return;
    }
    channel = DatagramChannel.open();
    socket = channel.socket(); // bind to any available port because we will be sending datagrams with included host:port info
    socket.setTrafficClass(0x10 + 0x08); // low delay
    socket.setSendBufferSize(BUFFER_SIZE_BYTES); // TODO chyanging buffer size later doesn't change this initial value
    allocateBuffers();
    consumerThread = new Thread(new Consumer(exchanger, initialFullBuffer));
    consumerThread.setName("UDPMessageSender");
    consumerThread.setPriority(Thread.NORM_PRIORITY + 1);
    consumerThread.start();
    isOpen = true;
    log.info("opened UDPMessageSender on local port=" + socket.getLocalPort() + " with bufferSize=" + BUFFER_SIZE_BYTES);
}

/**
 * Create a new broadcaster
 *
 * @param address - multicast group address
 * @param srcAddress - address of interface we should use to broadcast.
 * @param port - udp port to use
 * @param ttl - packet ttl
 * @throws IOException
 */
public JdpBroadcaster(InetAddress address, InetAddress srcAddress, int port, int ttl)
        throws IOException, JdpException {
    this.addr = address;
    this.port = port;

    ProtocolFamily family = (address instanceof Inet6Address)
            ? StandardProtocolFamily.INET6 : StandardProtocolFamily.INET;

    channel = DatagramChannel.open(family);
    channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
    channel.setOption(StandardSocketOptions.IP_MULTICAST_TTL, ttl);

    // with srcAddress equal to null, this constructor do exactly the same as
    // if srcAddress is not passed
    if (srcAddress != null) {
        // User requests particular interface to bind to
        NetworkInterface interf = NetworkInterface.getByInetAddress(srcAddress);
        try {
            channel.bind(new InetSocketAddress(srcAddress, 0));
        } catch (UnsupportedAddressTypeException ex) {
            throw new JdpException("Unable to bind to source address");
        }
        channel.setOption(StandardSocketOptions.IP_MULTICAST_IF, interf);
    }
}

private void closeChannel(DatagramChannel channel)
{
    try
    {
        channel.close();
    }
    catch (IOException e)
    {
        // Ignore
    }
}

/**
 * send a reply-acknowledgement (6,2,3), sends it doing a busy write, the ACK is so small
 * that it should always go to the buffer
 * @param key
 * @param channel
 */
protected void sendAck(SelectionKey key, WritableByteChannel channel, byte[] command, SocketAddress udpaddr) {
    try {

        ByteBuffer buf = ByteBuffer.wrap(command);
        int total = 0;
        if (channel instanceof DatagramChannel) {
            DatagramChannel dchannel = (DatagramChannel)channel;
            //were using a shared channel, document says its thread safe
            //TODO check optimization, one channel per thread?
            while ( total < command.length ) {
                total += dchannel.send(buf, udpaddr);
            }
        } else {
            while ( total < command.length ) {
                total += channel.write(buf);
            }
        }
        if (log.isTraceEnabled()) {
            log.trace("ACK sent to " +
                    ( (channel instanceof SocketChannel) ?
                      ((SocketChannel)channel).socket().getInetAddress() :
                      ((DatagramChannel)channel).socket().getInetAddress()));
        }
    } catch ( java.io.IOException x ) {
        log.warn("Unable to send ACK back through channel, channel disconnected?: "+x.getMessage());
    }
}

static void wakeupWhenBound(final DatagramChannel dc) {
    Runnable wakeupTask = new Runnable() {
        public void run() {
            try {
                // poll for local address
                InetSocketAddress local;
                do {
                    Thread.sleep(50);
                    local = (InetSocketAddress)dc.getLocalAddress();
                } while (local == null);

                // send message to channel to wakeup receiver
                DatagramChannel sender = DatagramChannel.open();
                try {
                    ByteBuffer bb = ByteBuffer.wrap("hello".getBytes());
                    InetAddress lh = InetAddress.getLocalHost();
                    SocketAddress target =
                        new InetSocketAddress(lh, local.getPort());
                    sender.send(bb, target);
                } finally {
                    sender.close();
                }

            } catch (Exception x) {
                x.printStackTrace();
            }
        }};
    new Thread(wakeupTask).start();
}

private static void test() throws Exception {
    DatagramChannel dc = DatagramChannel.open();
    InetAddress localHost = InetAddress.getLocalHost();
    dc.bind(new InetSocketAddress(localHost, 0));

    Server server = new Server(dc.getLocalAddress());
    Thread serverThread = new Thread(server);
    serverThread.start();

    try {
        InetSocketAddress isa = new InetSocketAddress(localHost, server.port());
        dc.connect(isa);

        ByteBuffer bb = ByteBuffer.allocateDirect(12);
        bb.order(ByteOrder.BIG_ENDIAN);
        bb.putInt(1).putLong(1);
        bb.flip();

        dc.write(bb);
        bb.rewind();
        dc.write(bb);
        bb.rewind();
        dc.write(bb);

        Thread.sleep(2000);

        serverThread.interrupt();
        server.throwException();
    } finally {
        dc.close();
    }
}

@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    DatagramChannel ch = javaChannel();
    UkcpServerChannelConfig config = config();
    RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();

    ByteBuf data = allocHandle.allocate(config.getAllocator());
    allocHandle.attemptedBytesRead(data.writableBytes());
    boolean free = true;
    try {
        ByteBuffer nioData = data.internalNioBuffer(data.writerIndex(), data.writableBytes());
        int pos = nioData.position();
        InetSocketAddress remoteAddress = (InetSocketAddress) ch.receive(nioData);
        if (remoteAddress == null) {
            return 0;
        }

        allocHandle.lastBytesRead(nioData.position() - pos);
        buf.add(UkcpPacket.newInstance(data.writerIndex(data.writerIndex() + allocHandle.lastBytesRead()),
                remoteAddress));
        free = false;
        return 1;
    } catch (Throwable cause) {
        PlatformDependent.throwException(cause);
        return -1;
    } finally {
        if (free) {
            data.release();
        }
    }
}

static void wakeupWhenBound(final DatagramChannel dc) {
    Runnable wakeupTask = new Runnable() {
        public void run() {
            try {
                // poll for local address
                InetSocketAddress local;
                do {
                    Thread.sleep(50);
                    local = (InetSocketAddress)dc.getLocalAddress();
                } while (local == null);

                // send message to channel to wakeup receiver
                DatagramChannel sender = DatagramChannel.open();
                try {
                    ByteBuffer bb = ByteBuffer.wrap("hello".getBytes());
                    InetAddress lh = InetAddress.getLocalHost();
                    SocketAddress target =
                        new InetSocketAddress(lh, local.getPort());
                    sender.send(bb, target);
                } finally {
                    sender.close();
                }

            } catch (Exception x) {
                x.printStackTrace();
            }
        }};
    new Thread(wakeupTask).start();
}

public static void main(String[] args) throws Exception {
    InetAddress lh = InetAddress.getLocalHost();
    SocketAddress remote = new InetSocketAddress(lh, 1234);

    DatagramSocket ds = null;
    DatagramChannel dc = null;
    try {

        ds = new DatagramSocket();
        dc = DatagramChannel.open().bind(new InetSocketAddress(0));
        check(ds, dc);

        ds.connect(remote);
        dc.connect(remote);
        check(ds, dc);

        ds.disconnect();
        dc.disconnect();
        check(ds, dc);

        // repeat tests using socket adapter
        ds.connect(remote);
        dc.socket().connect(remote);
        check(ds, dc);

        ds.disconnect();
        dc.socket().disconnect();
        check(ds, dc);

   } finally {
        if (ds != null) ds.close();
        if (dc != null) dc.close();
   }
}

private void openChannel() {
    try {
        channel = DatagramChannel.open();
        channel.socket().bind(new InetSocketAddress(DEFAULT_PORT));
    } catch (IOException e) {
        e.printStackTrace();
    }
}

static void test2() {
    for (int i=0; i<11000; i++) {
        try {
            DatagramChannel sc = DatagramChannel.open();
        } catch (Exception e) {
            // Presumably "Too many open files"
        }
    }
}

public static void main(String[] args) throws Exception {

        // Load necessary classes ahead of time
        DatagramChannel dc = DatagramChannel.open();
        Exception se = new SocketException();
        SelectorProvider sp = SelectorProvider.provider();
        Pipe p = sp.openPipe();
        ServerSocketChannel ssc = ServerSocketChannel.open();

        test1();
        test2();
        test3();
        test4();
    }

private static void doIt(DatagramChannel dc) throws IOException {
    ByteBuffer bb = ByteBuffer.allocate(1024);
    SocketAddress sa = dc.receive(bb);
    bb.flip();
    dc.send(bb, sa);
    dc.close();
}

static void check(DatagramSocket ds, DatagramChannel dc) {
    InetAddress expected = ds.getLocalAddress();
    InetAddress actual = dc.socket().getLocalAddress();
    // okay if one bound to 0.0.0.0 and the other to ::0
    if ((expected.isAnyLocalAddress() != actual.isAnyLocalAddress()) &&
        !expected.equals(actual))
    {
        throw new RuntimeException("Expected: " + expected + ", actual: " + actual);
    }
}

/**
 * {@inheritDoc}
 */
@Override
protected boolean isBrokenConnection() throws IOException {
    // A flag set to true if we find a broken session
    boolean brokenSession = false;

    synchronized (selector) {
        // Get the selector keys
        Set<SelectionKey> keys = selector.keys();

        // Loop on all the keys to see if one of them
        // has a closed channel
        for (SelectionKey key : keys) {
            SelectableChannel channel = key.channel();

            if ((((channel instanceof DatagramChannel) && !((DatagramChannel) channel).isConnected()))
                    || ((channel instanceof SocketChannel) && !((SocketChannel) channel).isConnected())) {
                // The channel is not connected anymore. Cancel
                // the associated key then.
                key.cancel();

                // Set the flag to true to avoid a selector switch
                brokenSession = true;
            }
        }
    }

    return brokenSession;
}

@Override
protected boolean connect(DatagramChannel handle, SocketAddress remoteAddress) throws Exception {
    handle.connect(remoteAddress);
    return true;
}

@Override
protected void close(DatagramChannel handle) throws Exception {
    handle.disconnect();
    handle.close();
}

/**
 * Creates a new acceptor-side session instance.
 */
NioDatagramSession(IoService service, DatagramChannel channel, IoProcessor<NioSession> processor,
        SocketAddress remoteAddress) {
    super(processor, service, channel);
    config = new NioDatagramSessionConfig(channel);
    config.setAll(service.getSessionConfig());
    this.remoteAddress = (InetSocketAddress) remoteAddress;
    this.localAddress = (InetSocketAddress) channel.socket().getLocalSocketAddress();
}