Использование вебсокетов в приложениях Java с Netty

public class WebSocketServerHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
        // Вызывается при получении WebSocket-сообщения
        if (frame instanceof TextWebSocketFrame) {
            // Обработка текстового сообщения
            String text = ((TextWebSocketFrame) frame).text();
            // Бизнес-логика...
        } else if (frame instanceof BinaryWebSocketFrame) {
            // Обработка бинарного сообщения
        }
    }
 
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        // Вызывается при добавлении обработчика к каналу
        System.out.println("Клиент подключился: " + ctx.channel().id());
    }
 
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        // Вызывается при удалении обработчика из канала
        System.out.println("Клиент отключился: " + ctx.channel().id());
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Обработка исключений
        cause.printStackTrace();
        ctx.close();
    }
}

protected void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    
    // Обработчики входящих сообщений (inbound)
    pipeline.addLast(new HttpServerCodec()); // HTTP-кодек
    pipeline.addLast(new HttpObjectAggregator(65536)); // Объединяет фрагментированые HTTP-сообщения
    pipeline.addLast(new WebSocketServerProtocolHandler("/websocket")); // Обработка WebSocket-рукопожатия
    
    // Бизнес-логика
    pipeline.addLast(new WebSocketFrameHandler()); // Наш пользовательский обработчик
    
    // Обработчики исходящих сообщений (outbound)
    pipeline.addLast(new WebSocketEncoder()); // Кодирование исходящих WebSocket-сообщений
}

// Определение ключа атрибута
private static final AttributeKey<UserSession> SESSION = AttributeKey.valueOf("session");
 
// Сохранение данных сессии
ctx.channel().attr(SESSION).set(userSession);
 
// Получение данных сессии
UserSession session = ctx.channel().attr(SESSION).get();

public class WebSocketServer {
    private final int port;
    
    public WebSocketServer(int port) {
        this.port = port;
    }
    
    public void run() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new WebSocketServerInitializer());
            
            ChannelFuture future = bootstrap.bind(port).sync();
            System.out.println("Сервер запущен на порту " + port);
            
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    
    public static void main(String[] args) throws Exception {
        int port = 8080;
        new WebSocketServer(port).run();
    }
}

public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {
    private static final String WEBSOCKET_PATH = "/websocket";
    
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        
        // HTTP-кодек для обработки рукопожатия
        pipeline.addLast(new HttpServerCodec());
        
        // Агрегатор HTTP-сообщений, объединяет фрагментированные запросы
        pipeline.addLast(new HttpObjectAggregator(65536));
        
        // Обработчик сжатия для WebSocket (опционально)
        pipeline.addLast(new WebSocketServerCompressionHandler());
        
        // Основной обработчик протокола WebSocket
        pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
        
        // Наш пользовательский обработчик для бизнес-логики
        pipeline.addLast(new WebSocketFrameHandler());
    }
}

public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    private static final Logger logger = LoggerFactory.getLogger(WebSocketFrameHandler.class);
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof TextWebSocketFrame) {
            TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
            String text = textFrame.text();
            
            logger.info("Получено сообщение: {}", text);
            
            // Эхо-ответ, в реальном приложении здесь будет бизнес-логика
            ctx.channel().writeAndFlush(new TextWebSocketFrame("Сервер получил: " + text));
        } else if (frame instanceof BinaryWebSocketFrame) {
            BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame) frame;
            ByteBuf content = binaryFrame.content();
            
            // Обработка бинарных данных
            logger.info("Получены бинарные данные размером {} байт", content.readableBytes());
            
            // Пример обработки бинарных данных
            ByteBuf response = Unpooled.buffer();
            response.writeBytes("Получены бинарные данные".getBytes());
            
            ctx.channel().writeAndFlush(new BinaryWebSocketFrame(response));
        } else if (frame instanceof PingWebSocketFrame) {
            // Отправляем pong в ответ на ping
            ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
        } else if (frame instanceof CloseWebSocketFrame) {
            // Закрываем соединение в ответ на close-фрейм
            logger.info("Получен запрос на закрытие соединения");
            ctx.close();
        }
    }
    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        logger.info("Клиент подключился: {}", ctx.channel().remoteAddress());
    }
    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        logger.info("Клиент отключился: {}", ctx.channel().remoteAddress());
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("Ошибка обработки WebSocket", cause);
        ctx.close();
    }
}

public class Message {
    private String type;
    private String content;
    private String sender;
    // Геттеры и сеттеры опущены для краткости
}
 
public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    private static final ObjectMapper mapper = new ObjectMapper();
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof TextWebSocketFrame) {
            try {
                String text = ((TextWebSocketFrame) frame).text();
                Message message = mapper.readValue(text, Message.class);
                
                // Обработка сообщения в зависимости от его типа
                switch (message.getType()) {
                    case "CHAT":
                        broadcastMessage(message);
                        break;
                    case "JOIN":
                        handleJoin(ctx, message);
                        break;
                    case "LEAVE":
                        handleLeave(ctx, message);
                        break;
                    default:
                        System.out.println("Неизвестный тип сообщения: " + message.getType());
                }
            } catch (Exception e) {
                ctx.channel().writeAndFlush(
                    new TextWebSocketFrame("Ошибка формата: " + e.getMessage())
                );
            }
        }
        // Обработка других типов фреймов...
    }
    
    // Методы broadcastMessage, handleJoin, handleLeave...
}

public class WebSocketServer {
    // Статическая группа каналов для всех подключеных клиентов
    private static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
    // Остальной код...
}
 
public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        // Информируем всех о новом подключении
        WebSocketServer.channels.writeAndFlush(
            new TextWebSocketFrame("Пользователь " + channel.remoteAddress() + " присоединился")
        );
        
        // Добавляем новый канал в группу
        WebSocketServer.channels.add(channel);
    }
    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        // Информируем всех об отключении
        WebSocketServer.channels.writeAndFlush(
            new TextWebSocketFrame("Пользователь " + channel.remoteAddress() + " вышел")
        );
        
        // Канал автоматически удаляется из группы при закрытии
    }
    
    // Метод для отправки сообщения всем клиентам
    private void broadcastMessage(Message message) {
        String json = null;
        try {
            json = mapper.writeValueAsString(message);
        } catch (JsonProcessingException e) {
            System.err.println("Ошибка сериализации сообщения: " + e.getMessage());
            return;
        }
        
        WebSocketServer.channels.writeAndFlush(new TextWebSocketFrame(json));
    }
}

public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
    private static final int HEARTBEAT_INTERVAL = 30; // секунды
    
    private ScheduledFuture<?> heartbeatTask;
    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        scheduleHeartbeat(ctx);
    }
    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        cancelHeartbeat();
    }
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // Перезапускаем таймер при получении любого сообщения
        cancelHeartbeat();
        scheduleHeartbeat(ctx);
        
        // Передаем сообщение следующему обработчику
        ctx.fireChannelRead(msg);
    }
    
    private void scheduleHeartbeat(final ChannelHandlerContext ctx) {
        heartbeatTask = ctx.executor().scheduleAtFixedRate(
            () -> {
                ctx.writeAndFlush(new PingWebSocketFrame());
                System.out.println("Отправлен ping: " + ctx.channel().remoteAddress());
            },
            HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL, TimeUnit.SECONDS
        );
    }
    
    private void cancelHeartbeat() {
        if (heartbeatTask != null) {
            heartbeatTask.cancel(false);
            heartbeatTask = null;
        }
    }
}

pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
pipeline.addLast(new HeartbeatHandler());
pipeline.addLast(new WebSocketFrameHandler());

public class SessionStore {
    // Мапа: идентификатор сессии -> данные сессии
    private static final ConcurrentMap<String, UserSession> sessions = new ConcurrentHashMap<>();
    
    // Мапа: канал -> идентификатор сессии
    private static final ConcurrentMap<Channel, String> channelMap = new ConcurrentHashMap<>();
    
    public static void register(Channel channel, String sessionId, UserSession session) {
        sessions.put(sessionId, session);
        channelMap.put(channel, sessionId);
    }
    
    public static void unregister(Channel channel) {
        String sessionId = channelMap.remove(channel);
        if (sessionId != null) {
            // При отключении не удаляем сессию, а помечаем как неактивную
            UserSession session = sessions.get(sessionId);
            if (session != null) {
                session.setActive(false);
                session.setLastDisconnect(System.currentTimeMillis());
            }
        }
    }
    
    public static UserSession getSession(String sessionId) {
        return sessions.get(sessionId);
    }
    
    // Метод для очистки устаревших сессий
    public static void cleanupSessions() {
        long now = System.currentTimeMillis();
        long maxAge = 24 * 60 * 60 * 1000; // 24 часа
        
        Iterator<Map.Entry<String, UserSession>> it = sessions.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, UserSession> entry = it.next();
            UserSession session = entry.getValue();
            
            if (!session.isActive() && (now - session.getLastDisconnect() > maxAge)) {
                it.remove();
            }
        }
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
    if (frame instanceof TextWebSocketFrame) {
        try {
            String text = ((TextWebSocketFrame) frame).text();
            Message message = mapper.readValue(text, Message.class);
            
            if ("RECONNECT".equals(message.getType())) {
                // Логика восстановления сессии
                String sessionId = message.getContent(); // Предполагаем, что ID передается в content
                UserSession session = SessionStore.getSession(sessionId);
                
                if (session != null) {
                    session.setActive(true);
                    SessionStore.register(ctx.channel(), sessionId, session);
                    
                    // Отправляем клиенту подтверждение и ранее пропущенные сообщения
                    Message response = new Message();
                    response.setType("RECONNECT_OK");
                    response.setContent("Сессия восстановлена");
                    ctx.channel().writeAndFlush(
                        new TextWebSocketFrame(mapper.writeValueAsString(response))
                    );
                    
                    // Отправляем пропущеные сообщения...
                } else {
                    // Сессия не найдена или устарела
                    Message response = new Message();
                    response.setType("RECONNECT_FAIL");
                    response.setContent("Сессия не найдена");
                    ctx.channel().writeAndFlush(
                        new TextWebSocketFrame(mapper.writeValueAsString(response))
                    );
                }
                return;
            }
            
            // Обработка других типов сообщений...
        } catch (Exception e) {
            // Обработка ошибок...
        }
    }
}

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(
    SessionStore::cleanupSessions,
    1, 1, TimeUnit.HOURS
);

// Плохой подход: отправка избыточных данных
JsonObject fullUserObject = new JsonObject();
fullUserObject.addProperty("id", user.getId());
fullUserObject.addProperty("name", user.getName());
fullUserObject.addProperty("email", user.getEmail());
fullUserObject.addProperty("phone", user.getPhone());
fullUserObject.addProperty("address", user.getAddress());
// ... и ещё 20 полей, которые не нужны в данном контексте
ctx.channel().writeAndFlush(new TextWebSocketFrame(fullUserObject.toString()));
 
// Хороший подход: отправка только необходимых данных
JsonObject minimalUserObject = new JsonObject();
minimalUserObject.addProperty("id", user.getId());
minimalUserObject.addProperty("name", user.getName());
ctx.channel().writeAndFlush(new TextWebSocketFrame(minimalUserObject.toString()));

ObjectMapper mapper = new ObjectMapper();
// Отключаем "pretty printing"
String compactJson = mapper.writeValueAsString(message);

// Класс для буферизации исходящих сообщений
public class MessageAggregator {
    private final Queue<Message> messageQueue = new ConcurrentLinkedQueue<>();
    private final Channel channel;
    private final int maxBatchSize;
    private final long maxDelayMs;
    private ScheduledFuture<?> scheduledFlush;
    
    public MessageAggregator(Channel channel, int maxBatchSize, long maxDelayMs) {
        this.channel = channel;
        this.maxBatchSize = maxBatchSize;
        this.maxDelayMs = maxDelayMs;
    }
    
    public void addMessage(Message message) {
        messageQueue.add(message);
        
        if (messageQueue.size() >= maxBatchSize) {
            flush();
        } else if (scheduledFlush == null) {
            // Планируем отправку с задержкой, если очередь не пуста, но меньше порога
            scheduledFlush = channel.eventLoop().schedule(
                this::flush,
                maxDelayMs, 
                TimeUnit.MILLISECONDS
            );
        }
    }
    
    private void flush() {
        if (scheduledFlush != null) {
            scheduledFlush.cancel(false);
            scheduledFlush = null;
        }
        
        if (messageQueue.isEmpty()) {
            return;
        }
        
        List<Message> batch = new ArrayList<>(maxBatchSize);
        Message msg;
        while ((msg = messageQueue.poll()) != null && batch.size() < maxBatchSize) {
            batch.add(msg);
        }
        
        if (!batch.isEmpty()) {
            // Создаём объект-контейнер для пакета сообщений
            MessageBatch messageBatch = new MessageBatch(batch);
            ObjectMapper mapper = new ObjectMapper();
            try {
                String json = mapper.writeValueAsString(messageBatch);
                channel.writeAndFlush(new TextWebSocketFrame(json));
            } catch (JsonProcessingException e) {
                // Обработка ошибок
            }
        }
    }
}

// Использование Protocol Buffers вместо JSON
// Предполагается, что у нас есть сгенерированный класс UserProto
UserProto.User userProto = UserProto.User.newBuilder()
    .setId(user.getId())
    .setName(user.getName())
    .build();
 
ByteBuf binaryData = Unpooled.wrappedBuffer(userProto.toByteArray());
ctx.channel().writeAndFlush(new BinaryWebSocketFrame(binaryData));

// Определяем количество потоков на основе доступных ядер
int bossThreads = 1; // Для приёма соединений достаточно одного потока
int workerThreads = Runtime.getRuntime().availableProcessors() * 2;
 
EventLoopGroup bossGroup = new NioEventLoopGroup(bossThreads);
EventLoopGroup workerGroup = new NioEventLoopGroup(workerThreads);

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childOption(ChannelOption.TCP_NODELAY, true) // Отключаем алгоритм Нагла
        .childOption(ChannelOption.SO_KEEPALIVE, true) // Включаем TCP keepalive
        .childOption(ChannelOption.SO_REUSEADDR, true) // Разрешаем переиспользование адресов
        .childOption(ChannelOption.SO_RCVBUF, 128 * 1024) // Размер буфера приёма
        .childOption(ChannelOption.SO_SNDBUF, 128 * 1024) // Размер буфера отправки
        .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(
            8 * 1024, // Низкая отметка
            32 * 1024  // Высокая отметка
        ));

// Настройка пулирования буферов
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
        .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

public void sendMessage(ChannelHandlerContext ctx, WebSocketFrame frame) {
    Channel channel = ctx.channel();
    
    if (channel.isWritable()) {
        channel.writeAndFlush(frame);
    } else {
        // Канал перегружен, нужна альтернативная стратегия
        // Например, сбросить низкоприоритетные сообщения или поставить в очередь
        if (frame.isFlagSet(Priority.HIGH)) {
            // Высокоприоритетные сообщения отправляем в любом случае
            channel.writeAndFlush(frame);
        } else {
            // Низкоприоритетные можно отбросить или отложить
            messageQueue.offer(frame);
        }
    }
}

pipeline.addLast(new HttpServerCodec());
pipeline.addLast(new HttpObjectAggregator(65536));
pipeline.addLast(new WebSocketServerCompressionHandler()); // Включаем сжатие
pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH));

// Более тонкая настройка сжатия
ZlibCodecFactory zlibCodecFactory = new ZlibCodecFactory();
final boolean server = true;
final ZlibWrapper wrapper = ZlibWrapper.GZIP;
final int compressionLevel = 6; // Уровень от 0 (без сжатия) до 9 (максимальное сжатие)
final int windowBits = 15;
final int memLevel = 8;
 
pipeline.addLast("gzipDecoder", zlibCodecFactory.newZlibDecoder(wrapper));
pipeline.addLast("gzipEncoder", zlibCodecFactory.newZlibEncoder(wrapper, compressionLevel, windowBits, memLevel));

// Создаём отдельный пул для бизнес-логики
EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(16);
 
// Добавляем обработчик бизнес-логики в отдельный пул
pipeline.addLast(businessGroup, "businessHandler", new WebSocketBusinessHandler());

// Создаём несколько групп для разных типов операций
EventExecutorGroup databaseGroup = new DefaultEventExecutorGroup(4);
EventExecutorGroup computationGroup = new DefaultEventExecutorGroup(8);
EventExecutorGroup notificationGroup = new DefaultEventExecutorGroup(2);

public class ChatServer {
    private static final int PORT = 8080;
    private static final Map<String, Channel> users = new ConcurrentHashMap<>();
    
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChatServerInitializer());
            
            Channel channel = bootstrap.bind(PORT).sync().channel();
            System.out.println("Чат-сервер запущен на порту " + PORT);
            
            channel.closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    
    public static void registerUser(String username, Channel channel) {
        users.put(username, channel);
        broadcastMessage("SERVER", username + " присоединился к чату");
    }
    
    public static void removeUser(String username) {
        users.remove(username);
        broadcastMessage("SERVER", username + " покинул чат");
    }
    
    public static void broadcastMessage(String sender, String message) {
        TextWebSocketFrame frame = new TextWebSocketFrame(
            String.format("{\"type\":\"broadcast\",\"sender\":\"%s\",\"message\":\"%s\"}", 
                          sender, message));
        
        users.values().forEach(ch -> ch.writeAndFlush(frame.retainedDuplicate()));
        frame.release();
    }
    
    public static void sendPrivateMessage(String sender, String recipient, String message) {
        Channel channel = users.get(recipient);
        if (channel != null && channel.isActive()) {
            TextWebSocketFrame frame = new TextWebSocketFrame(
                String.format("{\"type\":\"private\",\"sender\":\"%s\",\"message\":\"%s\"}", 
                              sender, message));
            channel.writeAndFlush(frame);
        }
    }
}

public class ChatServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();
        
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(65536));
        pipeline.addLast(new WebSocketServerProtocolHandler("/chat"));
        pipeline.addLast(new ChatFrameHandler());
    }
}

public class ChatFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    private static final ObjectMapper MAPPER = new ObjectMapper();
    private String username;
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
        if (frame instanceof TextWebSocketFrame) {
            String text = ((TextWebSocketFrame) frame).text();
            JsonNode json = MAPPER.readTree(text);
            
            String type = json.get("type").asText();
            switch (type) {
                case "login":
                    username = json.get("username").asText();
                    ChatServer.registerUser(username, ctx.channel());
                    break;
                case "broadcast":
                    String message = json.get("message").asText();
                    ChatServer.broadcastMessage(username, message);
                    break;
                case "private":
                    String recipient = json.get("recipient").asText();
                    String privateMessage = json.get("message").asText();
                    ChatServer.sendPrivateMessage(username, recipient, privateMessage);
                    break;
            }
        }
    }
    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        if (username != null) {
            ChatServer.removeUser(username);
        }
    }
}

public class ChatClient {
    private final URI serverUri;
    private Channel channel;
    private final EventLoopGroup group = new NioEventLoopGroup();
    private final String username;
    
    public ChatClient(String host, int port, String username) {
        try {
            this.serverUri = new URI("ws://" + host + ":" + port + "/chat");
            this.username = username;
        } catch (URISyntaxException e) {
            throw new RuntimeException(e);
        }
    }
    
    public void connect() throws Exception {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new HttpClientCodec());
                        pipeline.addLast(new HttpObjectAggregator(65536));
                        pipeline.addLast(new WebSocketClientProtocolHandler(
                                WebSocketClientHandshakerFactory.newHandshaker(
                                        serverUri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
                        pipeline.addLast(new ChatClientHandler());
                    }
                });
        
        this.channel = bootstrap.connect(serverUri.getHost(), serverUri.getPort()).sync().channel();
        
        // Ждем успешного завершения рукопожатия
        ChatClientHandler handler = (ChatClientHandler) channel.pipeline().last();
        handler.handshakeFuture().sync();
        
        // Отправляем сообщение о логине
        login();
    }
    
    public void login() throws Exception {
        String loginJson = String.format("{\"type\":\"login\",\"username\":\"%s\"}", username);
        channel.writeAndFlush(new TextWebSocketFrame(loginJson));
    }
    
    public void sendMessage(String message) {
        if (channel != null && channel.isActive()) {
            String messageJson = String.format("{\"type\":\"broadcast\",\"message\":\"%s\"}", message);
            channel.writeAndFlush(new TextWebSocketFrame(messageJson));
        }
    }
    
    public void sendPrivateMessage(String recipient, String message) {
        if (channel != null && channel.isActive()) {
            String messageJson = String.format(
                "{\"type\":\"private\",\"recipient\":\"%s\",\"message\":\"%s\"}", 
                recipient, message);
            channel.writeAndFlush(new TextWebSocketFrame(messageJson));
        }
    }
    
    public void disconnect() {
        if (channel != null) {
            channel.close();
        }
        group.shutdownGracefully();
    }
    
    public static void main(String[] args) throws Exception {
        String host = "localhost";
        int port = 8080;
        String username = "user" + new Random().nextInt(1000);
        
        final ChatClient client = new ChatClient(host, port, username);
        client.connect();
        
        // Демонстрационая отправка сообщений
        client.sendMessage("Привет всем!");
        Thread.sleep(1000);
        client.sendPrivateMessage("admin", "Привет, админ!");
        
        // Добавляем хук для корректного закрытия при выходе
        Runtime.getRuntime().addShutdownHook(new Thread(client::disconnect));
        
        // Консольный ввод для интерактивного тестирования
        BufferedReader console = new BufferedReader(new InputStreamReader(System.in));
        while (true) {
            String line = console.readLine();
            if ("/quit".equals(line.toLowerCase())) {
                break;
            }
            
            if (line.startsWith("/pm ")) {
                String[] parts = line.split(" ", 3);
                if (parts.length == 3) {
                    client.sendPrivateMessage(parts[1], parts[2]);
                }
            } else {
                client.sendMessage(line);
            }
        }
        
        client.disconnect();
    }
}

public class ChatClientHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    private final ObjectMapper MAPPER = new ObjectMapper();
    private ChannelPromise handshakeFuture;
    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        handshakeFuture = ctx.newPromise();
    }
    
    public ChannelFuture handshakeFuture() {
        return handshakeFuture;
    }
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
        if (frame instanceof TextWebSocketFrame) {
            String text = ((TextWebSocketFrame) frame).text();
            JsonNode json = MAPPER.readTree(text);
            
            String type = json.get("type").asText();
            String sender = json.has("sender") ? json.get("sender").asText() : "server";
            String message = json.get("message").asText();
            
            if ("broadcast".equals(type)) {
                System.out.printf("[%s]: %s%n", sender, message);
            } else if ("private".equals(type)) {
                System.out.printf("[PM от %s]: %s%n", sender, message);
            }
        }
    }
    
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) {
            handshakeFuture.setSuccess();
            System.out.println("WebSocket подключение установлено!");
        }
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        if (!handshakeFuture.isDone()) {
            handshakeFuture.setFailure(cause);
        }
        ctx.close();
    }
}

public class RedisChatService {
    private final JedisPool jedisPool;
    private final Executor redisExecutor;
    private final Map<String, Channel> localUsers = new ConcurrentHashMap<>();
    private Thread subscriberThread;
    private volatile boolean isRunning = true;
    
    public RedisChatService(String redisHost, int redisPort) {
        this.jedisPool = new JedisPool(redisHost, redisPort);
        this.redisExecutor = Executors.newSingleThreadExecutor();
        
        // Запуск подписчика Redis
        startSubscriber();
    }
    
    private void startSubscriber() {
        subscriberThread = new Thread(() -> {
            try (Jedis jedis = new Jedis(jedisPool.getResource().getClient().getHost(),
                                        jedisPool.getResource().getClient().getPort())) {
                jedis.subscribe(new JedisPubSub() {
                    @Override
                    public void onMessage(String channel, String message) {
                        try {
                            if ("chat:broadcast".equals(channel)) {
                                handleBroadcastMessage(message);
                            } else if (channel.startsWith("chat:user:")) {
                                String username = channel.substring("chat:user:".length());
                                handlePrivateMessage(username, message);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }, "chat:broadcast", "chat:user:*");
            }
        });
        subscriberThread.setDaemon(true);
        subscriberThread.start();
    }
    
    public void registerUser(String username, Channel channel) {
        localUsers.put(username, channel);
    }
    
    public void removeUser(String username) {
        localUsers.remove(username);
    }
    
    public void broadcastMessage(String sender, String message) {
        redisExecutor.execute(() -> {
            try (Jedis jedis = jedisPool.getResource()) {
                String json = String.format(
                    "{\"type\":\"broadcast\",\"sender\":\"%s\",\"message\":\"%s\"}",
                    sender, message);
                jedis.publish("chat:broadcast", json);
            }
        });
    }
    
    public void sendPrivateMessage(String sender, String recipient, String message) {
        redisExecutor.execute(() -> {
            try (Jedis jedis = jedisPool.getResource()) {
                String json = String.format(
                    "{\"type\":\"private\",\"sender\":\"%s\",\"message\":\"%s\"}",
                    sender, message);
                jedis.publish("chat:user:" + recipient, json);
            }
        });
    }
    
    private void handleBroadcastMessage(String json) {
        TextWebSocketFrame frame = new TextWebSocketFrame(json);
        
        // Отправляем всем локальным пользователям
        for (Channel channel : localUsers.values()) {
            if (channel.isActive()) {
                channel.writeAndFlush(frame.retainedDuplicate());
            }
        }
        frame.release();
    }
    
    private void handlePrivateMessage(String recipient, String json) {
        Channel channel = localUsers.get(recipient);
        if (channel != null && channel.isActive()) {
            channel.writeAndFlush(new TextWebSocketFrame(json));
        }
    }
    
    public void shutdown() {
        isRunning = false;
        if (subscriberThread != null) {
            subscriberThread.interrupt();
        }
        jedisPool.close();
    }
}

public class ChatServer {
    private static final int PORT = 8080;
    private static final RedisChatService chatService = 
        new RedisChatService("localhost", 6379);
    
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChatServerInitializer(chatService));
            
            Channel channel = bootstrap.bind(PORT).sync().channel();
            System.out.println("Чат-сервер запущен на порту " + PORT);
            
            channel.closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            chatService.shutdown();
        }
    }
}

public class ChatFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {
    private static final ObjectMapper MAPPER = new ObjectMapper();
    private String username;
    private final RedisChatService chatService;
    
    public ChatFrameHandler(RedisChatService chatService) {
        this.chatService = chatService;
    }
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
        if (frame instanceof TextWebSocketFrame) {
            String text = ((TextWebSocketFrame) frame).text();
            JsonNode json = MAPPER.readTree(text);
            
            String type = json.get("type").asText();
            switch (type) {
                case "login":
                    username = json.get("username").asText();
                    chatService.registerUser(username, ctx.channel());
                    chatService.broadcastMessage("SERVER", username + " присоединился к чату");
                    break;
                case "broadcast":
                    String message = json.get("message").asText();
                    chatService.broadcastMessage(username, message);
                    break;
                case "private":
                    String recipient = json.get("recipient").asText();
                    String privateMessage = json.get("message").asText();
                    chatService.sendPrivateMessage(username, recipient, privateMessage);
                    break;
            }
        }
    }
    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        if (username != null) {
            chatService.removeUser(username);
            chatService.broadcastMessage("SERVER", username + " покинул чат");
        }
    }
}

public class ChatServerInitializer extends ChannelInitializer<SocketChannel> {
    private final RedisChatService chatService;
    
    public ChatServerInitializer(RedisChatService chatService) {
        this.chatService = chatService;
    }
    
    @Override
    protected void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();
        
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpObjectAggregator(65536));
        pipeline.addLast(new WebSocketServerProtocolHandler("/chat"));
        pipeline.addLast(new ChatFrameHandler(chatService));
    }
}

public class KafkaWebSocketService {
    private final KafkaConsumer<String, String> consumer;
    private final KafkaProducer<String, String> producer;
    private final Map<String, Channel> connectedUsers = new ConcurrentHashMap<>();
    private final ExecutorService executorService;
    private volatile boolean isRunning = true;
    
    public KafkaWebSocketService(String bootstrapServers) {
        // Настройка Kafka Consumer
        Properties consumerProps = new Properties();
        consumerProps.put("bootstrap.servers", bootstrapServers);
        consumerProps.put("group.id", "websocket-service");
        consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        this.consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Arrays.asList("user-notifications", "broadcasts"));
        
        // Настройка Kafka Producer
        Properties producerProps = new Properties();
        producerProps.put("bootstrap.servers", bootstrapServers);
        producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        this.producer = new KafkaProducer<>(producerProps);
        
        // Запуск потока для обработки сообщений
        this.executorService = Executors.newSingleThreadExecutor();
        this.executorService.submit(this::consumeMessages);
    }
    
    private void consumeMessages() {
        try {
            while (isRunning) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String topic = record.topic();
                    String key = record.key();
                    String value = record.value();
                    
                    if ("broadcasts".equals(topic)) {
                        // Отправляем всем
                        TextWebSocketFrame frame = new TextWebSocketFrame(value);
                        for (Channel channel : connectedUsers.values()) {
                            if (channel.isActive()) {
                                channel.writeAndFlush(frame.retainedDuplicate());
                            }
                        }
                        frame.release();
                    } else if ("user-notifications".equals(topic) && key != null) {
                        // Отправляем конкретному пользователю
                        Channel channel = connectedUsers.get(key);
                        if (channel != null && channel.isActive()) {
                            channel.writeAndFlush(new TextWebSocketFrame(value));
                        }
                    }
                }
            }
        } catch (Exception e) {
            if (isRunning) {  // Игнорируем исключения при остановке
                e.printStackTrace();
            }
        }
    }
    
    public void registerUser(String username, Channel channel) {
        connectedUsers.put(username, channel);
        // Отправляем событие о подключении пользователя в Kafka
        producer.send(new ProducerRecord<>("user-events", username, 
                "{\"type\":\"connected\",\"username\":\"" + username + "\"}"));
    }
    
    public void removeUser(String username) {
        connectedUsers.remove(username);
        // Отправляем событие об отключении пользователя в Kafka
        producer.send(new ProducerRecord<>("user-events", username, 
                "{\"type\":\"disconnected\",\"username\":\"" + username + "\"}"));
    }
    
    public void sendUserMessage(String username, String message) {
        // Отправляем сообщение пользователя в Kafka для обработки
        producer.send(new ProducerRecord<>("chat-messages", username, message));
    }
    
    public void shutdown() {
        isRunning = false;
        consumer.wakeup();
        executorService.shutdownNow();
        consumer.close();
        producer.close();
    }
}