From 164a3286e49a2574f712c26ddc948c1025e1b4c2 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Sat, 1 Aug 2026 04:24:50 +0000 Subject: [PATCH] =?UTF-8?q?=F0=9F=93=9D=20Add=20docstrings=20to=20`binary-?= =?UTF-8?q?eio-v3`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Docstrings generation was requested by @sanjomo. * https://github.com/socketio4j/netty-socketio/pull/228#issuecomment-5134504333 The following files were modified: * `netty-socketio-core/src/main/java/com/socketio4j/socketio/SingleRoomBroadcastOperations.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/SocketIOServer.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/EncoderHandler.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/InPacketHandler.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/ClientListeners.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/namespace/Namespace.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/JacksonJsonSupport.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Packet.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketDecoder.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketEncoder.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/EventMessageJsonSupport.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageDeserializer.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageSerializer.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/EventMessageCodec.java` * `netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java` * `netty-socketio-core/src/test/java/com/socketio4j/socketio/store/CustomizedHazelcastContainer.java` * `netty-socketio-examples/netty-socketio-core-example/src/main/java/com/socketio4j/example/core/CoreExampleMain.java` --- .../SingleRoomBroadcastOperations.java | 12 ++ .../socketio4j/socketio/SocketIOServer.java | 10 + .../socketio/handler/EncoderHandler.java | 20 ++ .../socketio/handler/InPacketHandler.java | 6 + .../socketio/listener/ClientListeners.java | 28 ++- .../socketio/namespace/Namespace.java | 21 +++ .../socketio/protocol/JacksonJsonSupport.java | 6 + .../socketio4j/socketio/protocol/Packet.java | 29 ++- .../socketio/protocol/PacketDecoder.java | 175 ++++++++---------- .../socketio/protocol/PacketEncoder.java | 38 ++++ .../store/event/EventMessageJsonSupport.java | 17 ++ .../hazelcast/HazelcastPubSubEventStore.java | 7 + .../EventMessageDeserializer.java | 7 + .../serialization/EventMessageSerializer.java | 8 + .../store/nats_pubsub/EventMessageCodec.java | 3 + .../redis_stream/RedisStreamEventStore.java | 15 ++ .../store/CustomizedHazelcastContainer.java | 6 + .../example/core/CoreExampleMain.java | 48 ++++- 18 files changed, 339 insertions(+), 117 deletions(-) diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/SingleRoomBroadcastOperations.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/SingleRoomBroadcastOperations.java index e29208ad..d42df9ef 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/SingleRoomBroadcastOperations.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/SingleRoomBroadcastOperations.java @@ -58,6 +58,11 @@ public Collection getClients() { return new IterableCollection<>(clients); } + /** + * Sends a packet to every client in the room and dispatches it to the event store. + * + * @param packet the packet to send + */ @Override public void send(Packet packet) { for (SocketIOClient client : clients) { @@ -89,6 +94,13 @@ public void sendEvent(String name, SocketIOClient excludedClient, Object... data sendEvent(name, excludePredicate, data); } + /** + * Sends a named event with the specified data to clients that do not match the exclusion predicate. + * + * @param name the event name + * @param excludePredicate the predicate identifying clients to exclude + * @param data the event data + */ @Override public void sendEvent(String name, Predicate excludePredicate, Object... data) { Packet packet = new Packet(PacketType.MESSAGE, EngineIOVersion.UNKNOWN); diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/SocketIOServer.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/SocketIOServer.java index fa786813..03d3dda4 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/SocketIOServer.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/SocketIOServer.java @@ -1068,11 +1068,21 @@ public void addConnectListener(ConnectListener listener) { mainNamespace.addConnectListener(listener); } + /** + * Removes a listener invoked when a client connects. + * + * @param listener the connection listener to remove + */ @Override public void removeConnectListener(ConnectListener listener) { mainNamespace.removeConnectListener(listener); } + /** + * Removes a disconnect listener from the main namespace. + * + * @param listener the disconnect listener to remove + */ @Override public void removeDisconnectListener(DisconnectListener listener) { mainNamespace.removeDisconnectListener(listener); diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/EncoderHandler.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/EncoderHandler.java index 3cd3a1ce..77e7a5cf 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/EncoderHandler.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/EncoderHandler.java @@ -266,6 +266,18 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) private static final int FRAME_BUFFER_SIZE = 8192; + /** + * Encodes and sends queued packets and attachments as WebSocket frames. + * + *

Large packet payloads are fragmented according to the configured maximum frame + * payload length. Attachment frames include the Engine.IO v2 or v3 prefix when + * required, and the supplied promise is completed after queued writes finish.

+ * + * @param msg the outbound packet message containing the client packet queue + * @param ctx the channel handler context used for encoding and writing frames + * @param promise the promise completed when processing finishes + * @throws IOException if packet encoding fails + */ private void handleWebsocket(final OutPacketMessage msg, ChannelHandlerContext ctx, ChannelPromise promise) throws IOException { if (log.isDebugEnabled()) { log.debug("Starting WebSocket message processing, sessionId: {}", msg.getSessionId()); @@ -352,6 +364,14 @@ private void handleWebsocket(final OutPacketMessage msg, ChannelHandlerContext c } } + /** + * Encodes and sends queued packets as an HTTP polling response. + * + *

Processes one response per channel and selects the response encoding and content type + * according to the Engine.IO version and queued packet content.

+ * + * @throws IOException if packet encoding fails + */ private void handleHTTP(OutPacketMessage msg, ChannelHandlerContext ctx, ChannelPromise promise) throws IOException { if (log.isDebugEnabled()) { log.debug("Starting HTTP polling message processing, sessionId: {}", msg.getSessionId()); diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/InPacketHandler.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/InPacketHandler.java index cf2a0a62..90b6dd96 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/InPacketHandler.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/InPacketHandler.java @@ -58,6 +58,12 @@ public InPacketHandler(PacketListener packetListener, PacketDecoder decoder, Nam this.exceptionListener = exceptionListener; } + /** + * Processes inbound packets for a client and dispatches them to the appropriate namespace. + * + * @param message the inbound packet message containing the payload, client, and transport + * @throws Exception if packet decoding or processing fails + */ @Override protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsMessage message) throws Exception { diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/ClientListeners.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/ClientListeners.java index db7856f9..a962a894 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/ClientListeners.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/ClientListeners.java @@ -25,13 +25,33 @@ public interface ClientListeners { void addEventInterceptor(EventInterceptor eventInterceptor); - void addDisconnectListener(DisconnectListener listener); + /** + * Registers a listener to be notified when the client disconnects. + * + * @param listener the disconnect listener to register + */ +void addDisconnectListener(DisconnectListener listener); - void removeDisconnectListener(DisconnectListener listener); + /** + * Removes a disconnect listener. + * + * @param listener the disconnect listener to remove + */ +void removeDisconnectListener(DisconnectListener listener); - void addConnectListener(ConnectListener listener); + /** + * Registers a listener to be notified when a client connects. + * + * @param listener the listener to register + */ +void addConnectListener(ConnectListener listener); - void removeConnectListener(ConnectListener listener); + /** + * Removes a listener that is notified when a client connects. + * + * @param listener the connect listener to remove + */ +void removeConnectListener(ConnectListener listener); /** * from v4, ping will always be sent by server except probe ping packet sent from client, diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/namespace/Namespace.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/namespace/Namespace.java index 42903840..2b469884 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/namespace/Namespace.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/namespace/Namespace.java @@ -289,16 +289,31 @@ public void addConnectListener(ConnectListener listener) { connectListeners.add(listener); } + /** + * Removes a listener invoked when a client connects to the namespace. + * + * @param listener the connection listener to remove + */ @Override public void removeConnectListener(ConnectListener listener) { connectListeners.remove(listener); } + /** + * Removes a disconnect listener from this namespace. + * + * @param listener the disconnect listener to remove + */ @Override public void removeDisconnectListener(DisconnectListener listener) { disconnectListeners.remove(listener); } + /** + * Registers a client with the namespace and notifies the connection listeners. + * + * @param client the client connecting to the namespace + */ public void onConnect(SocketIOClient client) { if (roomClients.containsKey(getName()) && roomClients.get(getName()).contains(client.getSessionId())) { @@ -430,6 +445,12 @@ public void joinRooms(Set rooms, final UUID sessionId) { storeFactory.eventStore().publish(EventType.BULK_JOIN, new BulkJoinMessage(sessionId, rooms, getName())); } + /** + * Sends a packet to every locally connected client in the specified room. + * + * @param room the target room + * @param packet the packet to send + */ public void dispatch(String room, Packet packet) { int size = forEachRoomClient(room, client -> { // Produce a per-client copy so that the shared Packet is never mutated. diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/JacksonJsonSupport.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/JacksonJsonSupport.java index f80d4165..65a94ef1 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/JacksonJsonSupport.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/JacksonJsonSupport.java @@ -76,6 +76,12 @@ protected AckArgsDeserializer() { super(AckArgs.class); } + /** + * Deserializes acknowledgment arguments according to the current callback's expected types. + * + * @param jp the JSON parser containing the acknowledgment arguments + * @return the deserialized acknowledgment arguments + */ @Override public AckArgs deserialize(JsonParser jp, DeserializationContext ctxt) throws IOException { List args = new ArrayList(); diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Packet.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Packet.java index 26a21dec..1347f92d 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Packet.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Packet.java @@ -86,13 +86,13 @@ public T getData() { } /** - * Creates a copy of #{@link Packet} with new namespace set - * if it differs from current namespace. - * Otherwise, returns original object unchanged + * Creates a packet with the specified namespace and Engine.IO version when the + * namespace differs from the current namespace. * - * @param namespace - * @param engineIOVersion - * @return packet + * @param namespace the namespace to set + * @param engineIOVersion the Engine.IO version to set on a copied packet + * @return the original packet when the namespace matches case-insensitively; + * otherwise, a shallow copy with the specified namespace and version */ public Packet withNsp(String namespace, EngineIOVersion engineIOVersion) { if (this.nsp.equalsIgnoreCase(namespace)) { @@ -112,16 +112,10 @@ public Packet withNsp(String namespace, EngineIOVersion engineIOVersion) { } /** - * Returns a packet with the given {@link EngineIOVersion} stamped in. - *

- * If {@code engineIOVersion} is already equal to this packet's version, {@code this} is - * returned unchanged — no allocation. Otherwise a shallow copy is created so that the - * shared original is never mutated. This matters during room broadcasts: {@code ClientHead.send} - * only enqueues the packet; {@code EncoderHandler} reads the version later on a Netty - * event-loop thread, so every client must hold its own stable version reference. + * Creates a packet associated with the specified Engine.IO version. * - * @param engineIOVersion the EIO version to stamp onto the packet - * @return {@code this} if the version already matches, otherwise a new {@link Packet} + * @param engineIOVersion the Engine.IO version to associate with the packet + * @return this packet if the version matches; otherwise, a shallow copy with the specified version */ public Packet withEngineIOVersion(EngineIOVersion engineIOVersion) { if (engineIOVersion == this.engineIOVersion) { @@ -139,6 +133,11 @@ public Packet withEngineIOVersion(EngineIOVersion engineIOVersion) { return copy; } + /** + * Sets the packet namespace, converting the empty namespace object representation to an empty string. + * + * @param endpoint the namespace endpoint + */ public void setNsp(String endpoint) { //patch for #903 if ("{}".equals(endpoint)){ diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketDecoder.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketDecoder.java index 266109e3..32a109b1 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketDecoder.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketDecoder.java @@ -218,6 +218,12 @@ private PacketType readInnerType(ByteBuf buffer) { return PacketType.valueOfInner(typeId); } + /** + * Detects whether the buffer begins with a numeric length header. + * + * @param buffer the buffer to inspect + * @return {@code true} if a colon follows one or more decimal digits within the first ten readable bytes, {@code false} otherwise + */ private boolean hasLengthHeader(ByteBuf buffer) { for (int i = 0; i < Math.min(buffer.readableBytes(), 10); i++) { byte b = buffer.getByte(buffer.readerIndex() + i); @@ -235,6 +241,15 @@ public Packet decodePackets(ByteBuf buffer, ClientHead client) throws IOExceptio return decodePackets(buffer, client, client.getCurrentTransport()); } + /** + * Decodes a packet from the supplied buffer using the framing format detected in its contents. + * + * @param buffer the buffer containing the encoded packet + * @param client the client associated with the packet + * @param transport the transport used to receive the packet + * @return the decoded packet, or {@code null} when the buffer contains no packet + * @throws IOException if the packet cannot be decoded + */ public Packet decodePackets(ByteBuf buffer, ClientHead client, Transport transport) throws IOException { if (isStringPacket(buffer)) { return decodeWithStringHeader(buffer, client, transport); @@ -245,8 +260,13 @@ public Packet decodePackets(ByteBuf buffer, ClientHead client, Transport transpo } /** - * Decode packet with string header format - * Handles packets that start with 0x0 byte + * Decodes a packet using a delimiter-terminated numeric length header. + * + * @param buffer the buffer containing the framed packet + * @param client the client associated with the packet + * @param transport the transport used to receive the packet + * @return the decoded packet + * @throws IOException if the packet frame cannot be decoded */ private Packet decodeWithStringHeader(ByteBuf buffer, ClientHead client, Transport transport) throws IOException { int maxLength = Math.min(buffer.readableBytes(), 10); @@ -259,8 +279,13 @@ private Packet decodeWithStringHeader(ByteBuf buffer, ClientHead client, Transpo } /** - * Decode packet with length header format - * Handles packets with format "length:data" + * Decodes a packet framed with a {@code length:data} header. + * + * @param buffer the buffer containing the framed packet + * @param client the client associated with the packet + * @param transport the transport used to receive the packet + * @return the decoded packet + * @throws IOException if the packet cannot be decoded */ private Packet decodeWithLengthHeader(ByteBuf buffer, ClientHead client, Transport transport) throws IOException { int lengthEndIndex = buffer.bytesBefore((byte) ':'); @@ -270,8 +295,14 @@ private Packet decodeWithLengthHeader(ByteBuf buffer, ClientHead client, Transpo } /** - * Common frame decoding logic - * Extracts frame data and advances buffer position + * Decodes a length-delimited frame from the buffer. + * + * @param buffer the buffer containing the frame + * @param client the client associated with the packet + * @param len the frame length in bytes + * @param transport the transport used to receive the frame + * @return the decoded packet + * @throws IOException if the frame cannot be decoded */ private Packet decodeFrame(ByteBuf buffer, ClientHead client, int len, Transport transport) throws IOException { ByteBuf frame = buffer.slice(buffer.readerIndex() + 1, len); @@ -283,12 +314,28 @@ private String readString(ByteBuf frame) { return readString(frame, frame.readableBytes()); } + /** + * Reads a UTF-8 string of the specified byte length from the frame. + * + * @param frame the buffer containing the string data + * @param size the number of bytes to read + * @return the decoded UTF-8 string + */ private String readString(ByteBuf frame, int size) { byte[] bytes = new byte[size]; frame.readBytes(bytes); return new String(bytes, CharsetUtil.UTF_8); } + /** + * Decodes the next packet or binary attachment from a framed buffer. + * + * @param head the client whose packet and attachment state are updated + * @param frame the buffer containing the next encoded packet + * @param transport the transport used to interpret packet data + * @return the decoded packet, or {@code null} when the frame contains no packet + * @throws IOException if packet or attachment data cannot be decoded + */ private Packet decode(ClientHead head, ByteBuf frame, Transport transport) throws IOException { Packet lastPacket = head.getLastBinaryPacket(); @@ -342,6 +389,13 @@ private Packet decode(ClientHead head, ByteBuf frame, Transport transport) throw return packet; } + /** + * Parses packet attachments, namespace, and acknowledgement metadata from the frame header. + * + * @param frame the buffer containing the packet header + * @param packet the packet to populate + * @param innerType the packet subtype used to determine whether attachments are supported + */ private void parseHeader(ByteBuf frame, Packet packet, PacketType innerType) { int endIndex = frame.bytesBefore((byte) '['); if (endIndex <= 0) { @@ -380,94 +434,17 @@ private void parseHeader(ByteBuf frame, Packet packet, PacketType innerType) { } /** - * Decodes and appends an incoming binary attachment to the given packet. - *

- * Depending on the negotiated Engine.IO version and transport, the incoming buffer - * has different frame layouts: - *

- * - *

Engine.IO v3 (Socket.IO 2.x and older)

- *
    - *
  • - * WebSocket (Raw Binary Frame): - * 
    -     *     +---------------+---------------------------------+
-     *     | Byte 0        | Bytes 1..N                      |
-     *     +---------------+---------------------------------+
-     *     | Type (0x04)   | Raw binary payload              |
-     *     +---------------+---------------------------------+
-     *
    - * The leading byte value 4 (Engine.IO MESSAGE packet type) is stripped, and the - * remainder is base64-encoded and appended as an attachment. - *
    • - *
  • - * WebSocket/Polling (Base64 Text Frame): - * 
    -     *     +-----------------+-------------------------------+
-     *     | Bytes 0..1      | Bytes 2..N                    |
-     *     +-----------------+-------------------------------+
-     *     | Prefix ("b4")   | Base64 string payload         |
-     *     +-----------------+-------------------------------+
-     *
    - * The leading ASCII prefix "b4" is stripped, and the remaining base64 payload is - * appended directly without double-encoding. - *
    - * Ref: Engine.IO v3 Packet String Encoding Spec - *
    - * "Sometimes, it is not possible to send binary data over the transport [...]. In that case, - * the packet is encoded as a string, and prepended with a 'b' character. For example: a packet - * of type message containing the buffer <01 02 03> is encoded as 'b4AQID'" - *
    - *
    • - *
  • - * Polling (Raw Binary Wrapper): - * 
    -     *     +--------+---------------+--------+---------------+--------------------+
-     *     | Byte 0 | Bytes 1..K    | Byte K | Byte K+1      | Bytes K+2..N       |
-     *     +--------+---------------+--------+---------------+--------------------+
-     *     | 0x01   | Length (ASCII) | 0xFF   | Type (0x04)   | Raw binary payload |
-     *     +--------+---------------+--------+---------------+--------------------+
-     *
    - * The binary envelope is stripped to retrieve the inner packet, which is then - * processed normally (stripping the type prefix as described above). - *
    - * Ref: Engine.IO v3 Payload Spec - *
    - * "If the payload contains at least one binary packet, the payload is encoded as a binary buffer: - * - a binary indicator: 1 (representing a binary packet) or 0 (representing a string packet) - * - the length of the packet (as a series of characters) - * - a separator: 255 - * - the packet itself" - *
    - *
    • - *
- * - *

Engine.IO v4 (Socket.IO 3.x and newer)

- *
    - *
  • - * WebSocket/Polling (Raw Binary Frame): - * 
    -     *     +-------------------------------------------------+
-     *     | Bytes 0..N                                      |
-     *     +-------------------------------------------------+
-     *     | Raw binary payload                              |
-     *     +-------------------------------------------------+
-     *
    - * Engine.IO v4 does not prepend any packet types or metadata to binary attachments. - * The entire buffer is base64-encoded as-is and stored. - *
    - * Ref: Engine.IO v4 Protocol Spec - *
    - * "Binary packets are sent as-is without any modifications." - *
    - *
    • - *
+ * Adds an incoming binary attachment to a packet and completes the packet when all + * attachments have been received. * - * @param head the client connection head - * @param frame the incoming byte buffer frame + * @param head the client connection head + * @param frame the incoming attachment frame * @param binaryPacket the packet being assembled - * @return the packet if fully assembled (all attachments loaded), or an empty MESSAGE packet - * @throws IOException if a decoding error occurs + * @param transport the negotiated transport + * @return the completed packet, or an empty message packet while attachments remain + * incomplete + * @throws IOException if the attachment frame is malformed + * @throws IllegalStateException if an attachment placeholder cannot be found */ private Packet addAttachment(ClientHead head, ByteBuf frame, Packet binaryPacket, Transport transport) throws IOException { EngineIOVersion version = head.getEngineIOVersion(); @@ -605,6 +582,14 @@ private Packet addAttachment(ClientHead head, ByteBuf frame, Packet binaryPacket return new Packet(PacketType.MESSAGE, head.getEngineIOVersion()); } + /** + * Parses the body of a Socket.IO message according to its subtype. + * + * @param head the client connection associated with the packet + * @param frame the buffer containing the packet body + * @param packet the packet whose body is parsed + * @throws IOException if binary attachment processing fails + */ private void parseBody(ClientHead head, ByteBuf frame, Packet packet) throws IOException { // Early return for non-MESSAGE packets if (packet.getType() != PacketType.MESSAGE) { @@ -647,7 +632,11 @@ private void parseBody(ClientHead head, ByteBuf frame, Packet packet) throws IOE } /** - * Parse ERROR packet bodies + * Parses an ERROR packet body, including its optional namespace and error data. + * + * @param frame the buffer containing the packet body + * @param packet the packet to populate + * @throws IOException if the error data cannot be read as text */ private void parseErrorBody(ByteBuf frame, Packet packet) throws IOException { String nsp = readNamespace(frame, false); diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketEncoder.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketEncoder.java index ac0e568f..25c57aeb 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketEncoder.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/PacketEncoder.java @@ -59,6 +59,16 @@ public ByteBuf allocateBuffer(ByteBufAllocator allocator) { return allocator.heapBuffer(); } + /** + * Encodes queued packets and their binary attachments into an Engine.IO polling payload. + * + * @param jsonpIndex the JSONP callback index, or {@code null} to omit the JSONP wrapper + * @param packets the queue of packets to encode + * @param out the buffer receiving the encoded payload + * @param allocator the allocator used for temporary buffers + * @param limit the maximum number of packets to encode + * @throws IOException if packet serialization fails + */ public void encodeJsonP(Integer jsonpIndex, Queue packets, ByteBuf out, ByteBufAllocator allocator, int limit) throws IOException { boolean jsonpMode = jsonpIndex != null; @@ -112,6 +122,14 @@ public void encodeJsonP(Integer jsonpIndex, Queue packets, ByteBuf out, } + /** + * Writes the input bytes in UTF-8 form, escaping backslashes and single quotes + * when JSONP mode is enabled. + * + * @param in the source buffer + * @param out the destination buffer + * @param jsonpMode whether JSONP escaping is enabled + */ private void processUtf8(ByteBuf in, ByteBuf out, boolean jsonpMode) { while (in.isReadable()) { short value = (short) (in.readByte() & 0xFF); @@ -127,6 +145,16 @@ private void processUtf8(ByteBuf in, ByteBuf out, boolean jsonpMode) { } } + /** + * Encodes queued packets into the provided buffer according to their Engine.IO versions. + * + * @param packets the queue of packets to encode + * @param buffer the destination buffer + * @param allocator the allocator for temporary buffers + * @param limit the maximum number of packets to encode + * @throws IOException if packet serialization fails + * @throws IllegalStateException if a packet uses an unsupported Engine.IO version + */ public void encodePackets(Queue packets, ByteBuf buffer, ByteBufAllocator allocator, @@ -316,6 +344,16 @@ public static byte[] longToBytes(long number) { return res; } + /** + * Encodes a packet into the specified buffer, including its type, metadata, payload, + * and any attachment information. + * + * @param packet the packet to encode + * @param buffer the destination buffer + * @param allocator the allocator used for temporary buffers + * @param binary whether to write directly without text-packet framing + * @throws IOException if packet data serialization fails + */ public void encodePacket(Packet packet, ByteBuf buffer, ByteBufAllocator allocator, boolean binary) throws IOException { ByteBuf buf = buffer; if (!binary) { diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/EventMessageJsonSupport.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/EventMessageJsonSupport.java index 37de0865..044d56ea 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/EventMessageJsonSupport.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/event/EventMessageJsonSupport.java @@ -47,6 +47,12 @@ public final class EventMessageJsonSupport { private EventMessageJsonSupport() { } + /** + * Creates an {@link ObjectMapper} configured for JSON event message serialization and deserialization, + * including lossless encoding and decoding of byte arrays. + * + * @return an object mapper configured for event message JSON + */ public static ObjectMapper createObjectMapper() { SimpleModule module = new SimpleModule("EventMessageJsonModule"); @@ -85,12 +91,23 @@ public EventMessageObjectDeserializer() { super((JavaType) null, (JavaType) null); } + /** + * Deserializes an untyped JSON value and converts byte-array placeholders into byte arrays. + * + * @return the deserialized value with byte-array placeholders converted to {@code byte[]} + */ @Override public Object deserialize(JsonParser p, DeserializationContext ctxt) throws IOException { Object obj = super.deserialize(p, ctxt); return convertBytesPlaceholders(obj); } + /** + * Converts byte placeholders in nested maps and lists into byte arrays. + * + * @param obj the object to process + * @return the processed object with valid byte placeholders decoded + */ private Object convertBytesPlaceholders(Object obj) { if (obj instanceof Map) { Map map = (Map) obj; diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java index 6ee6c03e..4fee0d67 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/hazelcast/HazelcastPubSubEventStore.java @@ -104,6 +104,13 @@ public EventStoreMode getEventStoreMode(){ return eventStoreMode; } + /** + * Subscribes to events of the specified type from other nodes. + * + * @param type the event type to subscribe to + * @param listener the listener notified when a matching event is received + * @param clazz the event message class + */ @Override public void subscribe0(EventType type, final EventListener listener, Class clazz) { diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageDeserializer.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageDeserializer.java index 2ba1df28..9d5ab0ea 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageDeserializer.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageDeserializer.java @@ -34,6 +34,13 @@ public final class EventMessageDeserializer private static final ObjectMapper MAPPER = EventMessageJsonSupport.createObjectMapper(); + /** + * Deserializes message data into an {@code EventMessage}. + * + * @param topic the Kafka topic containing the message + * @param data the serialized message data + * @return the deserialized event message, or {@code null} when the data is empty or cannot be deserialized + */ @Override public EventMessage deserialize(String topic, byte[] data) { diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageSerializer.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageSerializer.java index 0c11ea52..84a8a5ba 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageSerializer.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/kafka/serialization/EventMessageSerializer.java @@ -35,6 +35,14 @@ public final class EventMessageSerializer private static final ObjectMapper MAPPER = EventMessageJsonSupport.createObjectMapper(); + /** + * Serializes an event message to JSON bytes. + * + * @param topic the Kafka topic associated with the message + * @param data the event message to serialize + * @return the serialized JSON bytes, or {@code null} when {@code data} is {@code null} + * @throws SerializationException if serialization fails + */ @Override public byte[] serialize(String topic, EventMessage data) { diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/EventMessageCodec.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/EventMessageCodec.java index 0899e97e..bf22536a 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/EventMessageCodec.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/nats_pubsub/EventMessageCodec.java @@ -29,6 +29,9 @@ public final class EventMessageCodec { private static final ObjectMapper MAPPER = EventMessageJsonSupport.createObjectMapper(); + /** + * Prevents instantiation of this utility class. + */ private EventMessageCodec() { } diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java index b55c8ce2..7ea23f55 100644 --- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java +++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/store/redis_stream/RedisStreamEventStore.java @@ -334,6 +334,9 @@ public void unsubscribe0(EventType type) { } } + /** + * Stops all polling activity and clears listeners, offsets, and stream references. + */ @Override public void shutdown0() { running.set(false); @@ -345,6 +348,12 @@ public void shutdown0() { subStreams.clear(); } + /** + * Determines whether a throwable indicates that Redisson has shut down. + * + * @param t the throwable to inspect + * @return {@code true} if the throwable indicates Redisson shutdown, {@code false} otherwise + */ private boolean isRedissonShutdown(Throwable t) { if (t == null) { return false; @@ -355,6 +364,12 @@ private boolean isRedissonShutdown(Throwable t) { return t.getMessage() != null && t.getMessage().contains("Redisson is shutdown"); } + /** + * Builds the Redis stream name for an event type. + * + * @param type the event type whose stream name is being built + * @return the configured prefix followed by the shared stream name in single-channel mode or the event type name otherwise + */ private String streamName(EventType type) { if (EventStoreMode.SINGLE_CHANNEL.equals(eventStoreMode)) { return streamNamePrefix + EventType.ALL_SINGLE_CHANNEL.name(); diff --git a/netty-socketio-core/src/test/java/com/socketio4j/socketio/store/CustomizedHazelcastContainer.java b/netty-socketio-core/src/test/java/com/socketio4j/socketio/store/CustomizedHazelcastContainer.java index cc36a028..668301f9 100644 --- a/netty-socketio-core/src/test/java/com/socketio4j/socketio/store/CustomizedHazelcastContainer.java +++ b/netty-socketio-core/src/test/java/com/socketio4j/socketio/store/CustomizedHazelcastContainer.java @@ -38,10 +38,16 @@ public class CustomizedHazelcastContainer extends GenericContainer