merge

README.md

@@ -102,7 +102,7 @@ Feature - Compression support
 Fixed - DotNET client request handling  
 Fixed - Packet length format parsing  
 Fixed - skipping 'd=' in packet  
-Fixed - Polling clients sporatically get prematurely disconnected (thanks to lpage30)  
+Fixed - Polling clients sporadically get prematurely disconnected (thanks to lpage30)  
 Fixed - connections stay open forever if server sent `close` packet  
 Fixed - compatibility with Redisson latest version  
 

pom.xml

@@ -85,37 +85,37 @@
   <dependency>
   	<groupId>io.netty</groupId>
   	<artifactId>netty-buffer</artifactId>
-  	<version>4.1.32.Final</version>
+  	<version>4.1.39.Final</version>
   </dependency>
   <dependency>
   	<groupId>io.netty</groupId>
   	<artifactId>netty-common</artifactId>
-  	<version>4.1.32.Final</version>
+  	<version>4.1.39.Final</version>
   </dependency>
   <dependency>
   	<groupId>io.netty</groupId>
   	<artifactId>netty-transport</artifactId>
-       <version>4.1.32.Final</version>
+       <version>4.1.39.Final</version>
   </dependency>
   <dependency>
   	<groupId>io.netty</groupId>
   	<artifactId>netty-handler</artifactId>
-  	<version>4.1.32.Final</version>
+  	<version>4.1.39.Final</version>
   </dependency>
   <dependency>
   	<groupId>io.netty</groupId>
   	<artifactId>netty-codec-http</artifactId>
-  	<version>4.1.32.Final</version>
+  	<version>4.1.39.Final</version>
   </dependency>
   <dependency>
   	<groupId>io.netty</groupId>
   	<artifactId>netty-codec</artifactId>
-  	<version>4.1.32.Final</version>
+  	<version>4.1.39.Final</version>
   </dependency>
   <dependency>
         <groupId>io.netty</groupId>
         <artifactId>netty-transport-native-epoll</artifactId>
-        <version>4.1.32.Final</version>
+        <version>4.1.39.Final</version>
   </dependency>
 
   <dependency>
@@ -140,12 +140,12 @@
   <dependency>
   	<groupId>com.fasterxml.jackson.core</groupId>
   	<artifactId>jackson-core</artifactId>
-  	<version>2.9.7</version>
+  	<version>2.9.9</version>
   </dependency>
   <dependency>
   	<groupId>com.fasterxml.jackson.core</groupId>
   	<artifactId>jackson-databind</artifactId>
-  	<version>2.9.7</version>
+  	<version>2.9.9.2</version>
   </dependency>
 
 
@@ -172,7 +172,7 @@
     <dependency>
         <groupId>org.redisson</groupId>
         <artifactId>redisson</artifactId>
-        <version>3.10.0</version>
+        <version>3.11.3</version>
         <scope>provided</scope>
     </dependency>
     <dependency>
@@ -294,7 +294,7 @@
         <plugin>
            <groupId>org.apache.felix</groupId>
            <artifactId>maven-bundle-plugin</artifactId>
-           <version>3.5.0</version>
+           <version>4.2.1</version>
            <extensions>true</extensions>
            <configuration>
                <instructions>

src/main/java/com/corundumstudio/socketio/SocketIOServer.java

@@ -229,6 +229,13 @@ public <T> void addEventListener(String eventName, Class<T> eventClass, DataList
         mainNamespace.addEventListener(eventName, eventClass, listener);
     }
 
+    @Override
+    public void addEventInterceptor(EventInterceptor eventInterceptor) {
+        mainNamespace.addEventInterceptor(eventInterceptor);
+
+    }
+
+
     @Override
     public void removeAllListeners(String eventName) {
         mainNamespace.removeAllListeners(eventName);

src/main/java/com/corundumstudio/socketio/listener/ClientListeners.java

@@ -15,12 +15,15 @@
  */
 package com.corundumstudio.socketio.listener;
 
+
 public interface ClientListeners {
 
     void addMultiTypeEventListener(String eventName, MultiTypeEventListener listener, Class<?> ... eventClass);
 
     <T> void addEventListener(String eventName, Class<T> eventClass, DataListener<T> listener);
 
+    void addEventInterceptor(EventInterceptor eventInterceptor);
+
     void addDisconnectListener(DisconnectListener listener);
 
     void addConnectListener(ConnectListener listener);

src/main/java/com/corundumstudio/socketio/listener/EventInterceptor.java

@@ -0,0 +1,9 @@
+package com.corundumstudio.socketio.listener;
+
+import com.corundumstudio.socketio.AckRequest;
+import com.corundumstudio.socketio.transport.NamespaceClient;
+import java.util.List;
+
+public interface  EventInterceptor {
+    void onEvent(NamespaceClient client, String eventName, List<Object> args, AckRequest ackRequest);
+}

src/main/java/com/corundumstudio/socketio/misc/CompositeIterable.java

@@ -19,14 +19,11 @@
 import java.util.Iterator;
 import java.util.List;
 
-public class CompositeIterable<T> implements Iterable<T>, Iterator<T> {
+public class CompositeIterable<T> implements Iterable<T> {
 
     private List<Iterable<T>> iterablesList;
     private Iterable<T>[] iterables;
 
-    private Iterator<Iterator<T>> listIterator;
-    private Iterator<T> currentIterator;
-
     public CompositeIterable(List<Iterable<T>> iterables) {
         this.iterablesList = iterables;
     }
@@ -52,35 +49,8 @@ public Iterator<T> iterator() {
                 iterators.add(iterable.iterator());
             }
         }
-        listIterator = iterators.iterator();
-        currentIterator = null;
-        return this;
-    }
-
-    @Override
-    public boolean hasNext() {
-        if (currentIterator == null || !currentIterator.hasNext()) {
-            while (listIterator.hasNext()) {
-                Iterator<T> iterator = listIterator.next();
-                if (iterator.hasNext()) {
-                    currentIterator = iterator;
-                    return true;
-                }
-            }
-            return false;
-        }
-        return currentIterator.hasNext();
+        return new CompositeIterator<T>(iterators.iterator());
     }
 
-    @Override
-    public T next() {
-        hasNext();
-        return currentIterator.next();
-    }
-
-    @Override
-    public void remove() {
-        currentIterator.remove();
-    }
 
 }

src/main/java/com/corundumstudio/socketio/misc/CompositeIterator.java

@@ -0,0 +1,55 @@
+/**
+ * Copyright (c) 2012-2019 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.corundumstudio.socketio.misc;
+
+import java.util.Iterator;
+
+public class CompositeIterator<T> implements Iterator<T> {
+
+    private Iterator<Iterator<T>> listIterator;
+    private Iterator<T> currentIterator;
+
+    public CompositeIterator(Iterator<Iterator<T>> listIterator) {
+        this.currentIterator = null;
+        this.listIterator = listIterator;
+    }
+
+    @Override
+    public boolean hasNext() {
+        if (currentIterator == null || !currentIterator.hasNext()) {
+            while (listIterator.hasNext()) {
+                Iterator<T> iterator = listIterator.next();
+                if (iterator.hasNext()) {
+                    currentIterator = iterator;
+                    return true;
+                }
+            }
+            return false;
+        }
+        return currentIterator.hasNext();
+    }
+
+    @Override
+    public T next() {
+        hasNext();
+        return currentIterator.next();
+    }
+
+    @Override
+    public void remove() {
+        currentIterator.remove();
+    }
+}

src/main/java/com/corundumstudio/socketio/namespace/Namespace.java

@@ -59,6 +59,7 @@ public class Namespace implements SocketIONamespace {
     private final Queue<ConnectListener> connectListeners = new ConcurrentLinkedQueue<ConnectListener>();
     private final Queue<DisconnectListener> disconnectListeners = new ConcurrentLinkedQueue<DisconnectListener>();
     private final Queue<PingListener> pingListeners = new ConcurrentLinkedQueue<PingListener>();
+    private final Queue<EventInterceptor> eventInterceptors = new ConcurrentLinkedQueue<EventInterceptor>();
 
     private final Map<UUID, SocketIOClient> allClients = PlatformDependent.newConcurrentHashMap();
     private final ConcurrentMap<String, Set<UUID>> roomClients = PlatformDependent.newConcurrentHashMap();
@@ -126,6 +127,11 @@ public <T> void addEventListener(String eventName, Class<T> eventClass, DataList
         jsonSupport.addEventMapping(name, eventName, eventClass);
     }
 
+    @Override
+    public void addEventInterceptor(EventInterceptor eventInterceptor) {
+        eventInterceptors.add(eventInterceptor);
+    }
+
     @SuppressWarnings({"rawtypes", "unchecked"})
     public void onEvent(NamespaceClient client, String eventName, List<Object> args, AckRequest ackRequest) {
         EventEntry entry = eventListeners.get(eventName);
@@ -139,6 +145,10 @@ public void onEvent(NamespaceClient client, String eventName, List<Object> args,
                 Object data = getEventData(args, dataListener);
                 dataListener.onData(client, data, ackRequest);
             }
+
+            for (EventInterceptor eventInterceptor : eventInterceptors) {
+                eventInterceptor.onEvent(client, eventName, args, ackRequest);
+            }
         } catch (Exception e) {
             exceptionListener.onEventException(e, args, client);
             if (ackMode == AckMode.AUTO_SUCCESS_ONLY) {

src/main/java/com/corundumstudio/socketio/protocol/Event.java

@@ -17,7 +17,7 @@
 
 import java.util.List;
 
-class Event {
+public class Event {
 
     private String name;
     private List<Object> args;

src/main/java/com/corundumstudio/socketio/protocol/Packet.java

@@ -77,6 +77,28 @@ public <T> T getData() {
         return (T)data;
     }
 
+    /**
+     * Creates a copy of #{@link Packet} with new namespace set
+     * if it differs from current namespace.
+     * Otherwise, returns original object unchanged
+     */
+    public Packet withNsp(String namespace) {
+        if (this.nsp.equalsIgnoreCase(namespace)) {
+            return this;
+        } else {
+            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);
+            newPacket.attachments = this.attachments;
+            newPacket.attachmentsCount = this.attachmentsCount;
+            return newPacket;
+        }
+    }
+
     public void setNsp(String endpoint) {
         this.nsp = endpoint;
     }

src/main/java/com/corundumstudio/socketio/transport/NamespaceClient.java

@@ -104,8 +104,8 @@ public void send(Packet packet) {
         if (!isConnected()) {
             return;
         }
-        packet.setNsp(namespace.getName());
-        baseClient.send(packet);
+
+        baseClient.send(packet.withNsp(namespace.getName()));
     }
 
     public void onDisconnect() {

src/test/java/com/corundumstudio/socketio/protocol/PacketTest.java

@@ -0,0 +1,76 @@
+/**
+ * Copyright (c) 2012-2019 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.corundumstudio.socketio.protocol;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+
+import io.netty.buffer.Unpooled;
+import org.junit.Test;
+
+public class PacketTest {
+
+    @Test
+    public void packetCopyIsCreatedWhenNamespaceDiffers() {
+        Packet oldPacket = createPacket();
+
+        String newNs = "new";
+        Packet newPacket = oldPacket.withNsp(newNs);
+        assertEquals(newNs, newPacket.getNsp());
+        assertPacketCopied(oldPacket, newPacket);
+    }
+
+    @Test
+    public void packetCopyIsCreatedWhenNewNamespaceDiffersAndIsNull() {
+        Packet packet = createPacket();
+        Packet newPacket = packet.withNsp(null);
+        assertNull(newPacket.getNsp());
+        assertPacketCopied(packet, newPacket);
+    }
+
+    @Test
+    public void originalPacketReturnedIfNamespaceIsTheSame() {
+        Packet packet = new Packet(PacketType.MESSAGE);
+        assertSame(packet, packet.withNsp(""));
+    }
+
+    private void assertPacketCopied(Packet oldPacket, Packet newPacket) {
+        assertNotSame(newPacket, oldPacket);
+        assertEquals(oldPacket.getName(), newPacket.getName());
+        assertEquals(oldPacket.getType(), newPacket.getType());
+        assertEquals(oldPacket.getSubType(), newPacket.getSubType());
+        assertEquals(oldPacket.getAckId(), newPacket.getAckId());
+        assertEquals(oldPacket.getAttachments().size(), newPacket.getAttachments().size());
+        assertSame(oldPacket.getAttachments(), newPacket.getAttachments());
+        assertEquals(oldPacket.getData(), newPacket.getData());
+        assertSame(oldPacket.getDataSource(), newPacket.getDataSource());
+    }
+
+    private Packet createPacket() {
+        Packet packet = new Packet(PacketType.MESSAGE);
+        packet.setSubType(PacketType.EVENT);
+        packet.setName("packetName");
+        packet.setData("data");
+        packet.setAckId(1L);
+        packet.setNsp("old");
+        packet.setDataSource(Unpooled.wrappedBuffer(new byte[]{10}));
+        packet.initAttachments(1);
+        packet.addAttachment(Unpooled.wrappedBuffer(new byte[]{20}));
+        return packet;
+    }
+}