) future -> {
if (future.isSuccess()) {
- ChannelFuture cf = (ChannelFuture) future;
- serverChannel.set(cf.channel());
- if (configCopy.getPort() == 0) {
- try {
- InetSocketAddress local = (InetSocketAddress) cf.channel().localAddress();
- int actualPort = local.getPort();
- configCopy.setPort(actualPort);
- configuration.setPort(actualPort);
- } catch (Exception ignore) {
- // keep configured port if localAddress is not InetSocketAddress
+ try {
+ ChannelFuture cf = (ChannelFuture) future;
+ serverChannel.set(cf.channel());
+ if (configCopy.getPort() == 0) {
+ try {
+ InetSocketAddress local = (InetSocketAddress) cf.channel().localAddress();
+ int actualPort = local.getPort();
+ configCopy.setPort(actualPort);
+ configuration.setPort(actualPort);
+ } catch (Exception ignore) {
+ // keep configured port if localAddress is not InetSocketAddress
+ }
}
+ serverStatus.set(ServerStatus.STARTED);
+ log.info("SocketIO server started on port {}", configCopy.getPort());
+ installShutdownHookOnce();
+ fireAfterStart();
+ startPromise.setSuccess(null);
+ } catch (Exception e) {
+ serverStatus.set(ServerStatus.INIT);
+ cleanUpResources(false);
+ log.error("Server start error on port {}", configCopy.getPort(), e);
+ startPromise.setFailure(e);
}
- serverStatus.set(ServerStatus.STARTED);
- log.info("SocketIO server started on port {}", configCopy.getPort());
- installShutdownHookOnce();
- fireAfterStart();
} else {
serverStatus.set(ServerStatus.INIT);
log.error("Failed to start server on port {}", configCopy.getPort());
cleanUpResources(false);
+ startPromise.setFailure(future.cause());
}
});
+ return startPromise;
} catch (Exception e) {
serverStatus.set(ServerStatus.INIT);
@@ -1068,6 +1083,16 @@ public void addConnectListener(ConnectListener listener) {
mainNamespace.addConnectListener(listener);
}
+ @Override
+ public void removeConnectListener(ConnectListener listener) {
+ mainNamespace.removeConnectListener(listener);
+ }
+
+ @Override
+ public void removeDisconnectListener(DisconnectListener listener) {
+ mainNamespace.removeDisconnectListener(listener);
+ }
+
/**
* Registers a listener that is notified when a ping frame
* is received from a client.
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/annotation/Internal.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/annotation/Internal.java
new file mode 100644
index 00000000..d8ee2fee
--- /dev/null
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/annotation/Internal.java
@@ -0,0 +1,54 @@
+/**
+ * Copyright (c) 2025 The Socketio4j Project
+ * Parent project : Copyright (c) 2012-2025 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.socketio4j.socketio.annotation;
+
+/**
+ * @author https://github.com/sanjomo
+ * @date 02/08/26 6:40 pm
+ */
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import static java.lang.annotation.ElementType.CONSTRUCTOR;
+import static java.lang.annotation.ElementType.FIELD;
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.PACKAGE;
+import static java.lang.annotation.ElementType.TYPE;
+import static java.lang.annotation.RetentionPolicy.CLASS;
+
+/**
+ * Marks an API as internal to socketio4j.
+ *
+ * Types and members annotated with {@code @Internal} are implementation
+ * details and are NOT part of the supported public API.
+ * They may change, move, or be removed without notice in any release.
+ *
+ *
Application code should not depend on these APIs.
+ */
+@Documented
+@Retention(CLASS)
+@Target({
+ TYPE,
+ METHOD,
+ CONSTRUCTOR,
+ FIELD,
+ PACKAGE
+})
+public @interface Internal {
+}
\ No newline at end of file
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/AuthorizeHandler.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/AuthorizeHandler.java
index b41f71a2..a5e6c884 100644
--- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/AuthorizeHandler.java
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/AuthorizeHandler.java
@@ -61,6 +61,7 @@
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.QueryStringDecoder;
@@ -134,9 +135,38 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
return;
}
+ if (queryDecoder.path().equals(connectPath)
+ && !hasSupportedEngineIOVersion(queryDecoder.parameters())) {
+ writeAndFlushBadRequest(channel);
+ req.release();
+ return;
+ }
+
+ if (queryDecoder.path().equals(connectPath)
+ && !hasSupportedTransport(queryDecoder.parameters())) {
+ writeAndFlushTransportError(channel, req.headers().get(HttpHeaderNames.ORIGIN));
+ req.release();
+ return;
+ }
+
+ // A CORS preflight is not an Engine.IO session handshake. In
+ // particular it must not allocate a sid just because it has no sid.
+ if (HttpMethod.OPTIONS.equals(req.method())) {
+ ctx.fireChannelRead(msg);
+ return;
+ }
+
List sid = queryDecoder.parameters().get("sid");
if (queryDecoder.path().equals(connectPath)
&& sid == null) {
+ // An Engine.IO session is opened only by a GET (including the
+ // HTTP GET that upgrades to WebSocket). A POST/PUT without a
+ // sid is never a handshake and must not allocate a session.
+ if (!HttpMethod.GET.equals(req.method())) {
+ writeAndFlushBadRequest(channel);
+ req.release();
+ return;
+ }
if (log.isDebugEnabled()) {
log.debug("Processing new connection request from client: {}", channel.remoteAddress());
}
@@ -256,8 +286,8 @@ private boolean authorize(ChannelHandlerContext ctx, Channel channel, String ori
//:TODO lyjnew Current WEBSOCKET retrun upgrade[] engine-io protocol
// the test case line
// https://github.com/socketio/engine.io-protocol/blob/de247df875ddcd4778d1165829c8644301750e9f/test-suite/test-suite.js#L131C43-L131C43
- if (configuration.getTransports().contains(Transport.WEBSOCKET)
- && !(EngineIOVersion.V4.equals(client.getEngineIOVersion()) && Transport.WEBSOCKET.equals(client.getCurrentTransport()))) {
+ if (Transport.POLLING.equals(client.getCurrentTransport())
+ && configuration.getTransports().contains(Transport.WEBSOCKET)) {
transports = new String[]{"websocket"};
if (log.isDebugEnabled()) {
log.debug("WebSocket upgrade available for client: {}", channel.remoteAddress());
@@ -265,8 +295,8 @@ private boolean authorize(ChannelHandlerContext ctx, Channel channel, String ori
}
AuthPacket authPacket = new AuthPacket(sessionId, transports, configuration.getPingInterval(),
- configuration.getPingTimeout());
- Packet packet = new Packet(PacketType.OPEN, client.getEngineIOVersion());
+ configuration.getPingTimeout(), configuration.getMaxHttpContentLength());
+ Packet packet = new Packet(PacketType.OPEN);
packet.setData(authPacket);
if (log.isDebugEnabled()) {
@@ -281,6 +311,29 @@ private boolean authorize(ChannelHandlerContext ctx, Channel channel, String ori
return true;
}
+ private boolean hasSupportedEngineIOVersion(Map> params) {
+ List versions = params.get(EngineIOVersion.EIO);
+ return versions != null && versions.size() == 1 && EngineIOVersion.isSupported(versions.get(0));
+ }
+
+ private boolean hasSupportedTransport(Map> params) {
+ List transports = params.get("transport");
+ if (transports == null || transports.size() != 1) {
+ return false;
+ }
+ for (Transport transport : Transport.values()) {
+ if (transport.getValue().equals(transports.get(0))) {
+ return configuration.getTransports().contains(transport);
+ }
+ }
+ return false;
+ }
+
+ private void writeAndFlushBadRequest(Channel channel) {
+ channel.writeAndFlush(new DefaultHttpResponse(HTTP_1_1, HttpResponseStatus.BAD_REQUEST))
+ .addListener(ChannelFutureListener.CLOSE);
+ }
+
private void writeAndFlushTransportError(Channel channel, String origin) {
Map errorData = new HashMap<>();
errorData.put("code", 0);
@@ -335,18 +388,23 @@ public void connect(ClientHead client) {
log.debug("Connecting client: {} to default namespace", client.getSessionId());
}
+ if (EngineIOVersion.V4.equals(client.getEngineIOVersion())) {
+ // Socket.IO protocol v5 requires the client to explicitly send its CONNECT
+ // packet. Registering the default namespace here would invoke application
+ // connect listeners before authentication and make the later "40" a second
+ // connection attempt.
+ return;
+ }
+
Namespace ns = namespacesHub.get(Namespace.DEFAULT_NAME);
if (!client.getNamespaces().contains(ns)) {
- Packet packet = new Packet(PacketType.MESSAGE, client.getEngineIOVersion());
+ Packet packet = new Packet(PacketType.MESSAGE);
packet.setSubType(PacketType.CONNECT);
- //::TODO lyjnew V4 delay send connect packet ON client add Namecapse
- if (!EngineIOVersion.V4.equals(client.getEngineIOVersion())) {
- if (log.isDebugEnabled()) {
- log.debug("Sending CONNECT packet to client: {}", client.getSessionId());
- }
- client.send(packet);
+ if (log.isDebugEnabled()) {
+ log.debug("Sending CONNECT packet to client: {}", client.getSessionId());
}
+ client.send(packet);
configuration.getStoreFactory().eventStore().publish(EventType.CONNECT, new ConnectMessage(client.getSessionId()));
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/ClientHead.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/ClientHead.java
index 8ae0fe43..2799967e 100644
--- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/ClientHead.java
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/ClientHead.java
@@ -26,9 +26,13 @@
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,6 +53,7 @@
import com.socketio4j.socketio.store.StoreFactory;
import com.socketio4j.socketio.transport.NamespaceClient;
+import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
@@ -62,6 +67,8 @@ public class ClientHead {
public static final AttributeKey CLIENT = AttributeKey.valueOf("client");
private final AtomicBoolean disconnected = new AtomicBoolean();
+ private final AtomicBoolean pollingPostActive = new AtomicBoolean();
+ private final AtomicBoolean upgradeInProgress = new AtomicBoolean();
private final Map namespaceClients = new ConcurrentHashMap<>();
private final Map channels = new HashMap(2);
private final HandshakeData handshakeData;
@@ -77,6 +84,7 @@ public class ClientHead {
private final Configuration configuration;
private Packet lastBinaryPacket;
+ private ByteBuf lastBinaryPacketSource;
// TODO use lazy set
private volatile Transport currentTransport;
@@ -99,13 +107,16 @@ public ClientHead(UUID sessionId, AckManager ackManager, DisconnectableHub disco
List versions = params.getOrDefault(EngineIOVersion.EIO, new ArrayList<>());
if (versions.isEmpty()) {
- engineIOVersion = EngineIOVersion.UNKNOWN;
+ engineIOVersion = EngineIOVersion.V4;
} else {
engineIOVersion = EngineIOVersion.fromValue(versions.get(0));
}
}
public void bindChannel(Channel channel, Transport transport) {
+ if (!isConnected()) {
+ return;
+ }
log.debug("binding channel: {} to transport: {}", channel, transport);
TransportState state = channels.get(transport);
@@ -114,16 +125,70 @@ public void bindChannel(Channel channel, Transport transport) {
clientsBox.remove(prevChannel);
}
clientsBox.add(channel, this);
-
+ if (!isConnected()) {
+ clientsBox.remove(channel);
+ state.compareAndSet(channel, null);
+ return;
+ }
sendPackets(transport, channel);
}
+ /**
+ * Binds the outstanding long-poll response, rejecting a second concurrent
+ * GET instead of replacing the first response channel.
+ */
+ public boolean tryBindPollingChannel(Channel channel) {
+ return tryBindChannel(channel, Transport.POLLING);
+ }
+
+ /** Engine.IO permits only one WebSocket connection for a session. */
+ public boolean tryBindWebSocketChannel(Channel channel) {
+ return tryBindChannel(channel, Transport.WEBSOCKET);
+ }
+
+ private boolean tryBindChannel(Channel channel, Transport transport) {
+ if (!isConnected()) {
+ return false;
+ }
+
+ TransportState state = channels.get(transport);
+ for (;;) {
+ Channel current = state.getChannel();
+ if (current != null && current != channel && current.isActive()) {
+ return false;
+ }
+ if (!state.compareAndSet(current, channel)) {
+ continue;
+ }
+
+ log.debug("binding channel: {} to transport: {}", channel, transport);
+ if (current != null) {
+ clientsBox.remove(current);
+ }
+ clientsBox.add(channel, this);
+ if (!isConnected()) {
+ clientsBox.remove(channel);
+ state.compareAndSet(channel, null);
+ return false;
+ }
+ sendPackets(transport, channel);
+ return true;
+ }
+ }
+
+ /** Engine.IO permits only one polling POST to be active for a session. */
+ public boolean tryAcquirePollingPost() {
+ return pollingPostActive.compareAndSet(false, true);
+ }
+
+ public void releasePollingPost() {
+ pollingPostActive.set(false);
+ }
+
public void releasePollingChannel(Channel channel) {
try {
- TransportState state = channels.get(Transport.POLLING);
- if (channel.equals(state.getChannel())) {
+ if (channels.get(Transport.POLLING).compareAndSet(channel, null)) {
clientsBox.remove(channel);
- state.update(null);
}
} catch (Exception e) {
log.error("Failed to release polling channel for session: {}", sessionId, e);
@@ -134,7 +199,7 @@ public String getOrigin() {
return handshakeData.getHttpHeaders().get(HttpHeaderNames.ORIGIN);
}
- public ChannelFuture send(Packet packet) {
+ public @Nullable ChannelFuture send(Packet packet) {
return send(packet, getCurrentTransport());
}
@@ -164,7 +229,7 @@ public void schedulePing() {
EngineIOVersion version = client.getEngineIOVersion();
//only send ping packet for engine.io version 4
if (EngineIOVersion.V4.equals(version)) {
- client.send(new Packet(PacketType.PING, version));
+ client.send(new Packet(PacketType.PING));
}
schedulePing();
}
@@ -183,7 +248,7 @@ public void schedulePingTimeout() {
}, configuration.getPingTimeout() + configuration.getPingInterval(), TimeUnit.MILLISECONDS);
}
- public ChannelFuture send(Packet packet, Transport transport) {
+ public @Nullable ChannelFuture send(Packet packet, Transport transport) {
TransportState state = channels.get(transport);
state.getPacketsQueue().add(packet);
@@ -201,9 +266,10 @@ private ChannelFuture sendPackets(Transport transport, Channel channel) {
public void removeNamespaceClient(NamespaceClient client) {
namespaceClients.remove(client.getNamespace());
- if (namespaceClients.isEmpty()) {
- disconnectableHub.onDisconnect(this);
- }
+ // A Socket.IO namespace disconnect does not necessarily close the
+ // underlying Engine.IO session. Keep its SID registered until the
+ // transport closes so a polling client can finish its final request
+ // without receiving a spurious "Session ID unknown" response.
}
public NamespaceClient getChildClient(Namespace namespace) {
@@ -212,7 +278,21 @@ public NamespaceClient getChildClient(Namespace namespace) {
public NamespaceClient addNamespaceClient(Namespace namespace) {
NamespaceClient client = new NamespaceClient(this, namespace);
- namespaceClients.put(namespace, client);
+ return addNamespaceClient(client);
+ }
+
+ /**
+ * Registers a namespace client after protocol-level validation has succeeded.
+ * A Socket.IO v3/v4 CONNECT (wire protocol v5) carrying authentication
+ * data must not become visible to namespace listeners before that
+ * authentication has been accepted.
+ */
+ public NamespaceClient addNamespaceClient(NamespaceClient client) {
+ NamespaceClient existing = namespaceClients.putIfAbsent(client.getNamespace(), client);
+ if (existing != null) {
+ return existing;
+ }
+ client.getNamespace().addClient(client);
return client;
}
@@ -224,18 +304,128 @@ public boolean isConnected() {
return !disconnected.get();
}
+ private final List pollFlushedListeners = new CopyOnWriteArrayList<>();
+ private final AtomicLong pollFlushTimeoutSequence = new AtomicLong();
+
+ public boolean hasPollFlushedListeners() {
+ return !pollFlushedListeners.isEmpty();
+ }
+
+ public void onPollFlushed(Runnable listener, long gracePeriodMs) {
+ if (!isConnected()) {
+ listener.run();
+ return;
+ }
+
+ SchedulerKey timeoutKey = null;
+ if (gracePeriodMs > 0 && scheduler != null) {
+ timeoutKey = new SchedulerKey(SchedulerKey.Type.POLL_FLUSH_TIMEOUT,
+ sessionId.toString() + ":" + pollFlushTimeoutSequence.incrementAndGet());
+ }
+ PollFlushedListener pollFlushedListener = new PollFlushedListener(listener, timeoutKey);
+ pollFlushedListeners.add(pollFlushedListener);
+
+ if (timeoutKey != null) {
+ scheduler.schedule(timeoutKey, () -> {
+ if (pollFlushedListeners.remove(pollFlushedListener)) {
+ log.debug("Polling disconnect grace period expired for session {}, executing deferred cleanup", sessionId);
+ listener.run();
+ }
+ }, gracePeriodMs, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ public void notifyPollFlushed() {
+ if (!pollFlushedListeners.isEmpty()) {
+ List listeners = new ArrayList<>(pollFlushedListeners);
+ for (PollFlushedListener pollFlushedListener : listeners) {
+ if (!pollFlushedListeners.remove(pollFlushedListener)) {
+ continue;
+ }
+ if (pollFlushedListener.timeoutKey != null && scheduler != null) {
+ scheduler.cancel(pollFlushedListener.timeoutKey);
+ }
+ try {
+ pollFlushedListener.listener.run();
+ } catch (Exception e) {
+ log.error("Error executing poll flushed listener for session {}", sessionId, e);
+ }
+ }
+ }
+ }
+
+ private static final class PollFlushedListener {
+ private final Runnable listener;
+ private final SchedulerKey timeoutKey;
+
+ private PollFlushedListener(Runnable listener, SchedulerKey timeoutKey) {
+ this.listener = listener;
+ this.timeoutKey = timeoutKey;
+ }
+ }
+
public void onChannelDisconnect() {
+ if (!disconnected.compareAndSet(false, true)) {
+ return;
+ }
+ cleanupDisconnectedSession();
+ }
+
+ private void cleanupDisconnectedSession() {
+ for (Transport transport : Transport.values()) {
+ TransportState state = channels.get(transport);
+ Channel channel = state.getChannel();
+ if (channel != null && state.compareAndSet(channel, null)) {
+ clientsBox.remove(channel);
+ }
+ }
+
+ notifyPollFlushed();
cancelPing();
cancelPingTimeout();
+ clearPendingBinaryPacket();
- disconnected.set(true);
- for (NamespaceClient client : namespaceClients.values()) {
+ for (NamespaceClient client : new ArrayList<>(namespaceClients.values())) {
client.onDisconnect();
}
- for (TransportState state : channels.values()) {
- if (state.getChannel() != null) {
- clientsBox.remove(state.getChannel());
- }
+ // Namespace teardown and Engine.IO teardown are separate. Once the
+ // transport closes, remove the head whether or not it had namespaces
+ // when disconnect processing began.
+ disconnectableHub.onDisconnect(this);
+ }
+
+ /**
+ * Terminates an Engine.IO session because a Socket.IO protocol violation
+ * occurred. A polling GET can bind in parallel with the POST that carried
+ * the invalid packet, so queue a transport CLOSE before unregistering the
+ * session. This guarantees that such a poll is completed rather than
+ * remaining open after the session has been removed.
+ */
+ public void disconnectWithProtocolClose() {
+ if (!disconnected.compareAndSet(false, true)) {
+ return;
+ }
+
+ Transport closeTransport = currentTransport;
+ TransportState state = channels.get(closeTransport);
+ state.getPacketsQueue().add(new Packet(PacketType.CLOSE));
+ Channel closeChannel = state.getChannel();
+ ChannelFuture future = null;
+ if (closeChannel != null
+ && (closeTransport != Transport.POLLING
+ || closeChannel.attr(EncoderHandler.WRITE_ONCE).get() == null)) {
+ future = sendPackets(closeTransport, closeChannel);
+ }
+ cleanupDisconnectedSession();
+
+ if (future != null) {
+ future.addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ public void releaseTransport(Transport transport, Channel channel) {
+ if (channels.get(transport).compareAndSet(channel, null)) {
+ clientsBox.remove(channel);
}
}
@@ -256,20 +446,23 @@ public SocketAddress getRemoteAddress() {
}
public void disconnect() {
- Packet packet = new Packet(PacketType.MESSAGE, engineIOVersion);
+ if (!disconnected.compareAndSet(false, true)) {
+ return;
+ }
+ Packet packet = new Packet(PacketType.MESSAGE);
packet.setSubType(PacketType.DISCONNECT);
ChannelFuture future = send(packet);
if (future != null) {
future.addListener(ChannelFutureListener.CLOSE);
}
- onChannelDisconnect();
+ cleanupDisconnectedSession();
}
public boolean isChannelOpen() {
for (TransportState state : channels.values()) {
- if (state.getChannel() != null
- && state.getChannel().isActive()) {
+ Channel channel = state.getChannel();
+ if (channel != null && channel.isActive()) {
return true;
}
}
@@ -281,28 +474,37 @@ public Store getStore() {
}
public boolean isTransportChannel(Channel channel, Transport transport) {
- TransportState state = channels.get(transport);
- if (state.getChannel() == null) {
- return false;
- }
- return state.getChannel().equals(channel);
+ Channel current = channels.get(transport).getChannel();
+ return current != null && current.equals(channel);
+ }
+
+ public void beginUpgrade() {
+ upgradeInProgress.set(true);
+ }
+
+ public boolean isUpgradeInProgress() {
+ return upgradeInProgress.get();
}
public void upgradeCurrentTransport(Transport currentTransport) {
+ upgradeInProgress.set(false);
TransportState state = channels.get(currentTransport);
-
for (Entry entry : channels.entrySet()) {
if (!entry.getKey().equals(currentTransport)) {
-
Queue queue = entry.getValue().getPacketsQueue();
+ // NOOP only releases the old polling transport. Once the client
+ // has selected the new transport it must not be replayed over it.
+ queue.removeIf(packet -> packet.getType() == PacketType.NOOP);
state.setPacketsQueue(queue);
-
- sendPackets(currentTransport, state.getChannel());
this.currentTransport = currentTransport;
log.debug("Transport upgraded to: {} for: {}", currentTransport, sessionId);
break;
}
}
+ Channel channel = state.getChannel();
+ if (channel != null) {
+ sendPackets(currentTransport, channel);
+ }
}
public Transport getCurrentTransport() {
@@ -313,13 +515,30 @@ public Queue getPacketsQueue(Transport transport) {
return channels.get(transport).getPacketsQueue();
}
- public void setLastBinaryPacket(Packet lastBinaryPacket) {
- this.lastBinaryPacket = lastBinaryPacket;
- }
+
public Packet getLastBinaryPacket() {
return lastBinaryPacket;
}
+ public ByteBuf getLastBinaryPacketSource() {
+ return lastBinaryPacketSource;
+ }
+
+ public void setPendingBinaryPacket(@NotNull Packet packet, @NotNull ByteBuf source) {
+ if (this.lastBinaryPacketSource != null && this.lastBinaryPacketSource != source) {
+ this.lastBinaryPacketSource.release();
+ }
+ this.lastBinaryPacket = packet;
+ this.lastBinaryPacketSource = source;
+ }
+ public void clearPendingBinaryPacket() {
+ this.lastBinaryPacket = null;
+ if (lastBinaryPacketSource != null) {
+ lastBinaryPacketSource.release();
+ lastBinaryPacketSource = null;
+ }
+ }
+
public EngineIOVersion getEngineIOVersion() {
return engineIOVersion;
}
@@ -334,6 +553,4 @@ public boolean isWritable() {
Channel channel = state.getChannel();
return channel != null && channel.isWritable();
}
-
-
}
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 a697d231..151ffd67 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
@@ -36,6 +36,9 @@
import com.socketio4j.socketio.messages.OutPacketMessage;
import com.socketio4j.socketio.messages.XHROptionsMessage;
import com.socketio4j.socketio.messages.XHRPostMessage;
+import com.socketio4j.socketio.protocol.EncodePacketsResult;
+import com.socketio4j.socketio.protocol.EncodeResult;
+import com.socketio4j.socketio.protocol.EngineIOVersion;
import com.socketio4j.socketio.protocol.Packet;
import com.socketio4j.socketio.protocol.PacketEncoder;
@@ -58,7 +61,6 @@
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
-import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.util.Attribute;
@@ -119,8 +121,10 @@ private void readVersion() throws IOException {
private void write(XHROptionsMessage msg, ChannelHandlerContext ctx, ChannelPromise promise) {
HttpResponse res = new DefaultHttpResponse(HTTP_1_1, HttpResponseStatus.OK);
- res.headers().add(HttpHeaderNames.SET_COOKIE, "io=" + msg.getSessionId())
- .add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE)
+ if (msg.getSessionId() != null) {
+ res.headers().add(HttpHeaderNames.SET_COOKIE, "io=" + msg.getSessionId());
+ }
+ res.headers().add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE)
.add(HttpHeaderNames.ACCESS_CONTROL_ALLOW_HEADERS, HttpHeaderNames.CONTENT_TYPE);
String origin = ctx.channel().attr(ORIGIN).get();
@@ -177,6 +181,15 @@ private void sendMessage(HttpMessage msg, Channel channel, ByteBuf out, HttpResp
out.release();
}
+ if (msg instanceof OutPacketMessage) {
+ OutPacketMessage outMsg = (OutPacketMessage) msg;
+ if (outMsg.getClientHead().hasPollFlushedListeners()) {
+ promise.addListener(f -> {
+ outMsg.getClientHead().notifyPollFlushed();
+ });
+ }
+ }
+
channel.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT, promise).addListener(ChannelFutureListener.CLOSE);
}
private void sendError(HttpErrorMessage errorMsg, ChannelHandlerContext ctx, ChannelPromise promise) throws IOException {
@@ -261,9 +274,6 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
}
- private static final int FRAME_BUFFER_SIZE = 8192;
-
-
private void handleWebsocket(final OutPacketMessage msg, ChannelHandlerContext ctx, ChannelPromise promise) throws IOException {
if (log.isDebugEnabled()) {
log.debug("Starting WebSocket message processing, sessionId: {}", msg.getSessionId());
@@ -287,45 +297,21 @@ private void handleWebsocket(final OutPacketMessage msg, ChannelHandlerContext c
}
ByteBuf out = encoder.allocateBuffer(ctx.alloc());
- encoder.encodePacket(packet, out, ctx.alloc(), true);
+ EngineIOVersion engineIOVersion = msg.getClientHead().getEngineIOVersion();
+ EncodeResult encodeResult = encoder.encodePacket(engineIOVersion, packet, out, ctx.alloc(), true);
if (log.isTraceEnabled()) {
log.trace("Out message: {} sessionId: {}", out.toString(CharsetUtil.UTF_8), msg.getSessionId());
}
- if (out.isReadable() && out.readableBytes() > configuration.getMaxFramePayloadLength()) {
- if (log.isDebugEnabled()) {
- log.debug("Message exceeds max frame payload length ({} > {}), fragmenting into {} frames, sessionId: {}",
- out.readableBytes(), configuration.getMaxFramePayloadLength(),
- (out.readableBytes() + FRAME_BUFFER_SIZE - 1) / FRAME_BUFFER_SIZE, msg.getSessionId());
- }
-
- ByteBuf dstStart = out.readSlice(FRAME_BUFFER_SIZE);
- dstStart.retain();
- WebSocketFrame start = new TextWebSocketFrame(false, 0, dstStart);
- ctx.channel().write(start);
-
- int fragmentCount = 1;
- while (out.isReadable()) {
- int re = Math.min(out.readableBytes(), FRAME_BUFFER_SIZE);
- ByteBuf dst = out.readSlice(re);
- dst.retain();
- WebSocketFrame res = new ContinuationWebSocketFrame(!out.isReadable(), 0, dst);
- ctx.channel().write(res);
- fragmentCount++;
- }
-
- if (log.isDebugEnabled()) {
- log.debug("Message fragmented into {} frames, sessionId: {}", fragmentCount, msg.getSessionId());
- }
-
- out.release();
- ctx.channel().flush();
- } else if (out.isReadable()){
+ if (out.isReadable()) {
if (log.isDebugEnabled()) {
log.debug("Sending single WebSocket frame, size: {} bytes, sessionId: {}",
out.readableBytes(), msg.getSessionId());
}
+ // Engine.IO requires every packet to occupy exactly one
+ // WebSocket frame. The configured max frame payload applies to
+ // inbound validation; splitting here would alter packet framing.
WebSocketFrame res = new TextWebSocketFrame(out);
ctx.channel().writeAndFlush(res);
} else {
@@ -335,9 +321,12 @@ private void handleWebsocket(final OutPacketMessage msg, ChannelHandlerContext c
out.release();
}
- for (ByteBuf buf : packet.getAttachments()) {
+ for (ByteBuf buf : encodeResult.getAttachments()) {
ByteBuf outBuf = encoder.allocateBuffer(ctx.alloc());
- outBuf.writeByte(4);
+ if (EngineIOVersion.V3.equals(engineIOVersion)
+ || EngineIOVersion.V2.equals(engineIOVersion)) {
+ outBuf.writeByte(4);
+ }
outBuf.writeBytes(buf);
if (log.isTraceEnabled()) {
log.trace("Out attachment: {} sessionId: {}", ByteBufUtil.hexDump(outBuf), msg.getSessionId());
@@ -366,29 +355,46 @@ private void handleHTTP(OutPacketMessage msg, ChannelHandlerContext ctx, Channel
return;
}
- if (log.isDebugEnabled()) {
- log.debug("Processing HTTP polling with {} packets, sessionId: {}", queue.size(), msg.getSessionId());
+ ClientHead clientHead = msg.getClientHead();
+ ByteBuf out = encoder.allocateBuffer(ctx.alloc());
+ EngineIOVersion engineIOVersion = clientHead.getEngineIOVersion();
+ if (engineIOVersion == null) {
+ engineIOVersion = EngineIOVersion.V4;
}
- ByteBuf out = encoder.allocateBuffer(ctx.alloc());
Boolean b64 = ctx.channel().attr(EncoderHandler.B64).get();
- if (b64 != null && b64) {
- Integer jsonpIndex = ctx.channel().attr(EncoderHandler.JSONP_INDEX).get();
+ Integer jsonpIndex = ctx.channel().attr(EncoderHandler.JSONP_INDEX).get();
+ // Engine.IO v3 selects JSONP with j=; b64=1 is a separate
+ // capability flag for base64 polling. Both use the legacy payload
+ // encoder, while only JSONP must be returned as JavaScript.
+ // Socket.IO v3/v4 also sends b64=1 but uses EIOv4 text framing.
+ if (!EngineIOVersion.V4.equals(engineIOVersion)
+ && (Boolean.TRUE.equals(b64) || jsonpIndex != null)) {
if (log.isDebugEnabled()) {
log.debug("Using JSONP encoding, index: {}, sessionId: {}", jsonpIndex, msg.getSessionId());
}
- encoder.encodeJsonP(jsonpIndex, queue, out, ctx.alloc(), 50);
+ encoder.encodeJsonP(engineIOVersion, jsonpIndex, queue, out, ctx.alloc(), 50);
String type = "application/javascript";
if (jsonpIndex == null) {
type = "text/plain";
}
sendMessage(msg, channel, out, type, promise, HttpResponseStatus.OK);
} else {
+ EncodePacketsResult result = encoder.encodePackets(engineIOVersion, queue, out, ctx.alloc(), 50);
+ // Engine.IO v4 polling serializes every binary packet as base64 text
+ // ("b") in a record-separated text payload. Only the legacy
+ // v2/v3 binary payload format is sent as application/octet-stream.
+ String contentType;
+ if (result.hasBinary() && !EngineIOVersion.V4.equals(engineIOVersion))
+ contentType = "application/octet-stream";
+ else
+ contentType = "text/plain";
+
if (log.isDebugEnabled()) {
- log.debug("Using binary encoding, sessionId: {}", msg.getSessionId());
+ log.debug("Using {} encoding, sessionId: {}", contentType, msg.getSessionId());
}
- encoder.encodePackets(queue, out, ctx.alloc(), 50);
- sendMessage(msg, channel, out, "application/octet-stream", promise, HttpResponseStatus.OK);
+
+ sendMessage(msg, channel, out, contentType, promise, HttpResponseStatus.OK);
}
}
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 9a03e306..19e1c3f6 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
@@ -44,6 +44,7 @@
public class InPacketHandler extends SimpleChannelInboundHandler {
private static final Logger log = LoggerFactory.getLogger(InPacketHandler.class);
+ private static final int MAX_LOG_PREVIEW = 64;
private final PacketListener packetListener;
private final PacketDecoder decoder;
@@ -59,7 +60,7 @@ public InPacketHandler(PacketListener packetListener, PacketDecoder decoder, Nam
}
@Override
- protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsMessage message)
+ protected void channelRead0(ChannelHandlerContext ctx, PacketsMessage message)
throws Exception {
ByteBuf content = message.getContent();
ClientHead client = message.getClient();
@@ -71,7 +72,10 @@ protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsM
int packetsProcessed = 0;
while (content.isReadable()) {
try {
- Packet packet = decoder.decodePackets(content, client);
+ Packet packet = decoder.decodePackets(content, client, message.getTransport());
+ if (packet == null) {
+ continue;
+ }
packetsProcessed++;
if (log.isDebugEnabled()) {
@@ -80,6 +84,16 @@ protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsM
client.getSessionId(), packet.hasAttachments());
}
+ // Engine.IO control packets are connection-level packets: they are not
+ // scoped to a Socket.IO namespace. In particular, an Engine.IO v4 client
+ // is required to reply to the server PING before it sends its Socket.IO
+ // CONNECT packet, so routing them through NamespaceClient would silently
+ // drop a perfectly valid PONG from a newly opened connection.
+ if (packet.getType() != PacketType.MESSAGE) {
+ packetListener.onTransportPacket(packet, client, message.getTransport());
+ continue;
+ }
+
Namespace ns = namespacesHub.get(packet.getNsp());
if (ns == null) {
if (packet.getSubType() == PacketType.CONNECT) {
@@ -87,10 +101,10 @@ protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsM
log.debug("Sending error response for invalid namespace: {} to client: {}",
packet.getNsp(), client.getSessionId());
}
- Packet p = new Packet(PacketType.MESSAGE, client.getEngineIOVersion());
+ Packet p = new Packet(PacketType.MESSAGE);
p.setSubType(PacketType.ERROR);
p.setNsp(packet.getNsp());
- p.setData("Invalid namespace");
+ p.setData(toConnectErrorPayload(client, "Invalid namespace"));
client.send(p);
return;
}
@@ -103,18 +117,24 @@ protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsM
log.debug("Processing CONNECT packet for namespace: {} from client: {}, Engine.IO version: {}",
ns.getName(), client.getSessionId(), client.getEngineIOVersion());
}
-
- client.addNamespaceClient(ns);
- NamespaceClient nClient = client.getChildClient(ns);
- //:TODO lyjnew client namespace send connect packet 0+namespace socket io v4
- // https://socket.io/docs/v4/socket-io-protocol/#connection-to-a-namespace
+ NamespaceClient nClient = new NamespaceClient(client, ns);
if (EngineIOVersion.V4.equals(client.getEngineIOVersion())) {
- handleV4Connect(packet, client, ns, nClient);
+ if (!handleV4Connect(packet, client, ns, nClient)) {
+ return;
+ }
}
+ client.addNamespaceClient(nClient);
}
NamespaceClient nClient = client.getChildClient(ns);
if (nClient == null) {
+ if (EngineIOVersion.V4.equals(client.getEngineIOVersion())) {
+ // The Socket.IO v3/v4 wire protocol (protocol v5) requires
+ // CONNECT before any other packet on a namespace. Do not let
+ // an unconnected client emit events or ACKs into application code.
+ client.disconnectWithProtocolClose();
+ ctx.close();
+ }
log.debug("Can't find namespace client in namespace: {}, sessionId: {} probably it was disconnected.", ns.getName(), client.getSessionId());
return;
}
@@ -122,8 +142,14 @@ protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsM
if (log.isDebugEnabled()) {
log.debug("Packet has unloaded attachments, deferring processing for client: {}, namespace: {}",
client.getSessionId(), ns.getName());
+ log.debug("Waiting for binary attachment...");
}
- return;
+ // Continue decoding remaining packets in the current POST body.
+ // A polling request may contain:
+ // attachment(A), header(B), attachment(B)
+ // Returning here would abandon unread bytes and leave later
+ // binary attachments unprocessed.
+ continue;
}
packetListener.onPacket(packet, nClient, message.getTransport());
if (log.isDebugEnabled()) {
@@ -131,13 +157,17 @@ protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsM
client.getSessionId(), ns.getName());
}
} catch (Exception ex) {
- String c;
- if (content.refCnt() > 0) {
- c = content.toString(CharsetUtil.UTF_8);
- } else {
- c = "";
+ final int payloadSize;
+ if (content.refCnt() > 0) payloadSize = content.readableBytes();
+ else payloadSize = -1;
+ log.error("Error during data processing. Client sessionId: {}, payloadSize={} bytes",
+ client.getSessionId(), payloadSize, ex);
+ if (log.isTraceEnabled() && content.refCnt() > 0) {
+ int length = Math.min(payloadSize, MAX_LOG_PREVIEW);
+ log.trace("Error payload hex preview for sessionId {}: {}",
+ client.getSessionId(),
+ io.netty.buffer.ByteBufUtil.hexDump(content, content.readerIndex(), length));
}
- log.error("Error during data processing. Client sessionId: {}, data: {}", client.getSessionId(), c, ex);
throw ex;
}
}
@@ -146,6 +176,28 @@ protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, PacketsM
log.debug("Completed processing {} packets for client: {}", packetsProcessed, client.getSessionId());
}
}
+ private static Object toConnectErrorPayload(ClientHead client, Object errorData) {
+
+ if (client.getEngineIOVersion() == EngineIOVersion.V4) {
+ if (errorData instanceof Map) {
+ return errorData;
+ }
+
+ if (errorData != null) {
+ return Collections.singletonMap(
+ "message",
+ String.valueOf(errorData));
+ }
+ return Collections.singletonMap(
+ "message",
+ "Authentication failed");
+ }
+
+ if (errorData != null) {
+ return String.valueOf(errorData);
+ }
+ return "Authentication failed";
+ }
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
@@ -169,7 +221,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Excep
}
}
- private void handleV4Connect(Packet packet, ClientHead client, Namespace ns, NamespaceClient nClient) {
+ private boolean handleV4Connect(Packet packet, ClientHead client, Namespace ns, NamespaceClient nClient) {
if (log.isDebugEnabled()) {
log.debug("Starting Engine.IO v4 connect handling for client: {}, namespace: {}, hasAuthData: {}",
client.getSessionId(), ns.getName(), packet.getData() != null);
@@ -194,12 +246,12 @@ private void handleV4Connect(Packet packet, ClientHead client, Namespace ns, Nam
client.getSessionId(), ns.getName());
}
- Packet p = new Packet(PacketType.MESSAGE, client.getEngineIOVersion());
+ Packet p = new Packet(PacketType.MESSAGE);
p.setSubType(PacketType.ERROR);
p.setNsp(packet.getNsp());
p.setData(toConnectErrorPayload(allowAuth.getErrorData()));
client.send(p);
- return;
+ return false;
}
} else {
if (log.isDebugEnabled()) {
@@ -207,7 +259,7 @@ private void handleV4Connect(Packet packet, ClientHead client, Namespace ns, Nam
client.getSessionId(), ns.getName());
}
}
- Packet p = new Packet(PacketType.MESSAGE, client.getEngineIOVersion());
+ Packet p = new Packet(PacketType.MESSAGE);
p.setSubType(PacketType.CONNECT);
p.setNsp(packet.getNsp());
p.setData(new ConnPacket(client.getSessionId()));
@@ -216,6 +268,7 @@ private void handleV4Connect(Packet packet, ClientHead client, Namespace ns, Nam
log.debug("Completed Engine.IO v4 connect handling for client: {}, namespace: {}",
client.getSessionId(), ns.getName());
}
+ return true;
}
/**
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/PacketListener.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/PacketListener.java
index bb7df6e2..e30970a0 100644
--- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/PacketListener.java
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/PacketListener.java
@@ -32,6 +32,8 @@
import com.socketio4j.socketio.transport.NamespaceClient;
import com.socketio4j.socketio.transport.PollingTransport;
+import io.netty.channel.ChannelFuture;
+
public class PacketListener {
private final NamespacesHub namespacesHub;
@@ -45,6 +47,91 @@ public PacketListener(AckManager ackManager, NamespacesHub namespacesHub, Pollin
this.scheduler = scheduler;
}
+ /**
+ * Handles Engine.IO packets that are valid before any Socket.IO namespace has
+ * been connected. Engine.IO ping/pong and transport upgrade are session-level
+ * concerns, while {@link #onPacket(Packet, NamespaceClient, Transport)} handles
+ * the namespace-scoped Socket.IO message layer.
+ */
+ public void onTransportPacket(Packet packet, ClientHead client, Transport transport) {
+ switch (packet.getType()) {
+ case PING: {
+ boolean upgrading = "probe".equals(packet.getData())
+ && transport == Transport.WEBSOCKET
+ && client.getCurrentTransport() == Transport.POLLING;
+ // EIO v3 is client-ping/server-pong. EIO v4 reverses this
+ // direction, except for the PING "probe" sent on the temporary
+ // WebSocket while upgrading from polling.
+ if (EngineIOVersion.V4.equals(client.getEngineIOVersion()) && !upgrading) {
+ client.onChannelDisconnect();
+ return;
+ }
+ Packet outPacket = new Packet(PacketType.PONG);
+ outPacket.setData(packet.getData());
+ if (upgrading) {
+ ChannelFuture pongFuture = client.send(outPacket, transport);
+ if (pongFuture != null) {
+ pongFuture.addListener(future -> {
+ if (future.isSuccess()) {
+ client.beginUpgrade();
+ client.send(new Packet(PacketType.NOOP), Transport.POLLING);
+ }
+ });
+ }
+ } else {
+ client.send(outPacket, transport);
+ client.schedulePingTimeout();
+ }
+ notifyPing(client, packet, true);
+ break;
+ }
+ case PONG:
+ // EIO v4 is server-ping/client-pong. A PONG from an EIO v3
+ // client is therefore a protocol error.
+ if (!EngineIOVersion.V4.equals(client.getEngineIOVersion())) {
+ client.onChannelDisconnect();
+ return;
+ }
+ client.schedulePingTimeout();
+ notifyPing(client, packet, false);
+ break;
+
+ case UPGRADE:
+ // An upgrade is valid only after the WebSocket probe succeeded.
+ if (transport != Transport.WEBSOCKET || !client.isUpgradeInProgress()) {
+ client.onChannelDisconnect();
+ return;
+ }
+ client.schedulePingTimeout();
+ scheduler.cancel(new SchedulerKey(SchedulerKey.Type.UPGRADE_TIMEOUT, client.getSessionId()));
+ client.upgradeCurrentTransport(transport);
+ break;
+
+ case CLOSE:
+ client.onChannelDisconnect();
+ break;
+
+ default:
+ break;
+ }
+ }
+
+ private void notifyPing(ClientHead client, Packet packet, boolean ping) {
+ Namespace namespace = namespacesHub.get(packet.getNsp());
+ if (namespace == null) {
+ return;
+ }
+ NamespaceClient namespaceClient = client.getChildClient(namespace);
+ if (namespaceClient == null) {
+ return;
+ }
+ if (ping) {
+ namespace.onPing(namespaceClient);
+ } else {
+ namespace.onPong(namespaceClient);
+ }
+ }
+
public void onPacket(Packet packet, NamespaceClient client, Transport transport) {
final AckRequest ackRequest = new AckRequest(packet, client);
@@ -54,12 +141,12 @@ public void onPacket(Packet packet, NamespaceClient client, Transport transport)
switch (packet.getType()) {
case PING: {
- Packet outPacket = new Packet(PacketType.PONG, client.getEngineIOVersion());
+ Packet outPacket = new Packet(PacketType.PONG);
outPacket.setData(packet.getData());
// TODO use future
client.getBaseClient().send(outPacket, transport);
if ("probe".equals(packet.getData())) {
- client.getBaseClient().send(new Packet(PacketType.NOOP, client.getEngineIOVersion()), Transport.POLLING);
+ client.getBaseClient().send(new Packet(PacketType.NOOP), Transport.POLLING);
} else {
client.getBaseClient().schedulePingTimeout();
}
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/TransportState.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/TransportState.java
index 0674c94c..aafe024d 100644
--- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/TransportState.java
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/handler/TransportState.java
@@ -18,6 +18,7 @@
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicReference;
import com.socketio4j.socketio.protocol.Packet;
@@ -26,7 +27,7 @@
public class TransportState {
private Queue packetsQueue = new ConcurrentLinkedQueue<>();
- private Channel channel;
+ private final AtomicReference channel = new AtomicReference<>();
public void setPacketsQueue(Queue packetsQueue) {
this.packetsQueue = packetsQueue;
@@ -37,13 +38,15 @@ public Queue getPacketsQueue() {
}
public Channel getChannel() {
- return channel;
+ return channel.get();
}
public Channel update(Channel channel) {
- Channel prevChannel = this.channel;
- this.channel = channel;
- return prevChannel;
+ return this.channel.getAndSet(channel);
+ }
+
+ public boolean compareAndSet(Channel expected, Channel updated) {
+ return channel.compareAndSet(expected, updated);
}
}
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 36db5eb7..16c1cde1 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
@@ -27,8 +27,16 @@ public interface ClientListeners {
void addDisconnectListener(DisconnectListener listener);
+ default void removeDisconnectListener(DisconnectListener listener) {
+ throw new UnsupportedOperationException("removeDisconnectListener is not implemented");
+ }
+
void addConnectListener(ConnectListener listener);
+ default void removeConnectListener(ConnectListener listener) {
+ throw new UnsupportedOperationException("removeConnectListener is not implemented");
+ }
+
/**
* from v4, ping will always be sent by server except probe ping packet sent from client,
* and pong will always be responded by client while receiving ping except probe pong packet responded from server
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/DefaultExceptionListener.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/DefaultExceptionListener.java
index da18ec4b..16c33af5 100644
--- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/DefaultExceptionListener.java
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/listener/DefaultExceptionListener.java
@@ -16,7 +16,11 @@
*/
package com.socketio4j.socketio.listener;
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.channels.ClosedChannelException;
import java.util.List;
+import java.util.Locale;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,11 +59,50 @@ public void onPongException(Exception e, SocketIOClient client) {
}
@Override
- public boolean exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
- log.error(e.getMessage(), e);
+ public boolean exceptionCaught(ChannelHandlerContext ctx, Throwable e) {
+ logException(e);
return true;
}
+ private void logException(Throwable t) {
+ if (log.isDebugEnabled()) {
+ log.debug("Exception caught", t);
+ return;
+ }
+
+ if (!isExpectedDisconnect(t)) {
+ log.error("Unhandled exception", t);
+ }
+ }
+
+ private boolean isExpectedDisconnect(Throwable t) {
+ while (t != null) {
+ if (t instanceof ClosedChannelException
+ || t instanceof EOFException) {
+ return true;
+ }
+
+ if (t instanceof IOException) {
+ String msg = t.getMessage();
+ if (msg != null) {
+ msg = msg.toLowerCase(Locale.ROOT);
+ if (msg.contains("connection reset")
+ || msg.contains("broken pipe")
+ || msg.contains("connection aborted")
+ || msg.contains("connection closed")
+ || msg.contains("forcibly closed")
+ || msg.contains("software caused connection abort")) {
+ return true;
+ }
+ }
+ }
+
+ t = t.getCause();
+ }
+
+ return false;
+ }
+
@Override
public void onAuthException(Throwable e, SocketIOClient client) {
log.error(e.getMessage(), e);
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 2168f802..6aebfee7 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
@@ -149,6 +149,68 @@ public void removeAllListeners(String eventName) {
}
}
+ /**
+ * Verifies that no client or room membership remains in this namespace.
+ * Package-private so test infrastructure can enforce reuse isolation
+ * without expanding the public Socket.IO API.
+ */
+ void assertEmptyForTestReuse(String phase) {
+ if (!allClients.isEmpty() || !roomClients.isEmpty() || !clientRooms.isEmpty()) {
+ throw new IllegalStateException("Namespace '" + name + "' retained state " + phase
+ + ": clients=" + allClients.keySet()
+ + ", rooms=" + roomClients.keySet()
+ + ", clientRooms=" + clientRooms.keySet());
+ }
+ }
+
+ /**
+ * Removes every listener type and its JSON event mapping. This is only
+ * visible to package-level test infrastructure used by reusable test
+ * servers; production callers retain the existing targeted APIs.
+ */
+ void clearListenersForTestReuse() {
+ for (String eventName : new ArrayList<>(eventListeners.keySet())) {
+ removeAllListeners(eventName);
+ }
+ catchAllEventListeners.clear();
+ connectListeners.clear();
+ disconnectListeners.clear();
+ pingListeners.clear();
+ pongListeners.clear();
+ eventInterceptors.clear();
+ authDataInterceptors.clear();
+
+ if (!eventListeners.isEmpty()
+ || !catchAllEventListeners.isEmpty()
+ || !connectListeners.isEmpty()
+ || !disconnectListeners.isEmpty()
+ || !pingListeners.isEmpty()
+ || !pongListeners.isEmpty()
+ || !eventInterceptors.isEmpty()
+ || !authDataInterceptors.isEmpty()) {
+ throw new IllegalStateException("Namespace '" + name
+ + "' retained listeners after reusable-test cleanup");
+ }
+ }
+
+ /**
+ * Fails if a reusable test starts with an event or lifecycle callback from
+ * a previous case.
+ */
+ void assertNoListenersForTestReuse(String phase) {
+ if (!eventListeners.isEmpty()
+ || !catchAllEventListeners.isEmpty()
+ || !connectListeners.isEmpty()
+ || !disconnectListeners.isEmpty()
+ || !pingListeners.isEmpty()
+ || !pongListeners.isEmpty()
+ || !eventInterceptors.isEmpty()
+ || !authDataInterceptors.isEmpty()) {
+ throw new IllegalStateException("Namespace '" + name
+ + "' retained listeners " + phase);
+ }
+ }
+
@Override
public void addOnAnyEventListener(CatchAllEventListener listener) {
catchAllEventListeners.add(listener);
@@ -289,6 +351,16 @@ public void addConnectListener(ConnectListener listener) {
connectListeners.add(listener);
}
+ @Override
+ public void removeConnectListener(ConnectListener listener) {
+ connectListeners.remove(listener);
+ }
+
+ @Override
+ public void removeDisconnectListener(DisconnectListener listener) {
+ disconnectListeners.remove(listener);
+ }
+
public void onConnect(SocketIOClient client) {
if (roomClients.containsKey(getName())
&& roomClients.get(getName()).contains(client.getSessionId())) {
@@ -424,7 +496,9 @@ public void dispatch(String room, Packet packet) {
int size = forEachRoomClient(room, client -> {
client.send(packet);
});
-
+ if (log.isDebugEnabled()) {
+ log.debug("[DISPATCH] namespace={} room={} → found {} local client(s)", name, room, size);
+ }
if (size > 0) {
metrics.eventSent(name, size);
}
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/AuthPacket.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/AuthPacket.java
index 6cbc76b2..21dc908e 100644
--- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/AuthPacket.java
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/AuthPacket.java
@@ -25,13 +25,19 @@ public class AuthPacket {
private final String[] upgrades;
private final int pingInterval;
private final int pingTimeout;
+ private final int maxPayload;
public AuthPacket(UUID sid, String[] upgrades, int pingInterval, int pingTimeout) {
+ this(sid, upgrades, pingInterval, pingTimeout, 0);
+ }
+
+ public AuthPacket(UUID sid, String[] upgrades, int pingInterval, int pingTimeout, int maxPayload) {
super();
this.sid = sid;
this.upgrades = upgrades;
this.pingInterval = pingInterval;
this.pingTimeout = pingTimeout;
+ this.maxPayload = maxPayload;
}
public int getPingInterval() {
@@ -42,6 +48,16 @@ public int getPingTimeout() {
return pingTimeout;
}
+ /**
+ * Maximum size, in bytes, of an Engine.IO polling payload.
+ *
+ * Engine.IO v4 clients use this handshake value to decide how many packets
+ * to aggregate in a single polling POST.
+ */
+ public int getMaxPayload() {
+ return maxPayload;
+ }
+
public UUID getSid() {
return sid;
}
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EncodePacketsResult.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EncodePacketsResult.java
new file mode 100644
index 00000000..960cee93
--- /dev/null
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EncodePacketsResult.java
@@ -0,0 +1,37 @@
+/**
+ * Copyright (c) 2025 The Socketio4j Project
+ * Parent project : Copyright (c) 2012-2025 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.socketio4j.socketio.protocol;
+
+import com.socketio4j.socketio.annotation.Internal;
+
+/**
+ * @author https://github.com/sanjomo
+ * @date 02/08/26 2:53 am
+ */
+@Internal
+public final class EncodePacketsResult {
+
+ private final boolean hasBinary;
+
+ public EncodePacketsResult(boolean hasBinary) {
+ this.hasBinary = hasBinary;
+ }
+
+ public boolean hasBinary() {
+ return hasBinary;
+ }
+}
\ No newline at end of file
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EncodeResult.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EncodeResult.java
new file mode 100644
index 00000000..5f9c1f51
--- /dev/null
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EncodeResult.java
@@ -0,0 +1,71 @@
+/**
+ * Copyright (c) 2025 The Socketio4j Project
+ * Parent project : Copyright (c) 2012-2025 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.socketio4j.socketio.protocol;
+
+
+import java.util.Collections;
+import java.util.List;
+
+import com.socketio4j.socketio.annotation.Internal;
+
+import io.netty.buffer.ByteBuf;
+
+
+
+/**
+ * @author https://github.com/sanjomo
+ * @date 02/08/26 2:36 am
+ */
+@Internal
+public final class EncodeResult {
+
+ private final ByteBuf encodedPacket;
+ private final List attachments;
+
+ public EncodeResult(ByteBuf encodedPacket, List attachments) {
+ this.encodedPacket = encodedPacket;
+ if (attachments == null) {
+ this.attachments = Collections.emptyList();
+ } else {
+ this.attachments = attachments;
+ }
+ }
+
+ public ByteBuf getEncodedPacket() {
+ return encodedPacket;
+ }
+
+ public List getAttachments() {
+ return attachments;
+ }
+
+ public boolean hasAttachments() {
+ return !attachments.isEmpty();
+ }
+
+ public int getAttachmentsCount() {
+ return attachments.size();
+ }
+
+ @Override
+ public String toString() {
+ return "EncodeResult{" +
+ "encodedPacket=" + encodedPacket +
+ ", attachments=" + attachments.size() +
+ '}';
+ }
+}
\ No newline at end of file
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EngineIOVersion.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EngineIOVersion.java
index e31e7e38..ad282049 100644
--- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EngineIOVersion.java
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/EngineIOVersion.java
@@ -19,9 +19,12 @@
import java.util.HashMap;
import java.util.Map;
+import com.socketio4j.socketio.annotation.Internal;
+
/**
* Engine.IO protocol version
*/
+@Internal
public enum EngineIOVersion {
/**
* @link Engine.IO version 2
@@ -35,9 +38,7 @@ public enum EngineIOVersion {
* current version
* @link Engine.IO version 4
*/
- V4("4"),
-
- UNKNOWN("");
+ V4("4");
public static final String EIO = "EIO";
@@ -64,6 +65,16 @@ public static EngineIOVersion fromValue(String value) {
if (engineIOVersion != null) {
return engineIOVersion;
}
- return UNKNOWN;
+ return V4;
+ }
+
+ /**
+ * Whether a query-string EIO value names a protocol revision this server
+ * actually implements. {@link #fromValue(String)} deliberately retains its
+ * historic v4 fallback for internal callers; HTTP handshakes must reject
+ * missing and unknown revisions instead of silently negotiating v4.
+ */
+ public static boolean isSupported(String value) {
+ return VERSIONS.containsKey(value);
}
}
diff --git a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Event.java b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Event.java
index 2e57747e..01e4a58d 100644
--- a/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Event.java
+++ b/netty-socketio-core/src/main/java/com/socketio4j/socketio/protocol/Event.java
@@ -18,6 +18,9 @@
import java.util.List;
+import com.socketio4j.socketio.annotation.Internal;
+
+@Internal
public class Event {
private String name;
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 c7b1c007..0ecb5836 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
@@ -96,11 +96,22 @@ public AckArgs deserialize(JsonParser jp, DeserializationContext ctxt) throws IO
}
JsonNode arg = iter.next();
- if (arg.isTextual() || arg.isBoolean()) {
+ if ((arg.isTextual() || arg.isBoolean()) && !byte[].class.equals(clazz)) {
clazz = Object.class;
}
- val = mapper.treeToValue(arg, clazz);
+ // Fix: HTTP Polling form-urlencoded decoding converts '+' in Base64 strings to ' ' (0x20).
+ // Intercept byte[] textual nodes, restore '+' characters, and decode directly via Base64.
+ if (byte[].class.equals(clazz) && arg.isTextual()) {
+ String text = arg.asText();
+ if (text.contains(" ")) {
+ text = text.replace(' ', '+');
+ }
+ val = java.util.Base64.getDecoder().decode(text);
+ } else {
+ val = mapper.treeToValue(arg, clazz);
+ }
+
args.add(val);
i++;
}
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 5e90a66a..53917113 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
@@ -21,23 +21,25 @@
import java.util.Collections;
import java.util.List;
+import com.socketio4j.socketio.annotation.Internal;
import com.socketio4j.socketio.namespace.Namespace;
import io.netty.buffer.ByteBuf;
+@Internal
public class Packet implements Serializable {
private static final long serialVersionUID = 4560159536486711426L;
private PacketType type;
- private EngineIOVersion engineIOVersion;
+
private PacketType subType;
private Long ackId;
private String name;
private String nsp = Namespace.DEFAULT_NAME;
+
private Object data;
- private ByteBuf dataSource;
private int attachmentsCount;
private List attachments = Collections.emptyList();
@@ -49,10 +51,6 @@ public Packet(PacketType type) {
super();
this.type = type;
}
- public Packet(PacketType type, EngineIOVersion engineIOVersion) {
- this(type);
- this.engineIOVersion = engineIOVersion;
- }
public PacketType getSubType() {
return subType;
@@ -93,14 +91,13 @@ public T getData() {
* @param engineIOVersion
* @return packet
*/
- public Packet withNsp(String namespace, EngineIOVersion engineIOVersion) {
+ public Packet withNsp(String namespace) {
if (this.nsp.equalsIgnoreCase(namespace)) {
return this;
} else {
- Packet newPacket = new Packet(this.type, engineIOVersion);
+ Packet newPacket = new Packet(this.type);
newPacket.setAckId(this.ackId);
newPacket.setData(this.data);
- newPacket.setDataSource(this.dataSource);
newPacket.setName(this.name);
newPacket.setSubType(this.subType);
newPacket.setNsp(namespace);
@@ -109,7 +106,6 @@ public Packet withNsp(String namespace, EngineIOVersion engineIOVersion) {
return newPacket;
}
}
-
public void setNsp(String endpoint) {
//patch for #903
if ("{}".equals(endpoint)){
@@ -161,21 +157,6 @@ public boolean isAttachmentsLoaded() {
return this.attachments.size() == attachmentsCount;
}
- public ByteBuf getDataSource() {
- return dataSource;
- }
- public void setDataSource(ByteBuf dataSource) {
- this.dataSource = dataSource;
- }
-
- public EngineIOVersion getEngineIOVersion() {
- return engineIOVersion;
- }
-
- public void setEngineIOVersion(EngineIOVersion engineIOVersion) {
- this.engineIOVersion = engineIOVersion;
- }
-
@Override
public String toString() {
return "Packet [type=" + type + ", ackId=" + ackId + "]";
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 9a0562c7..6dfea299 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
@@ -21,11 +21,14 @@
import java.util.LinkedList;
import java.util.Map;
+import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.socketio4j.socketio.AckCallback;
+import com.socketio4j.socketio.Transport;
import com.socketio4j.socketio.ack.AckManager;
+import com.socketio4j.socketio.annotation.Internal;
import com.socketio4j.socketio.handler.ClientHead;
import com.socketio4j.socketio.namespace.Namespace;
@@ -35,6 +38,7 @@
import io.netty.handler.codec.base64.Base64;
import io.netty.util.CharsetUtil;
+@Internal
public class PacketDecoder {
private static final Logger log = LoggerFactory.getLogger(PacketDecoder.class);
@@ -54,6 +58,37 @@ private boolean isStringPacket(ByteBuf content) {
return content.getByte(content.readerIndex()) == 0x0;
}
+ /**
+ * Engine.IO v2/v3 encodes a polling payload containing binary as a series
+ * of frames: {@code <0 = string | 1 = binary><0xFF>}.
+ * The length digits are bytes in the 0..9 range, not ASCII characters.
+ */
+ private boolean hasLegacyBinaryPayloadHeader(ByteBuf buffer) {
+ if (buffer.readableBytes() < 3) {
+ return false;
+ }
+
+ int readerIndex = buffer.readerIndex();
+ byte marker = buffer.getByte(readerIndex);
+ if (marker != 0 && marker != 1) {
+ return false;
+ }
+
+ int maxHeaderLength = Math.min(buffer.readableBytes(), 12);
+ int separatorIndex = buffer.bytesBefore(maxHeaderLength, (byte) -1);
+ if (separatorIndex <= 1) {
+ return false;
+ }
+
+ for (int i = 1; i < separatorIndex; i++) {
+ byte digit = buffer.getByte(readerIndex + i);
+ if ((digit < 0 || digit > 9) && (digit < '0' || digit > '9')) {
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* True zero-copy optimized version of preprocessJson that works directly with ByteBuf
* without string conversion and without creating new ByteBuf instances.
@@ -195,25 +230,70 @@ private int hexToInt(byte b) {
// fastest way to parse chars to int
private long readLong(ByteBuf chars, int length) {
+ if (length < 0 || length > chars.readableBytes()) {
+ throw new IllegalArgumentException("Invalid numeric field length: " + length);
+ }
long result = 0;
for (int i = chars.readerIndex(); i < chars.readerIndex() + length; i++) {
- int digit = (chars.getByte(i) & 0xF);
- for (int j = 0; j < chars.readerIndex() + length-1-i; j++) {
- digit *= 10;
+ byte value = chars.getByte(i);
+ if (value < '0' || value > '9') {
+ throw new IllegalArgumentException("Non-decimal byte in numeric packet field: " + (char) value);
+ }
+ int digit = value - '0';
+ if (result > (Long.MAX_VALUE - digit) / 10) {
+ throw new IllegalArgumentException("Numeric packet field overflow");
+ }
+ result = result * 10 + digit;
+ }
+ chars.readerIndex(chars.readerIndex() + length);
+ return result;
+ }
+
+ /**
+ * Engine.IO v2/v3's XHR2 binary wrapper encodes its length as either
+ * byte-valued digits (0..9) or ASCII digits. This representation is
+ * specific to that wrapper; all text packet headers use {@link #readLong}
+ * and must contain ASCII decimal characters.
+ */
+ private long readLegacyBinaryLength(ByteBuf chars, int length) {
+ if (length < 0 || length > chars.readableBytes()) {
+ throw new IllegalArgumentException("Invalid legacy binary length: " + length);
+ }
+ long result = 0;
+ for (int i = chars.readerIndex(); i < chars.readerIndex() + length; i++) {
+ byte value = chars.getByte(i);
+ int digit;
+ if (value >= 0 && value <= 9) {
+ digit = value;
+ } else if (value >= '0' && value <= '9') {
+ digit = value - '0';
+ } else {
+ throw new IllegalArgumentException("Non-decimal byte in legacy binary length: " + value);
+ }
+ if (result > (Long.MAX_VALUE - digit) / 10) {
+ throw new IllegalArgumentException("Legacy binary length overflow");
}
- result += digit;
+ result = result * 10 + digit;
}
chars.readerIndex(chars.readerIndex() + length);
return result;
}
private PacketType readType(ByteBuf buffer) {
- int typeId = buffer.readByte() & 0xF;
+ byte value = buffer.readByte();
+ if (value < '0' || value > '6') {
+ throw new IllegalArgumentException("Invalid Engine.IO packet type: " + (char) value);
+ }
+ int typeId = value - '0';
return PacketType.valueOf(typeId);
}
private PacketType readInnerType(ByteBuf buffer) {
- int typeId = buffer.readByte() & 0xF;
+ byte value = buffer.readByte();
+ if (value < '0' || value > '6') {
+ throw new IllegalArgumentException("Invalid Socket.IO packet type: " + (char) value);
+ }
+ int typeId = value - '0';
return PacketType.valueOfInner(typeId);
}
@@ -231,47 +311,113 @@ private boolean hasLengthHeader(ByteBuf buffer) {
}
public Packet decodePackets(ByteBuf buffer, ClientHead client) throws IOException {
+ return decodePackets(buffer, client, client.getCurrentTransport());
+ }
+
+ public @Nullable Packet decodePackets(ByteBuf buffer,
+ ClientHead client,
+ Transport transport) throws IOException {
+
+ if (transport == Transport.POLLING && hasLegacyBinaryPayloadHeader(buffer)) {
+ return decodeLegacyBinaryPayload(buffer, client, transport);
+ }
+
+ Packet pending = client.getLastBinaryPacket();
+
+ if (pending != null
+ && pending.hasAttachments()
+ && !pending.isAttachmentsLoaded()) {
+
+ if (transport == Transport.WEBSOCKET) {
+ return decode(client, buffer, transport);
+ }
+ }
+
if (isStringPacket(buffer)) {
- return decodeWithStringHeader(buffer, client);
- } else if (hasLengthHeader(buffer)) {
- return decodeWithLengthHeader(buffer, client);
+ return decodeWithStringHeader(buffer, client, transport);
}
- return decode(client, buffer);
+
+ if (hasLengthHeader(buffer)) {
+ return decodeWithLengthHeader(buffer, client, transport);
+ }
+
+ return decode(client, buffer, transport);
+ }
+
+ private Packet decodeLegacyBinaryPayload(ByteBuf buffer,
+ ClientHead client,
+ Transport transport) throws IOException {
+ byte marker = buffer.readByte();
+ int maxHeaderLength = Math.min(buffer.readableBytes(), 11);
+ int lengthHeaderSize = buffer.bytesBefore(maxHeaderLength, (byte) -1);
+ if (lengthHeaderSize <= 0) {
+ throw new IOException("Malformed legacy polling payload: missing length separator");
+ }
+
+ long rawLength = readLegacyBinaryLength(buffer, lengthHeaderSize);
+ if (rawLength < 0 || rawLength > Integer.MAX_VALUE) {
+ throw new IOException("Malformed legacy polling payload: length overflow " + rawLength);
+ }
+ if (!buffer.isReadable() || buffer.readByte() != (byte) -1) {
+ throw new IOException("Malformed legacy polling payload: missing 0xFF separator");
+ }
+
+ int length = (int) rawLength;
+ if (length > buffer.readableBytes()) {
+ throw new IOException("Malformed legacy polling payload: length " + length
+ + " exceeds remaining bytes " + buffer.readableBytes());
+ }
+ ByteBuf payload = buffer.readSlice(length);
+
+ if (marker == 0) {
+ Packet pending = client.getLastBinaryPacket();
+ if (pending != null && pending.hasAttachments() && !pending.isAttachmentsLoaded()
+ && payload.isReadable() && payload.getByte(payload.readerIndex()) == 'b') {
+ return addAttachment(client, payload, pending, transport);
+ }
+ return decode(client, payload, transport);
+ }
+
+ Packet pending = client.getLastBinaryPacket();
+ if (pending == null || !pending.hasAttachments() || pending.isAttachmentsLoaded()) {
+ throw new IOException("Unexpected binary Engine.IO polling payload without a pending attachment packet");
+ }
+ return addLegacyPollingBinaryAttachment(client, payload, pending);
}
/**
* Decode packet with string header format
* Handles packets that start with 0x0 byte
*/
- private Packet decodeWithStringHeader(ByteBuf buffer, ClientHead client) throws IOException {
+ private Packet decodeWithStringHeader(ByteBuf buffer, ClientHead client, Transport transport) throws IOException {
int maxLength = Math.min(buffer.readableBytes(), 10);
int headEndIndex = buffer.bytesBefore(maxLength, (byte) -1);
if (headEndIndex == -1) {
headEndIndex = buffer.bytesBefore(maxLength, (byte) 0x3f);
}
int len = (int) readLong(buffer, headEndIndex);
- return decodeFrame(buffer, client, len);
+ return decodeFrame(buffer, client, len, transport);
}
/**
* Decode packet with length header format
* Handles packets with format "length:data"
*/
- private Packet decodeWithLengthHeader(ByteBuf buffer, ClientHead client) throws IOException {
+ private Packet decodeWithLengthHeader(ByteBuf buffer, ClientHead client, Transport transport) throws IOException {
int lengthEndIndex = buffer.bytesBefore((byte) ':');
int lenHeader = (int) readLong(buffer, lengthEndIndex);
int len = utf8scanner.getActualLength(buffer, lenHeader);
- return decodeFrame(buffer, client, len);
+ return decodeFrame(buffer, client, len, transport);
}
/**
* Common frame decoding logic
* Extracts frame data and advances buffer position
*/
- private Packet decodeFrame(ByteBuf buffer, ClientHead client, int len) throws IOException {
+ private Packet decodeFrame(ByteBuf buffer, ClientHead client, int len, Transport transport) throws IOException {
ByteBuf frame = buffer.slice(buffer.readerIndex() + 1, len);
buffer.readerIndex(buffer.readerIndex() + 1 + len);
- return decode(client, frame);
+ return decode(client, frame, transport);
}
private String readString(ByteBuf frame) {
@@ -284,7 +430,7 @@ private String readString(ByteBuf frame, int size) {
return new String(bytes, CharsetUtil.UTF_8);
}
- private Packet decode(ClientHead head, ByteBuf frame) throws IOException {
+ private @Nullable Packet decode(ClientHead head, ByteBuf frame, Transport transport) throws IOException {
Packet lastPacket = head.getLastBinaryPacket();
// Assume attachments follow.
@@ -293,24 +439,34 @@ private Packet decode(ClientHead head, ByteBuf frame) throws IOException {
&& lastPacket.hasAttachments()
&& !lastPacket.isAttachmentsLoaded()
) {
- return addAttachment(head, frame, lastPacket);
- }
+ return addAttachment(head, frame, lastPacket, transport);
+ }
+ // Skip any leading 0x1E record separators (e.g. payload starting with 0x1e or consecutive 0x1e delimiters)
+ while (frame.readableBytes() > 0 && frame.getByte(frame.readerIndex()) == 0x1E) {
+ frame.skipBytes(1);
+ }
+ if (!frame.isReadable()) {
+ return null;
+ }
final int separatorPos = frame.bytesBefore((byte) 0x1E);
final ByteBuf packetBuf;
- if (separatorPos > 0) {
- // Multiple packets in one, copy out the next packet to parse
- packetBuf = frame.copy(frame.readerIndex(), separatorPos);
- frame.skipBytes(separatorPos + 1);
+ if (separatorPos >= 0) {
+ packetBuf = frame.readSlice(separatorPos);
+ frame.skipBytes(1); // skip 0x1E separator
} else {
packetBuf = frame;
}
+ if (!packetBuf.isReadable()) {
+ return null;
+ }
+
PacketType type = readType(packetBuf);
- Packet packet = new Packet(type, head.getEngineIOVersion());
+ Packet packet = new Packet(type);
- if (type == PacketType.PING) {
+ if (type == PacketType.PING || type == PacketType.PONG) {
packet.setData(readString(packetBuf));
return packet;
}
@@ -364,43 +520,253 @@ private void parseHeader(ByteBuf frame, Packet packet, PacketType innerType) {
}
}
- private Packet addAttachment(ClientHead head, ByteBuf frame, Packet binaryPacket) throws IOException {
- ByteBuf attachBuf = Base64.encode(frame);
- binaryPacket.addAttachment(Unpooled.copiedBuffer(attachBuf));
- attachBuf.release();
- frame.skipBytes(frame.readableBytes());
+ /**
+ * 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)
+ *
+ *
+ * @param head the client connection head
+ * @param frame the incoming byte buffer 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
+ */
+ private Packet addAttachment(ClientHead head, ByteBuf frame, Packet binaryPacket, Transport transport) throws IOException {
+ EngineIOVersion version = head.getEngineIOVersion();
+ if (version == null) {
+ log.warn("addAttachment called with null engineIOVersion for session {}, treating as V4",
+ head.getSessionId());
+ version = EngineIOVersion.V4;
+ }
- if (binaryPacket.isAttachmentsLoaded()) {
- LinkedList slices = new LinkedList<>();
- ByteBuf source = binaryPacket.getDataSource();
- for (int i = 0; i < binaryPacket.getAttachments().size(); i++) {
- ByteBuf attachment = binaryPacket.getAttachments().get(i);
- ByteBuf scanValue = Unpooled.copiedBuffer("{\"_placeholder\":true,\"num\":" + i + "}", CharsetUtil.UTF_8);
- int pos = PacketEncoder.find(source, scanValue);
- if (pos == -1) {
- scanValue = Unpooled.copiedBuffer("{\"num\":" + i + ",\"_placeholder\":true}", CharsetUtil.UTF_8);
- pos = PacketEncoder.find(source, scanValue);
- if (pos == -1) {
- throw new IllegalStateException("Can't find attachment by index: " + i + " in packet source");
+ int ri = frame.readerIndex();
+ if (transport == Transport.POLLING) {
+ boolean wrapperFound = false;
+
+ // 1. EIOv2/v3 Polling binary payload wrapper: 0x01 + length + 0xFF + 0x04 + payload
+ if (frame.readableBytes() > 0 && frame.getByte(ri) == 1) {
+ frame.readByte(); // skip 0x01
+ int maxLength = Math.min(frame.readableBytes(), 10);
+ int headEndIndex = frame.bytesBefore(maxLength, (byte) -1);
+ if (headEndIndex > 0) {
+ for (int i = 0; i < headEndIndex; i++) {
+ byte b = frame.getByte(frame.readerIndex() + i);
+ if ((b < 0 || b > 9) && (b < '0' || b > '9')) {
+ throw new IOException("Malformed polling wrapper: non-digit character in length header");
+ }
+ }
+ long rawLen = readLegacyBinaryLength(frame, headEndIndex);
+ if (rawLen < 0 || rawLen > Integer.MAX_VALUE) {
+ throw new IOException("Malformed polling wrapper: length overflow " + rawLen);
+ }
+ int len = (int) rawLen;
+ int payloadStart = frame.readerIndex() + 1; // skip 0xFF separator
+ if (payloadStart + len > frame.writerIndex()) {
+ throw new IOException("Malformed polling wrapper: length " + len
+ + " exceeds remaining frame bytes " + (frame.writerIndex() - payloadStart));
+ }
+ ByteBuf payload = frame.slice(payloadStart, len);
+ frame.readerIndex(payloadStart + len);
+ wrapperFound = true;
+
+ // Strip leading 0x04 type prefix if present
+ int payloadRi = payload.readerIndex();
+ if (payload.readableBytes() >= 1 && payload.getByte(payloadRi) == 4) {
+ payload.readerIndex(payloadRi + 1);
}
+ ByteBuf attachBuf = Base64.encode(payload);
+ binaryPacket.addAttachment(Unpooled.copiedBuffer(attachBuf));
+ attachBuf.release();
+ } else {
+ throw new IOException("Malformed polling wrapper: missing or invalid 0xFF separator");
+ }
+ } else if (frame.readableBytes() >= 1 && frame.getByte(ri) == 'b') {
+ // 2. Polling Base64 text attachment: 'b4' (EIOv3) or 'b' (EIOv4)
+ // In EIOv4 multi-packet polling, attachments in the POST body are separated by 0x1e.
+ // Slice out the current attachment frame up to 0x1e so remaining attachments remain readable.
+ int sepPos = frame.bytesBefore((byte) 0x1E);
+ ByteBuf attachFrame;
+ if (sepPos >= 0) {
+ attachFrame = frame.readSlice(sepPos);
+ frame.skipBytes(1); // skip 0x1e record separator
+ wrapperFound = true; // reader index already advanced to next packet
+ } else {
+ attachFrame = frame;
}
- ByteBuf prefixBuf = source.slice(source.readerIndex(), pos - source.readerIndex());
- slices.add(prefixBuf);
- slices.add(quotes);
- slices.add(attachment);
- slices.add(quotes);
+ int attachRi = attachFrame.readerIndex();
+ if ((version == EngineIOVersion.V2 || version == EngineIOVersion.V3)
+ && attachFrame.readableBytes() >= 2
+ && attachFrame.getByte(attachRi) == 'b'
+ && attachFrame.getByte(attachRi + 1) == '4') {
+ attachFrame.readerIndex(attachRi + 2); // skip 'b4' (EIOv2/v3)
+ } else if (attachFrame.readableBytes() >= 1 && attachFrame.getByte(attachRi) == 'b') {
+ attachFrame.readerIndex(attachRi + 1); // skip 'b' (EIOv4)
+ }
+ // Already base64-encoded text payload
+ binaryPacket.addAttachment(Unpooled.copiedBuffer(attachFrame));
+ if (!wrapperFound) {
+ attachFrame.skipBytes(attachFrame.readableBytes());
+ }
+ } else {
+ // 3. Fallback polling binary payload
+ ByteBuf attachBuf = Base64.encode(frame);
+ binaryPacket.addAttachment(Unpooled.copiedBuffer(attachBuf));
+ attachBuf.release();
+ frame.skipBytes(frame.readableBytes());
+ }
+
+ if (!wrapperFound && frame.readableBytes() > 0) {
+ frame.skipBytes(frame.readableBytes());
+ }
- source.readerIndex(pos + scanValue.readableBytes());
+ } else {
+ // WebSocket transport
+ boolean isV3orV2WebSocket = (version == EngineIOVersion.V3 || version == EngineIOVersion.V2);
+ if (isV3orV2WebSocket
+ && frame.readableBytes() >= 1
+ && frame.getByte(ri) == 4) {
+ frame.readerIndex(ri + 1); // skip 0x04 type prefix for V2/V3
}
- slices.add(source.slice());
+ ByteBuf attachBuf = Base64.encode(frame);
+ binaryPacket.addAttachment(Unpooled.copiedBuffer(attachBuf));
+ attachBuf.release();
+ frame.skipBytes(frame.readableBytes());
+ }
+
+ return completeAttachment(head, binaryPacket);
+ }
- ByteBuf compositeBuf = Unpooled.wrappedBuffer(slices.toArray(new ByteBuf[0]));
+ private Packet addLegacyPollingBinaryAttachment(ClientHead head,
+ ByteBuf payload,
+ Packet binaryPacket) throws IOException {
+ if (payload.isReadable() && payload.getByte(payload.readerIndex()) == 4) {
+ payload.skipBytes(1);
+ }
+ ByteBuf attachment = Base64.encode(payload);
+ try {
+ binaryPacket.addAttachment(Unpooled.copiedBuffer(attachment));
+ } finally {
+ attachment.release();
+ }
+ return completeAttachment(head, binaryPacket);
+ }
+
+ private Packet completeAttachment(ClientHead head, Packet binaryPacket) throws IOException {
+ if (!binaryPacket.isAttachmentsLoaded()) {
+ return new Packet(PacketType.MESSAGE);
+ }
+
+ LinkedList slices = new LinkedList<>();
+ ByteBuf source = head.getLastBinaryPacketSource();
+ for (int i = 0; i < binaryPacket.getAttachments().size(); i++) {
+ ByteBuf attachment = binaryPacket.getAttachments().get(i);
+ ByteBuf scanValue = Unpooled.copiedBuffer("{\"_placeholder\":true,\"num\":" + i + "}", CharsetUtil.UTF_8);
+ int pos = PacketEncoder.find(source, scanValue);
+ if (pos == -1) {
+ scanValue = Unpooled.copiedBuffer("{\"num\":" + i + ",\"_placeholder\":true}", CharsetUtil.UTF_8);
+ pos = PacketEncoder.find(source, scanValue);
+ if (pos == -1) {
+ throw new IllegalStateException("Can't find attachment by index: " + i + " in packet source");
+ }
+ }
+
+ ByteBuf prefixBuf = source.slice(source.readerIndex(), pos - source.readerIndex());
+ slices.add(prefixBuf);
+ slices.add(quotes);
+ slices.add(attachment);
+ slices.add(quotes);
+
+ source.readerIndex(pos + scanValue.readableBytes());
+ }
+ slices.add(source.slice());
+
+ ByteBuf compositeBuf = Unpooled.wrappedBuffer(slices.toArray(new ByteBuf[0]));
+ try {
parseBody(head, compositeBuf, binaryPacket);
- head.setLastBinaryPacket(null);
- return binaryPacket;
+ } finally {
+ head.clearPendingBinaryPacket();
}
- return new Packet(PacketType.MESSAGE, head.getEngineIOVersion());
+ return binaryPacket;
}
private void parseBody(ClientHead head, ByteBuf frame, Packet packet) throws IOException {
@@ -409,6 +775,11 @@ private void parseBody(ClientHead head, ByteBuf frame, Packet packet) throws IOE
return;
}
+ if (packet.hasAttachments() && !packet.isAttachmentsLoaded()) {
+ handleBinaryAttachments(head, frame, packet);
+ return;
+ }
+
PacketType subType = packet.getSubType();
// Handle different packet subtypes
@@ -428,6 +799,10 @@ private void parseBody(ClientHead head, ByteBuf frame, Packet packet) throws IOE
parseEventBody(frame, packet);
break;
+ case ERROR:
+ parseErrorBody(frame, packet);
+ break;
+
default:
// Handle binary attachments for other packet types
handleBinaryAttachments(head, frame, packet);
@@ -435,6 +810,28 @@ private void parseBody(ClientHead head, ByteBuf frame, Packet packet) throws IOE
}
}
+ /**
+ * Parse ERROR packet bodies
+ */
+ private void parseErrorBody(ByteBuf frame, Packet packet) throws IOException {
+ String nsp = readNamespace(frame, false);
+ if (nsp != null && !nsp.isEmpty()) {
+ packet.setNsp(nsp);
+ }
+
+ if (frame.readableBytes() > 0) {
+ try {
+ frame.markReaderIndex();
+ try (ByteBufInputStream in = new ByteBufInputStream(frame)) {
+ Object errorData = jsonSupport.readValue(packet.getNsp(), in, Object.class);
+ packet.setData(errorData);
+ }
+ } catch (Exception e) {
+ frame.resetReaderIndex();
+ packet.setData(readString(frame));
+ }
+ }
+ }
/**
* Parse CONNECT and DISCONNECT packet bodies
*/
@@ -478,9 +875,8 @@ private void parseEventBody(ByteBuf frame, Packet packet) throws IOException {
*/
private void handleBinaryAttachments(ClientHead head, ByteBuf frame, Packet packet) {
if (packet.hasAttachments() && !packet.isAttachmentsLoaded()) {
- packet.setDataSource(Unpooled.copiedBuffer(frame));
+ head.setPendingBinaryPacket(packet, Unpooled.copiedBuffer(frame));
frame.skipBytes(frame.readableBytes());
- head.setLastBinaryPacket(packet);
}
}
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 b29be4f9..430da44a 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
@@ -18,10 +18,12 @@
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Queue;
import com.socketio4j.socketio.Configuration;
+import com.socketio4j.socketio.annotation.Internal;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
@@ -31,6 +33,7 @@
import io.netty.handler.codec.base64.Base64Dialect;
import io.netty.util.CharsetUtil;
+@Internal
public class PacketEncoder {
private static final byte[] BINARY_HEADER = "b4".getBytes(CharsetUtil.UTF_8);
@@ -59,50 +62,88 @@ public ByteBuf allocateBuffer(ByteBufAllocator allocator) {
return allocator.heapBuffer();
}
- public void encodeJsonP(Integer jsonpIndex, Queue packets, ByteBuf out, ByteBufAllocator allocator, int limit) throws IOException {
- boolean jsonpMode = jsonpIndex != null;
+ /**
+ * Encodes Engine.IO polling responses using Base64 text encoding.
+ *
+ * If {@code jsonpIndex != null}, the encoded payload is additionally wrapped
+ * in a JSONP callback for legacy clients.
+ *
+ * Supports the following Engine.IO polling modes:
+ *
+ * - Base64 polling ({@code b64=1})
+ * - JSONP polling ({@code j=}), which uses the same Base64
+ * payload encoding wrapped in a JSONP callback.
+ *
+ *
+ * @param engineIOVersion Engine.IO protocol version.
+ * @param jsonpIndex JSONP callback index, or {@code null} for standard Base64
+ * polling.
+ * @param packets packets to encode.
+ * @param out destination buffer.
+ * @param allocator buffer allocator.
+ * @param limit maximum number of packets to encode.
+ * @throws IOException if packet encoding fails.
+ */
+ public void encodeJsonP(EngineIOVersion engineIOVersion,
+ Integer jsonpIndex,
+ Queue packets,
+ ByteBuf out,
+ ByteBufAllocator allocator,
+ int limit) throws IOException {
+
+ boolean wrapJsonp = jsonpIndex != null;
ByteBuf buf = allocateBuffer(allocator);
+ try {
+ int i = 0;
- int i = 0;
- while (true) {
- Packet packet = packets.poll();
- if (packet == null || i == limit) {
- break;
- }
-
- ByteBuf packetBuf = allocateBuffer(allocator);
- encodePacket(packet, packetBuf, allocator, true);
-
- int packetSize = packetBuf.writerIndex();
- buf.writeBytes(toChars(packetSize));
- buf.writeBytes(B64_DELIMITER);
- buf.writeBytes(packetBuf);
-
- packetBuf.release();
+ while (true) {
+ Packet packet = packets.poll();
+ if (packet == null || i == limit) {
+ break;
+ }
- i++;
+ ByteBuf packetBuf = allocateBuffer(allocator);
+ try {
+ EncodeResult encodeResult =
+ encodePacket(engineIOVersion, packet, packetBuf, allocator, true);
+
+ int packetSize = packetBuf.writerIndex();
+ buf.writeBytes(toChars(packetSize));
+ buf.writeBytes(B64_DELIMITER);
+ buf.writeBytes(packetBuf);
+
+ for (ByteBuf attachment : encodeResult.getAttachments()) {
+ ByteBuf encodedBuf = Base64.encode(attachment, Base64Dialect.STANDARD);
+ try {
+ buf.writeBytes(toChars(encodedBuf.readableBytes() + 2));
+ buf.writeBytes(B64_DELIMITER);
+ buf.writeBytes(BINARY_HEADER);
+ buf.writeBytes(encodedBuf);
+ } finally {
+ encodedBuf.release();
+ }
+ }
+ } finally {
+ packetBuf.release();
+ }
- for (ByteBuf attachment : packet.getAttachments()) {
- ByteBuf encodedBuf = Base64.encode(attachment, Base64Dialect.URL_SAFE);
- buf.writeBytes(toChars(encodedBuf.readableBytes() + 2));
- buf.writeBytes(B64_DELIMITER);
- buf.writeBytes(BINARY_HEADER);
- buf.writeBytes(encodedBuf);
+ i++;
}
- }
- if (jsonpMode) {
- out.writeBytes(JSONP_HEAD);
- out.writeBytes(toChars(jsonpIndex));
- out.writeBytes(JSONP_START);
- }
+ if (wrapJsonp) {
+ out.writeBytes(JSONP_HEAD);
+ out.writeBytes(toChars(jsonpIndex));
+ out.writeBytes(JSONP_START);
+ }
- processUtf8(buf, out, jsonpMode);
- buf.release();
+ processUtf8(buf, out, wrapJsonp);
- if (jsonpMode) {
- out.writeBytes(JSONP_END);
+ if (wrapJsonp) {
+ out.writeBytes(JSONP_END);
+ }
+ } finally {
+ buf.release();
}
}
@@ -121,34 +162,143 @@ private void processUtf8(ByteBuf in, ByteBuf out, boolean jsonpMode) {
}
}
- public void encodePackets(Queue packets, ByteBuf buffer, ByteBufAllocator allocator, int limit) throws IOException {
- int i = 0;
- boolean hasPrecedingPacket = false;
- while (true) {
- Packet packet = packets.poll();
- if (packet == null || i == limit) {
- break;
+ public EncodePacketsResult encodePackets(EngineIOVersion engineIOVersion,
+ Queue packets,
+ ByteBuf buffer,
+ ByteBufAllocator allocator,
+ int limit) throws IOException {
+
+ int count = 0;
+ boolean first = true;
+ boolean hasBinary = false;
+
+ if (EngineIOVersion.V4.equals(engineIOVersion)) {
+
+ while (count < limit) {
+ Packet packet = packets.poll();
+ if (packet == null) {
+ break;
+ }
+
+ if (!first) {
+ buffer.writeByte(0x1E);
+ }
+
+ EncodeResult result =
+ encodePacket(engineIOVersion, packet, buffer, allocator, false);
+
+ hasBinary |= result.hasAttachments();
+
+ for (ByteBuf attachment : result.getAttachments()) {
+ buffer.writeByte(0x1E);
+ buffer.writeByte('b');
+
+ ByteBuf encoded = Base64.encode(attachment, Base64Dialect.STANDARD);
+ try {
+ buffer.writeBytes(encoded);
+ } finally {
+ encoded.release();
+ }
+ }
+
+ first = false;
+ count++;
}
- // Multiple packets are separated by 0x1e from protocol version 3 on
- // see https://socket.io/docs/v4/socket-io-protocol/#sample-session
- final boolean isV3OrNewer = EngineIOVersion.V4.equals(packet.getEngineIOVersion())
- || EngineIOVersion.V3.equals(packet.getEngineIOVersion());
- if (hasPrecedingPacket && isV3OrNewer) {
- buffer.writeByte(0x1e);
+
+ return new EncodePacketsResult(hasBinary);
+ }
+
+ if (EngineIOVersion.V2.equals(engineIOVersion)
+ || EngineIOVersion.V3.equals(engineIOVersion)) {
+
+ class EncodedPacket {
+ final ByteBuf packet;
+ final EncodeResult result;
+
+ EncodedPacket(ByteBuf packet, EncodeResult result) {
+ this.packet = packet;
+ this.result = result;
+ }
}
- encodePacket(packet, buffer, allocator, false);
- i++;
+ List encodedPackets = new ArrayList<>();
+
+ try {
+
+ //
+ // First pass - encode everything once
+ //
+ while (count < limit) {
+
+ Packet packet = packets.poll();
+ if (packet == null) {
+ break;
+ }
+
+ ByteBuf packetBuf = allocator.buffer();
+
+ EncodeResult result =
+ encodePacket(engineIOVersion,
+ packet,
+ packetBuf,
+ allocator,
+ false);
- for (ByteBuf attachment : packet.getAttachments()) {
- buffer.writeByte(1);
- buffer.writeBytes(longToBytes(attachment.readableBytes() + 1));
- buffer.writeByte(0xff);
- buffer.writeByte(4);
- buffer.writeBytes(attachment);
+ hasBinary |= result.hasAttachments();
+
+ encodedPackets.add(new EncodedPacket(packetBuf, result));
+
+ count++;
+ }
+
+ //
+ // Second pass - write using the chosen framing
+ //
+ for (EncodedPacket encoded : encodedPackets) {
+
+ if (hasBinary) {
+
+ // Binary Engine.IO payload
+ buffer.writeByte(0);
+ buffer.writeBytes(longToBytes(encoded.packet.readableBytes()));
+ buffer.writeByte(0xFF);
+
+ } else {
+
+ // Text Engine.IO payload
+ int chars =
+ encoded.packet.toString(CharsetUtil.UTF_8).length();
+
+ buffer.writeCharSequence(
+ Integer.toString(chars),
+ CharsetUtil.US_ASCII);
+
+ buffer.writeByte(':');
+ }
+
+ buffer.writeBytes(encoded.packet);
+
+ for (ByteBuf attachment : encoded.result.getAttachments()) {
+ buffer.writeByte(1);
+ buffer.writeBytes(longToBytes(attachment.readableBytes() + 1));
+ buffer.writeByte(0xFF);
+ buffer.writeByte(4);
+ buffer.writeBytes(attachment);
+ }
+ }
+
+ } finally {
+
+ for (EncodedPacket encoded : encodedPackets) {
+ encoded.packet.release();
+ }
}
- hasPrecedingPacket = true;
+
+ return new EncodePacketsResult(hasBinary);
}
+
+ throw new IllegalStateException(
+ "Unsupported Engine.IO version: " + engineIOVersion);
}
private byte toChar(int number) {
@@ -257,21 +407,25 @@ public static byte[] longToBytes(long number) {
return res;
}
- public void encodePacket(Packet packet, ByteBuf buffer, ByteBufAllocator allocator, boolean binary) throws IOException {
- ByteBuf buf = buffer;
- if (!binary) {
+ public EncodeResult encodePacket(EngineIOVersion version, Packet packet, ByteBuf buffer,
+ ByteBufAllocator allocator,
+ boolean binary) throws IOException {
+
+ ByteBuf buf;
+ if (binary) {
+ buf = buffer;
+ } else {
buf = allocateBuffer(allocator);
}
- byte type = toChar(packet.getType().getValue());
- buf.writeByte(type);
+ List attachments = Collections.emptyList();
+ buf.writeByte(toChar(packet.getType().getValue()));
try {
switch (packet.getType()) {
- case PONG: {
+ case PONG:
buf.writeBytes(packet.getData().toString().getBytes(CharsetUtil.UTF_8));
break;
- }
case OPEN: {
ByteBufOutputStream out = new ByteBufOutputStream(buf);
@@ -282,66 +436,69 @@ public void encodePacket(Packet packet, ByteBuf buffer, ByteBufAllocator allocat
case MESSAGE: {
ByteBuf encBuf = null;
+ PacketType subType = packet.getSubType();
- if (packet.getSubType() == PacketType.ERROR) {
+ if (subType == PacketType.ERROR) {
encBuf = allocateBuffer(allocator);
-
ByteBufOutputStream out = new ByteBufOutputStream(encBuf);
jsonSupport.writeValue(out, packet.getData());
}
- if (packet.getSubType() == PacketType.EVENT
- || packet.getSubType() == PacketType.ACK) {
+ if (subType == PacketType.EVENT || subType == PacketType.ACK) {
- List