java - XNIO : How to get notified when a connection is closed?

Question

我正在尝试使用 XNIO 编写服务器。我希望在客户端关闭与服务器的连接时收到通知。

XnioWorker worker = Xnio.getInstance().createWorker(OptionMap.EMPTY);

AcceptingChannel<StreamConnection> server = worker.createStreamConnectionServer(
        new InetSocketAddress(0), acceptingChannel -> {
            try {
                StreamConnection connection = acceptingChannel.accept();
                if (connection != null) {
                    System.out.println("accepted");

                    connection.setCloseListener(channel -> System.out.println("closed"));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }, OptionMap.EMPTY);
server.resumeAccepts();

我为每个新连接附加一个密切监听器。问题是它永远不会被调用。

例如，如果我连接到服务器

Socket socket = new Socket();
socket.connect(server.getLocalAddress());

然后断开连接

socket.close();
System.out.println("client connection closed");

我在控制台中只能看到以下几行

accepted
client connection closed

但是没有来自密切监听者的消息。有没有办法获得此类通知？

更新

好的，我可以了解对等方何时关闭写入。然后我可以关闭源 channel :

ConduitStreamSourceChannel sourceChannel = connection.getSourceChannel();
sourceChannel.setReadListener(channel -> {
    try {
        int byteNum = channel.read(buffer);
        if (byteNum == -1) channel.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
});
sourceChannel.resumeReads();

但是接收 channel 又如何呢？也就是说，如何区分对端是否完全关闭连接

socket.close();

或者半关闭它？

socket.getOutputStream().close();

更新2

读取时无需显式关闭源 channel 。 XNIO 自己完成此操作。

ConduitStreamSourceChannel sourceChannel = connection.getSourceChannel();
sourceChannel.setReadListener(channel -> {
    try {
        channel.read(buffer);
    } catch (IOException e) {
        e.printStackTrace();
    }
});
sourceChannel.resumeReads();

问题仍然存在:我如何才能知道对等点何时完全断开连接或仅关闭写入？

Answer 1

尝试将监听器附加到下游 channel :

connection.getSourceChannel()
    .setCloseListener((ch)->System.out.println("client connection closed"));

//At some level you need to make sure your source channel is enabling reads.
connection.getSourceChannel().resumeReads();

附加到 StreamConnection 的“关闭”监听器仅当 SourceChannel(下游 channel )和 SinkChannel(上游 channel )都关闭时才会被调用。

编辑:(完整示例)

服务器:

XnioWorker worker = Xnio.getInstance().createWorker(OptionMap.EMPTY);

AcceptingChannel<StreamConnection> server = worker.createStreamConnectionServer(
    new InetSocketAddress(0), acceptingChannel -> {
        try {
            StreamConnection connection = acceptingChannel.accept();
            if (connection != null) {
                System.out.println("accepted");

                connection.setCloseListener(channel -> System.out.println("closed"));
            }
            ConduitStreamSourceChannel sourceChannel = connection.getSourceChannel();
            ConduitStreamSinkChannel sinkChannel = connection.getSinkChannel();

            sourceChannel
                        .setCloseListener((t)->System.out.println("Source Closed"));
            sinkChannel
                        .setCloseListener((t)->System.out.println("Sink Closed"));

            sourceChannel.setReadListener((channel)->{
                    try{
                        ByteBuffer bb = ByteBuffer.allocate(15);
                        StringBuilder builder = new StringBuilder();
                        while(sourceChannel.read(bb) > 0){
                            bb.flip();
                            while(bb.position() < bb.limit()){
                                builder.append((char)bb.get());
                            }
                            bb.clear();
                        }
                        System.out.println(builder.toString());
                    }catch(Exception e){
                        e.printStackTrace();
                    }
                });

            sourceChannel.resumeReads();
            sinkChannel.resumeWrites();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }, OptionMap.EMPTY);

server.resumeAccepts();

客户:

try(Socket client = new Socket(Server.SERVER_HOST,Server.SERVER_PORT);
        PrintWriter echoWriter = new PrintWriter(client.getOutputStream(),true);)
    {
        echoWriter.println("Hello");
        Thread.sleep(1000);
        client.close();
        //your Server Source Channel close listener will notify you from here
        Thread.sleep(1000);
    }catch(Exception e){
        e.printStackTrace();
    }