Logging details during incorrect data packing/unpacking. Issue #36

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

@@ -23,6 +23,7 @@ import org.jboss.netty.handler.codec.frame.TooLongFrameException;
 
 import com.corundumstudio.socketio.parser.JacksonJsonSupport;
 import com.corundumstudio.socketio.parser.JsonSupport;
+import com.corundumstudio.socketio.parser.JsonSupportWrapper;
 
 public class Configuration {
 
@@ -68,7 +69,7 @@ public class Configuration {
         setHeartbeatThreadPoolSize(conf.getHeartbeatThreadPoolSize());
         setHeartbeatTimeout(conf.getHeartbeatTimeout());
         setHostname(conf.getHostname());
-        setJsonSupport(conf.getJsonSupport());
+        setJsonSupport(new JsonSupportWrapper(conf.getJsonSupport()));
         setPort(conf.getPort());
         setWorkerExecutor(conf.getWorkerExecutor());
         setContext(conf.getContext());

src/main/java/com/corundumstudio/socketio/parser/JacksonJsonSupport.java

@@ -16,8 +16,6 @@
 package com.corundumstudio.socketio.parser;
 
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
 import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Iterator;
@@ -26,6 +24,9 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.jboss.netty.buffer.ChannelBufferInputStream;
+import org.jboss.netty.buffer.ChannelBufferOutputStream;
+
 import com.corundumstudio.socketio.Configuration;
 import com.corundumstudio.socketio.utils.ConcurrentHashSet;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
@@ -214,18 +215,18 @@ public class JacksonJsonSupport implements JsonSupport {
     }
 
     @Override
-    public <T> T readValue(InputStream src, Class<T> valueType) throws IOException {
+    public <T> T readValue(ChannelBufferInputStream src, Class<T> valueType) throws IOException {
         return objectMapper.readValue(src, valueType);
     }
 
     @Override
-    public AckArgs readAckArgs(InputStream src, Class<?> argType) throws IOException {
+    public AckArgs readAckArgs(ChannelBufferInputStream src, Class<?> argType) throws IOException {
         currentAckClass.set(argType);
         return objectMapper.readValue(src, AckArgs.class);
     }
 
     @Override
-    public void writeValue(OutputStream out, Object value) throws IOException {
+    public void writeValue(ChannelBufferOutputStream out, Object value) throws IOException {
         objectMapper.writeValue(out, value);
 
     }

src/main/java/com/corundumstudio/socketio/parser/JsonSupport.java

@@ -16,8 +16,9 @@
 package com.corundumstudio.socketio.parser;
 
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
+
+import org.jboss.netty.buffer.ChannelBufferInputStream;
+import org.jboss.netty.buffer.ChannelBufferOutputStream;
 
 /**
  * JSON infrastructure interface.
@@ -27,11 +28,11 @@ import java.io.OutputStream;
  */
 public interface JsonSupport {
 
-    AckArgs readAckArgs(InputStream src, Class<?> argType) throws IOException;
+    AckArgs readAckArgs(ChannelBufferInputStream src, Class<?> argType) throws IOException;
 
-    <T> T readValue(InputStream src, Class<T> valueType) throws IOException;
+    <T> T readValue(ChannelBufferInputStream src, Class<T> valueType) throws IOException;
 
-    void writeValue(OutputStream out, Object value) throws IOException;
+    void writeValue(ChannelBufferOutputStream out, Object value) throws IOException;
 
     String writeValueAsString(Object value) throws IOException;
 

src/main/java/com/corundumstudio/socketio/parser/JsonSupportWrapper.java

@@ -0,0 +1,99 @@
+/**
+ * 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.parser;
+
+import java.io.IOException;
+
+import org.jboss.netty.buffer.ChannelBufferInputStream;
+import org.jboss.netty.buffer.ChannelBufferOutputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JsonSupportWrapper implements JsonSupport {
+
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private final JsonSupport delegate;
+
+    public JsonSupportWrapper(JsonSupport delegate) {
+        this.delegate = delegate;
+    }
+
+    public AckArgs readAckArgs(ChannelBufferInputStream src, Class<?> argType) throws IOException {
+        try {
+            return delegate.readAckArgs(src, argType);
+        } catch (IOException e) {
+            src.reset();
+            log.error("Can't read ack args: " + src.readLine() + " for type: " + argType, e);
+            return null;
+        } catch (RuntimeException e) {
+            src.reset();
+            log.error("Can't read ack args: " + src.readLine() + " for type: " + argType, e);
+            return null;
+        }
+    }
+
+    public <T> T readValue(ChannelBufferInputStream src, Class<T> valueType) throws IOException {
+        try {
+            return delegate.readValue(src, valueType);
+        } catch (IOException e) {
+            src.reset();
+            log.error("Can't read value 1: " + src.readLine() + " for type: " + valueType, e);
+            return null;
+        } catch (RuntimeException e) {
+            src.reset();
+            log.error("Can't read value 2: " + src.readLine() + " for type: " + valueType, e);
+            return null;
+        }
+    }
+
+    public void writeValue(ChannelBufferOutputStream out, Object value) throws IOException {
+        try {
+            delegate.writeValue(out, value);
+        } catch (IOException e) {
+            log.error("Can't write value: " + value, e);
+        } catch (RuntimeException e) {
+            log.error("Can't write value: " + value, e);
+        }
+    }
+
+    public String writeValueAsString(Object value) throws IOException {
+        try {
+            return delegate.writeValueAsString(value);
+        } catch (IOException e) {
+            log.error("Can't convert value to string: " + value, e);
+            return null;
+        } catch (RuntimeException e) {
+            log.error("Can't convert value to string: " + value, e);
+            return null;
+        }
+    }
+
+    public void addEventMapping(String eventName, Class<?> eventClass) {
+        delegate.addEventMapping(eventName, eventClass);
+    }
+
+    public void addJsonClass(Class<?> clazz) {
+        delegate.addJsonClass(clazz);
+    }
+
+    public void removeEventMapping(String eventName) {
+        delegate.removeEventMapping(eventName);
+    }
+
+
+
+}

src/main/java/com/corundumstudio/socketio/utils/ConcurrentHashSet.java

@@ -24,6 +24,7 @@ import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * Adapter for ConcurrentHashMap, a-la jdk 1.6+ Collections.newSetFromMap(new ConcurrentHashMap....) idiom
+ *
  * @author radu.cirstoiu@gmail.com
  */
 public class ConcurrentHashSet<E> extends AbstractSet<E> implements Set<E> {
@@ -31,7 +32,7 @@ public class ConcurrentHashSet<E> extends AbstractSet<E> implements Set<E> {
     private final Set<E> backingMapSet;
 
     public ConcurrentHashSet() {
-		backingMap = (new ConcurrentHashMap<E, Boolean>());
+        backingMap = new ConcurrentHashMap<E, Boolean>();
         backingMapSet = backingMap.keySet();
     }