ExceptionListener added. #53

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

@@ -20,6 +20,8 @@ import io.netty.handler.codec.TooLongFrameException;
 import java.io.InputStream;
 
 import com.corundumstudio.socketio.handler.SuccessAuthorizationListener;
+import com.corundumstudio.socketio.listener.DefaultExceptionListener;
+import com.corundumstudio.socketio.listener.ExceptionListener;
 import com.corundumstudio.socketio.parser.JacksonJsonSupport;
 import com.corundumstudio.socketio.parser.JsonSupport;
 import com.corundumstudio.socketio.store.MemoryStoreFactory;
@@ -27,6 +29,8 @@ import com.corundumstudio.socketio.store.StoreFactory;
 
 public class Configuration {
 
+    private ExceptionListener exceptionListener = new DefaultExceptionListener();
+
     private String jsonTypeFieldName = "@class";
     private String context = "/socket.io";
 
@@ -383,5 +387,19 @@ public class Configuration {
         return authorizationListener;
     }
 
+    /**
+     * Exception listener invoked on any exception in
+     * SocketIO listener
+     *
+     * @param exceptionListener
+     *
+     * @see com.corundumstudio.socketio.listener.ExceptionListener
+     */
+    public void setExceptionListener(ExceptionListener exceptionListener) {
+        this.exceptionListener = exceptionListener;
+    }
+    public ExceptionListener getExceptionListener() {
+        return exceptionListener;
+    }
 
 }

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

@@ -31,6 +31,7 @@ import com.corundumstudio.socketio.listener.ClientListeners;
 import com.corundumstudio.socketio.listener.ConnectListener;
 import com.corundumstudio.socketio.listener.DataListener;
 import com.corundumstudio.socketio.listener.DisconnectListener;
+import com.corundumstudio.socketio.listener.ExceptionListener;
 import com.corundumstudio.socketio.listener.MultiTypeEventListener;
 import com.corundumstudio.socketio.namespace.Namespace;
 import com.corundumstudio.socketio.namespace.NamespacesHub;
@@ -57,7 +58,7 @@ public class SocketIOServer implements ClientListeners {
     public SocketIOServer(Configuration configuration) {
         this.configuration = configuration;
         this.configCopy = new Configuration(configuration);
-        namespacesHub = new NamespacesHub(configCopy.getJsonSupport(), configCopy.getStoreFactory());
+        namespacesHub = new NamespacesHub(configCopy.getJsonSupport(), configCopy.getStoreFactory(), configCopy.getExceptionListener());
         mainNamespace = addNamespace(Namespace.DEFAULT_NAME);
     }
 

src/main/java/com/corundumstudio/socketio/handler/PacketListener.java

@@ -90,10 +90,6 @@ public class PacketListener {
         default:
             break;
         }
-
-        // send ack response if it not executed
-        // during {@link DataListener#onData} invocation
-        ackRequest.sendAckData(Collections.emptyList());
     }
 
 }

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

@@ -0,0 +1,52 @@
+/**
+ * Copyright 2012 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.listener;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.corundumstudio.socketio.SocketIOClient;
+
+public class DefaultExceptionListener extends ExceptionListenerAdapter {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public void onEventException(Exception e, SocketIOClient client) {
+        log.error(e.getMessage(), e);
+    }
+
+    @Override
+    public void onDisconnectException(Exception e, SocketIOClient client) {
+        log.error(e.getMessage(), e);
+    }
+
+    @Override
+    public void onConnectException(Exception e, SocketIOClient client) {
+        log.error(e.getMessage(), e);
+    }
+
+    @Override
+    public void onMessageException(Exception e, SocketIOClient client) {
+        log.error(e.getMessage(), e);
+    }
+
+    @Override
+    public void onJsonException(Exception e, SocketIOClient client) {
+        log.error(e.getMessage(), e);
+    }
+
+}

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

@@ -0,0 +1,32 @@
+/**
+ * Copyright 2012 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.listener;
+
+import com.corundumstudio.socketio.SocketIOClient;
+
+public interface ExceptionListener {
+
+    void onEventException(Exception e, SocketIOClient client);
+
+    void onDisconnectException(Exception e, SocketIOClient client);
+
+    void onConnectException(Exception e, SocketIOClient client);
+
+    void onMessageException(Exception e, SocketIOClient client);
+
+    void onJsonException(Exception e, SocketIOClient client);
+
+}

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

@@ -0,0 +1,42 @@
+/**
+ * Copyright 2012 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.listener;
+
+import com.corundumstudio.socketio.SocketIOClient;
+
+public class ExceptionListenerAdapter implements ExceptionListener {
+
+    @Override
+    public void onEventException(Exception e, SocketIOClient client) {
+    }
+
+    @Override
+    public void onDisconnectException(Exception e, SocketIOClient client) {
+    }
+
+    @Override
+    public void onConnectException(Exception e, SocketIOClient client) {
+    }
+
+    @Override
+    public void onMessageException(Exception e, SocketIOClient client) {
+    }
+
+    @Override
+    public void onJsonException(Exception e, SocketIOClient client) {
+    }
+
+}

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

@@ -26,9 +26,6 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ConcurrentMap;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.corundumstudio.socketio.AckRequest;
 import com.corundumstudio.socketio.BroadcastOperations;
 import com.corundumstudio.socketio.MultiTypeArgs;
@@ -38,6 +35,7 @@ import com.corundumstudio.socketio.annotation.ScannerEngine;
 import com.corundumstudio.socketio.listener.ConnectListener;
 import com.corundumstudio.socketio.listener.DataListener;
 import com.corundumstudio.socketio.listener.DisconnectListener;
+import com.corundumstudio.socketio.listener.ExceptionListener;
 import com.corundumstudio.socketio.listener.MultiTypeEventListener;
 import com.corundumstudio.socketio.parser.JsonSupport;
 import com.corundumstudio.socketio.parser.Packet;
@@ -54,8 +52,6 @@ import com.corundumstudio.socketio.transport.NamespaceClient;
  */
 public class Namespace implements SocketIONamespace {
 
-    private final Logger log = LoggerFactory.getLogger(getClass());
-
     public static final String DEFAULT_NAME = "";
 
     private final ScannerEngine engine = new ScannerEngine();
@@ -74,12 +70,14 @@ public class Namespace implements SocketIONamespace {
     private final String name;
     private final JsonSupport jsonSupport;
     private final StoreFactory storeFactory;
+    private final ExceptionListener exceptionListener;
 
-    public Namespace(String name, JsonSupport jsonSupport, StoreFactory storeFactory) {
+    public Namespace(String name, JsonSupport jsonSupport, StoreFactory storeFactory, ExceptionListener exceptionListener) {
         super();
         this.name = name;
         this.jsonSupport = jsonSupport;
         this.storeFactory = storeFactory;
+        this.exceptionListener = exceptionListener;
     }
 
     public void addClient(SocketIOClient client) {
@@ -140,25 +138,46 @@ public class Namespace implements SocketIONamespace {
         if (entry == null) {
             return;
         }
+
+        try {
             Queue<DataListener> listeners = entry.getListeners();
             for (DataListener dataListener : listeners) {
-            Object data = null;
+                Object data = getEventData(args, dataListener);
+                dataListener.onData(client, data, ackRequest);
+            }
+        } catch (Exception e) {
+            exceptionListener.onEventException(e, client);
+            return;
+        }
+        // send ack response if it not executed
+        // during {@link DataListener#onData} invocation
+        ackRequest.sendAckData(Collections.emptyList());
+    }
 
+    private Object getEventData(List<Object> args, DataListener dataListener) {
         if (dataListener instanceof MultiTypeEventListener) {
-                data = new MultiTypeArgs(args);
+            return new MultiTypeArgs(args);
         } else {
             if (!args.isEmpty()) {
-                    data = args.get(0);
+                return args.get(0);
             }
         }
-            dataListener.onData(client, data, ackRequest);
-        }
+        return null;
     }
 
     public void onMessage(NamespaceClient client, String data, AckRequest ackRequest) {
+        try {
             for (DataListener<String> listener : messageListeners) {
                 listener.onData(client, data, ackRequest);
             }
+        } catch (Exception e) {
+            exceptionListener.onMessageException(e, client);
+            return;
+        }
+
+        // send ack response if it not executed
+        // during {@link DataListener#onData} invocation
+        ackRequest.sendAckData(Collections.emptyList());
     }
 
     @SuppressWarnings({"rawtypes", "unchecked"})
@@ -167,9 +186,19 @@ public class Namespace implements SocketIONamespace {
         if (queue == null) {
             return;
         }
+
+        try {
             for (DataListener dataListener : queue) {
                 dataListener.onData(client, data, ackRequest);
             }
+        } catch (Exception e) {
+            exceptionListener.onJsonException(e, client);
+            return;
+        }
+
+        // send ack response if it not executed
+        // during {@link DataListener#onData} invocation
+        ackRequest.sendAckData(Collections.emptyList());
     }
 
     @Override
@@ -183,12 +212,12 @@ public class Namespace implements SocketIONamespace {
         leave(getName(), client.getSessionId());
         storeFactory.pubSubStore().publish(PubSubStore.LEAVE, new JoinLeaveMessage(client.getSessionId(), getName(), getName()));
 
-        for (DisconnectListener listener : disconnectListeners) {
         try {
+            for (DisconnectListener listener : disconnectListeners) {
                 listener.onDisconnect(client);
-            } catch (Exception e) {
-                log.error("Can't execute onDisconnect listener", e);
             }
+        } catch (Exception e) {
+            exceptionListener.onDisconnectException(e, client);
         }
     }
 
@@ -198,16 +227,16 @@ public class Namespace implements SocketIONamespace {
     }
 
     public void onConnect(SocketIOClient client) {
-        for (ConnectListener listener : connectListeners) {
+        join(getName(), client.getSessionId());
+        storeFactory.pubSubStore().publish(PubSubStore.JOIN, new JoinLeaveMessage(client.getSessionId(), getName(), getName()));
+
         try {
+            for (ConnectListener listener : connectListeners) {
                 listener.onConnect(client);
-            } catch (Exception e) {
-                log.error("Can't execute onConnect listener", e);
             }
+        } catch (Exception e) {
+            exceptionListener.onConnectException(e, client);
         }
-
-        join(getName(), client.getSessionId());
-        storeFactory.pubSubStore().publish(PubSubStore.JOIN, new JoinLeaveMessage(client.getSessionId(), getName(), getName()));
     }
 
     @Override

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

@@ -21,6 +21,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
 import com.corundumstudio.socketio.SocketIOClient;
+import com.corundumstudio.socketio.listener.ExceptionListener;
 import com.corundumstudio.socketio.misc.CompositeIterable;
 import com.corundumstudio.socketio.parser.JsonSupport;
 import com.corundumstudio.socketio.store.StoreFactory;
@@ -30,16 +31,18 @@ public class NamespacesHub {
     private final ConcurrentMap<String, Namespace> namespaces = new ConcurrentHashMap<String, Namespace>();
     private final JsonSupport jsonSupport;
     private final StoreFactory storeFactory;
+    private final ExceptionListener exceptionListener;
 
-    public NamespacesHub(JsonSupport jsonSupport, StoreFactory storeFactory) {
+    public NamespacesHub(JsonSupport jsonSupport, StoreFactory storeFactory, ExceptionListener exceptionListener) {
         this.jsonSupport = jsonSupport;
         this.storeFactory = storeFactory;
+        this.exceptionListener = exceptionListener;
     }
 
     public Namespace create(String name) {
         Namespace namespace = namespaces.get(name);
         if (namespace == null) {
-            namespace = new Namespace(name, jsonSupport, storeFactory);
+            namespace = new Namespace(name, jsonSupport, storeFactory, exceptionListener);
             Namespace oldNamespace = namespaces.putIfAbsent(name, namespace);
             if (oldNamespace != null) {
                 namespace = oldNamespace;

src/test/java/com/corundumstudio/socketio/handler/PacketHandlerTest.java

@@ -57,7 +57,7 @@ public class PacketHandlerTest {
     private JsonSupport map = new JacksonJsonSupport(new Configuration());
     private Decoder decoder = new Decoder(map, new AckManager(null));
     private Encoder encoder = new Encoder(new Configuration(), map);
-    private NamespacesHub namespacesHub = new NamespacesHub(map, null);
+    private NamespacesHub namespacesHub = new NamespacesHub(map, null, null);
     @Mocked
     private Channel channel;
     private MainBaseClient client = new XHRPollingClient(null, null, UUID.randomUUID(), null, new MemoryStoreFactory(), null);