Merge pull request #262 from tsegismont/jira/HWKMETRICS-143

[HWKMETRICS-143] ptrans: ByteBuf leak in REST forwarding handler
hawkular · Jun 22, 2015 · 24f721b · 24f721b
2 parents 9a22c77 + 974ab28
commit 24f721b
Showing 1 changed file with 12 additions and 19 deletions.
diff --git a/...ator/src/main/java/org/hawkular/metrics/clients/ptrans/backend/RestForwardingHandler.java b/...ator/src/main/java/org/hawkular/metrics/clients/ptrans/backend/RestForwardingHandler.java
@@ -45,6 +45,7 @@
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.DefaultFullHttpRequest;
@@ -242,30 +243,22 @@ protected void initChannel(SocketChannel ch) throws Exception {
      * for the next try.
      * @author Heiko W. Rupp
     */
-    class HttpStatusWatcher extends ChannelInboundHandlerAdapter {
+    class HttpStatusWatcher extends SimpleChannelInboundHandler<FullHttpResponse> {
         private final Logger logger = LoggerFactory.getLogger(HttpStatusWatcher.class);
 
         @Override
-        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-            if (msg instanceof FullHttpResponse) {
-                FullHttpResponse response = (FullHttpResponse) msg;
-                HttpResponseStatus status = response.getStatus();
-
-                if (status.equals(HttpResponseStatus.NO_CONTENT) || status.equals(HttpResponseStatus.OK)) {
-
-                    List<SingleMetric> metricsSent = ctx.channel().attr(listKey).getAndRemove();
-
-                    if (metricsSent != null) {
-                        // only clear the ones we sent - new ones may have arrived between batching and now
-                        fifo.cleanout(metricsSent);
-                        numberOfMetrics += metricsSent.size();
-                        logger.debug("sent {} items", metricsSent.size());
-                    }
-                } else {
-                    logger.warn("Send to rest-server failed:" + status);
+        protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse response) throws Exception {
+            HttpResponseStatus status = response.getStatus();
+            if (status.equals(HttpResponseStatus.NO_CONTENT) || status.equals(HttpResponseStatus.OK)) {
+                List<SingleMetric> metricsSent = ctx.channel().attr(listKey).getAndRemove();
+                if (metricsSent != null) {
+                    // only clear the ones we sent - new ones may have arrived between batching and now
+                    fifo.cleanout(metricsSent);
+                    numberOfMetrics += metricsSent.size();
+                    logger.debug("sent {} items", metricsSent.size());
                 }
             } else {
-                logger.error("msg " + msg);
+                logger.warn("Send to rest-server failed:" + status);
             }
         }
     }