ReactiveSocket frame logger (tcp)

#### Problem

Tracking ReactiveSocket frames is not easy with wire logging as it logs the bytes written/read.

#### Modification

Added a netty handler to be added as the first handler in the pipeline and logs frame objects as-is written and read on the channel.
Following netty's logging handler design, this handler can be configured to log at a particular log level, which can be changed by the user at runtime.

#### Result

Better tracking of ReactiveSocket frames on the channel.

Author: NiteshKant

reactivesocket-transport-tcp/src/main/java/io/reactivesocket/transport/tcp/ReactiveSocketFrameLogger.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2016 Netflix, Inc.
+ * <p>
+ *  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
+ *  <p>
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  <p>
+ *  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 io.reactivesocket.transport.tcp;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.reactivesocket.Frame;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+public class ReactiveSocketFrameLogger extends ChannelDuplexHandler {
+
+    private final Logger logger;
+    private final Level logLevel;
+
+    public ReactiveSocketFrameLogger(String name, Level logLevel) {
+        this.logLevel = logLevel;
+        logger = LoggerFactory.getLogger(name);
+    }
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        logFrameIfEnabled(ctx, msg, " Writing frame: ");
+        super.write(ctx, msg, promise);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        logFrameIfEnabled(ctx, msg, " Read frame: ");
+        super.channelRead(ctx, msg);
+    }
+
+    private void logFrameIfEnabled(ChannelHandlerContext ctx, Object msg, String logMsgPrefix) {
+        if (msg instanceof Frame) {
+            Frame f = (Frame) msg;
+            switch (logLevel) {
+            case ERROR:
+                if (logger.isErrorEnabled()) {
+                    logger.error(ctx.channel() + logMsgPrefix + f);
+                }
+                break;
+            case WARN:
+                if (logger.isWarnEnabled()) {
+                    logger.warn(ctx.channel() + logMsgPrefix + f);
+                }
+                break;
+            case INFO:
+                if (logger.isInfoEnabled()) {
+                    logger.info(ctx.channel() + logMsgPrefix + f);
+                }
+                break;
+            case DEBUG:
+                if (logger.isDebugEnabled()) {
+                    logger.debug(ctx.channel() + logMsgPrefix + f);
+                }
+                break;
+            case TRACE:
+                if (logger.isTraceEnabled()) {
+                    logger.trace(ctx.channel() + logMsgPrefix + f);
+                }
+                break;
+            }
+        }
+    }
+}

reactivesocket-transport-tcp/src/main/java/io/reactivesocket/transport/tcp/client/TcpTransportClient.java

@@ -21,10 +21,12 @@
 import io.reactivesocket.Frame;
 import io.reactivesocket.transport.TransportClient;
 import io.reactivesocket.transport.tcp.ReactiveSocketFrameCodec;
+import io.reactivesocket.transport.tcp.ReactiveSocketFrameLogger;
 import io.reactivesocket.transport.tcp.ReactiveSocketLengthCodec;
 import io.reactivesocket.transport.tcp.TcpDuplexConnection;
 import io.reactivex.netty.protocol.tcp.client.TcpClient;
 import org.reactivestreams.Publisher;
+import org.slf4j.event.Level;
 
 import java.net.SocketAddress;
 import java.util.function.Function;
@@ -56,6 +58,20 @@ public TcpTransportClient configureClient(Function<TcpClient<Frame, Frame>, TcpC
         return new TcpTransportClient(configurator.apply(rxNettyClient));
     }
 
+    /**
+     * Enable logging of every frame read and written on every connection created by this client.
+     *
+     * @param name Name of the logger.
+     * @param logLevel Level at which the messages will be logged.
+     *
+     * @return A new {@link TcpTransportClient}
+     */
+    public TcpTransportClient logReactiveSocketFrames(String name, Level logLevel) {
+        return configureClient(c ->
+            c.addChannelHandlerLast("reactive-socket-frame-codec", () -> new ReactiveSocketFrameLogger(name, logLevel))
+        );
+    }
+
     public static TcpTransportClient create(SocketAddress serverAddress) {
         return new TcpTransportClient(_configureClient(TcpClient.newClient(serverAddress)));
     }

reactivesocket-transport-tcp/src/main/java/io/reactivesocket/transport/tcp/server/TcpTransportServer.java

@@ -20,11 +20,14 @@
 import io.reactivesocket.Frame;
 import io.reactivesocket.transport.TransportServer;
 import io.reactivesocket.transport.tcp.ReactiveSocketFrameCodec;
+import io.reactivesocket.transport.tcp.ReactiveSocketFrameLogger;
 import io.reactivesocket.transport.tcp.ReactiveSocketLengthCodec;
 import io.reactivesocket.transport.tcp.TcpDuplexConnection;
+import io.reactivesocket.transport.tcp.client.TcpTransportClient;
 import io.reactivex.netty.channel.Connection;
 import io.reactivex.netty.protocol.tcp.server.ConnectionHandler;
 import io.reactivex.netty.protocol.tcp.server.TcpServer;
+import org.slf4j.event.Level;
 import rx.Observable;
 
 import java.net.SocketAddress;
@@ -64,6 +67,20 @@ public TcpTransportServer configureServer(Function<TcpServer<Frame, Frame>, TcpS
         return new TcpTransportServer(configurator.apply(rxNettyServer));
     }
 
+    /**
+     * Enable logging of every frame read and written on every connection accepted by this server.
+     *
+     * @param name Name of the logger.
+     * @param logLevel Level at which the messages will be logged.
+     *
+     * @return A new {@link TcpTransportServer}
+     */
+    public TcpTransportServer logReactiveSocketFrames(String name, Level logLevel) {
+        return configureServer(c -> c.addChannelHandlerLast("reactive-socket-frame-codec",
+                                                            () -> new ReactiveSocketFrameLogger(name, logLevel))
+        );
+    }
+
     public static TcpTransportServer create() {
         return create(TcpServer.newServer());
     }