Smart-Socket Connection Management: Session Lifecycle, Limits, and Quality Monitoring

9.1 Session Lifecycle Management

In smart‑socket each client connection is represented by an AioSession object. The session progresses through a defined state machine.

9.1.1 Session State Machine

NEW_SESSION – connection established and AioSession created

SESSION_ENABLED – normal operation

SESSION_CLOSING – session is being closed

SESSION_CLOSED – session has been closed

INPUT_SHUTDOWN – read channel shut down

异常状态 – processing, decoding or I/O exceptions

9.1.2 State Event Listener

public class ConnectionStateListener<T> extends AbstractMessageProcessor<T> {
    @Override
    public void stateEvent0(AioSession session, StateMachineEnum stateMachineEnum, Throwable throwable) {
        switch (stateMachineEnum) {
            case NEW_SESSION:
                System.out.println("新连接建立: " + session.getSessionID());
                break;
            case SESSION_CLOSED:
                System.out.println("连接关闭: " + session.getSessionID());
                break;
            case PROCESS_EXCEPTION:
                System.err.println("处理异常: " + session.getSessionID());
                throwable.printStackTrace();
                break;
            default:
                // other state events
                break;
        }
    }
    @Override
    public void process0(AioSession session, T msg) {
        // message handling logic
    }
}

9.2 Connection Count Monitoring and Limiting

9.2.1 Using MonitorPlugin

public class MonitorExample {
    public static void main(String[] args) throws IOException {
        MessageProcessor<String> processor = new AbstractMessageProcessor<String>() {
            @Override
            public void process0(AioSession session, String msg) {
                System.out.println("收到消息: " + msg);
            }
            @Override
            public void stateEvent0(AioSession session, StateMachineEnum state, Throwable throwable) {
                // handle state events
            }
        };
        // add monitor plugin, output stats every 10 seconds
        processor.addPlugin(new MonitorPlugin<String>(10));
        AioQuickServer<String> server = new AioQuickServer<>(8888, new StringProtocol(), processor);
        server.start();
    }
}

Sample output (every 10 seconds):

-----10seconds ----
inflow:      0.00125(MB)
outflow:     0.00078(MB)
process fail:   0
process count:  25
process total:  1250
read count: 50   write count:    25
connect count:  5
disconnect count:   2
online count:   3
connected total:    25
Requests/sec:   2.5
Transfer/sec:   1.25E-4(MB)

9.2.2 Custom Connection Limit Plugin

public class ConnectionLimitPlugin<T> extends AbstractPlugin<T> {
    private final int maxConnections;
    private final AtomicInteger currentConnections = new AtomicInteger(0);
    public ConnectionLimitPlugin(int maxConnections) {
        this.maxConnections = maxConnections;
    }
    @Override
    public AsynchronousSocketChannel shouldAccept(AsynchronousSocketChannel channel) {
        if (currentConnections.get() >= maxConnections) {
            System.err.println("连接数已达上限: " + maxConnections);
            try { channel.close(); } catch (IOException e) { e.printStackTrace(); }
            return null; // reject connection
        }
        currentConnections.incrementAndGet();
        return channel;
    }
    @Override
    public void stateEvent(StateMachineEnum stateMachineEnum, AioSession session, Throwable throwable) {
        if (stateMachineEnum == StateMachineEnum.SESSION_CLOSED) {
            currentConnections.decrementAndGet();
        }
    }
}

Usage example sets the maximum connections to 100:

public class ConnectionLimitExample {
    public static void main(String[] args) throws IOException {
        MessageProcessor<String> processor = new AbstractMessageProcessor<String>() {
            @Override
            public void process0(AioSession session, String msg) {
                System.out.println("收到消息: " + msg);
            }
        };
        // limit to 100 concurrent connections
        processor.addPlugin(new ConnectionLimitPlugin<>(100));
        AioQuickServer<String> server = new AioQuickServer<>(8888, new StringProtocol(), processor);
        server.start();
    }
}

9.3 Connection State Tracking and Statistics

9.3.1 Session Attribute Management

public class SessionAttributeExample {
    public static void main(String[] args) throws IOException {
        MessageProcessor<String> processor = new AbstractMessageProcessor<String>() {
            @Override
            public void process0(AioSession session, String msg) {
                Long createTime = session.getAttachment();
                long duration = System.currentTimeMillis() - createTime;
                System.out.println("连接持续时间: " + duration + "ms");
                System.out.println("收到消息: " + msg);
            }
            @Override
            public void stateEvent0(AioSession session, StateMachineEnum state, Throwable throwable) {
                if (state == StateMachineEnum.NEW_SESSION) {
                    session.setAttachment(System.currentTimeMillis());
                }
            }
        };
        AioQuickServer<String> server = new AioQuickServer<>(8888, new StringProtocol(), processor);
        server.start();
    }
}

9.3.2 Connection Information Query

public class ConnectionInfoExample extends AbstractMessageProcessor<String> {
    @Override
    public void process0(AioSession session, String msg) {
        try {
            InetSocketAddress localAddress = session.getLocalAddress();
            InetSocketAddress remoteAddress = session.getRemoteAddress();
            System.out.println("本地地址: " + localAddress);
            System.out.println("远程地址: " + remoteAddress);
            System.out.println("会话ID: " + session.getSessionID());
            System.out.println("会话状态: " + (session.isInvalid() ? "无效" : "有效"));
        } catch (IOException e) {
            e.printStackTrace();
        }
        System.out.println("收到消息: " + msg);
    }
}

9.4 Connection Quality Assessment

9.4.1 Using NetMonitor

public class NetworkMonitor implements NetMonitor {
    private final Map<String, Long> sessionReadStats = new ConcurrentHashMap<>();
    private final Map<String, Long> sessionWriteStats = new ConcurrentHashMap<>();
    @Override
    public AsynchronousSocketChannel shouldAccept(AsynchronousSocketChannel channel) { return channel; }
    @Override
    public void afterRead(AioSession session, int readSize) {
        sessionReadStats.merge(session.getSessionID(), (long) readSize, Long::sum);
        System.out.println("会话 " + session.getSessionID() + " 读取 " + readSize + " 字节");
    }
    @Override
    public void beforeRead(AioSession session) { /* pre‑read logic */ }
    @Override
    public void afterWrite(AioSession session, int writeSize) {
        sessionWriteStats.merge(session.getSessionID(), (long) writeSize, Long::sum);
        System.out.println("会话 " + session.getSessionID() + " 写入 " + writeSize + " 字节");
    }
    @Override
    public void beforeWrite(AioSession session) { /* pre‑write logic */ }
    public long getTotalReadBytes(String sessionId) { return sessionReadStats.getOrDefault(sessionId, 0L); }
    public long getTotalWriteBytes(String sessionId) { return sessionWriteStats.getOrDefault(sessionId, 0L); }
}

Example attaching the monitor to a server:

public class NetworkMonitorExample {
    public static void main(String[] args) throws IOException {
        NetworkMonitor monitor = new NetworkMonitor();
        MessageProcessor<String> processor = new AbstractMessageProcessor<String>() {
            @Override
            public void process0(AioSession session, String msg) {
                System.out.println("收到消息: " + msg);
                try {
                    byte[] response = ("Echo: " + msg).getBytes();
                    session.writeBuffer().writeInt(response.length);
                    session.writeBuffer().write(response);
                    session.writeBuffer().flush();
                } catch (IOException e) { e.printStackTrace(); }
            }
        };
        AioQuickServer<String> server = new AioQuickServer<>(8888, new StringProtocol(), processor);
        server.setMonitor(monitor);
        server.start();
    }
}

9.4.2 Blacklist Management

public class BlackListExample {
    public static void main(String[] args) throws IOException {
        MessageProcessor<String> processor = new AbstractMessageProcessor<String>() {
            @Override
            public void process0(AioSession session, String msg) {
                System.out.println("收到消息: " + msg);
            }
        };
        BlackListPlugin<String> blackListPlugin = new BlackListPlugin<>();
        // reject connections from 192.168.1.100
        blackListPlugin.addRule(address -> !"192.168.1.100".equals(address.getAddress().getHostAddress()));
        // reject ports between 10000 and 20000
        blackListPlugin.addRule(address -> {
            int port = address.getPort();
            return port < 10000 || port > 20000;
        });
        processor.addPlugin(blackListPlugin);
        AioQuickServer<String> server = new AioQuickServer<>(8888, new StringProtocol(), processor);
        server.start();
    }
}