Repository: netty/netty
Issue: the traffic package need ported back to 4.0

We miss the traffic package so much,can you port it back to 4.0 please?

And how about  the JMX support plan in 4.0?


Issue comments:
@phoenix we are currently try to "port" one "piece" after the other. But as you may realize we are quite busy at the moment to the the API rock stable. 

So the best thing for you to get it add back in a quick manner is to chime in an port it by yourself.. After that a pull-request would be really welcome :)

The JMX stuff will hopefully get also some more attention in the near future. As we want to integrate our new metrics api in more parts of Netty. This could then be used to export the metrics via jmx. At the moment we have a default implementation which use yammer metrics, and as far as I can tell they support jmx out of the box

I've ported this over to 4. I just need to fix up the javadoc and then will issue a pull request. Will likely be tomorrow evening before I can get a chance to commit the last bit.

@lw346 thanks !!!

Also see #851 

Fixed by #851 


Pull request review comments:
I think we don't need that anymore as the readable(..) flag is now per ChannelHandlerContext and so it should be enough to just use this method...

See above...

I know it is just the same as in 3 but this just is an anti-pattern as it will block the IO-Thread. Let us fix this.

See above

The reason I kept this attribute setting was in case there's another handler in the pipeline that will modify readable() on the ChannelHandlerContext.  In this instance, we need to make sure that we're not overriding another handler's modifications - the attribute ensures that we will only make modifications if **this** handler needs to.

I'm quite happy to swap it out as you suggest, but it feels cleaner to keep the handlers self-contained.

The ChannelHandlerContext is "bound" to the ChannelHandler instance so only "this instance" can change it. In 3.x this was different as the readable(..) flag was modified on the Channel and so was not "per" ChannelHandlerContext. So we don't need it anymore..

This should be "true" all the times r I'm wrong ?


Patch that resolved the issue:
diff --git a/handler/src/main/java/io/netty/handler/traffic/AbstractTrafficShapingHandler.java b/handler/src/main/java/io/netty/handler/traffic/AbstractTrafficShapingHandler.java
new file mode 100644
index 00000000000..b061647b089
--- /dev/null
+++ b/handler/src/main/java/io/netty/handler/traffic/AbstractTrafficShapingHandler.java
@@ -0,0 +1,364 @@
+/*
+ * Copyright 2011 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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 io.netty.handler.traffic;
+
+import io.netty.buffer.Buf;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerAdapter;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundByteHandler;
+import io.netty.channel.ChannelOutboundByteHandler;
+import io.netty.util.AttributeKey;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * AbstractTrafficShapingHandler allows to limit the global bandwidth
+ * (see {@link GlobalTrafficShapingHandler}) or per session
+ * bandwidth (see {@link ChannelTrafficShapingHandler}), as traffic shaping.
+ * It allows you to implement an almost real time monitoring of the bandwidth using
+ * the monitors from {@link TrafficCounter} that will call back every checkInterval
+ * the method doAccounting of this handler.<br>
+ * <br>
+ *
+ * If you want for any particular reasons to stop the monitoring (accounting) or to change
+ * the read/write limit or the check interval, several methods allow that for you:<br>
+ * <ul>
+ * <li><tt>configure</tt> allows you to change read or write limits, or the checkInterval</li>
+ * <li><tt>getTrafficCounter</tt> allows you to have access to the TrafficCounter and so to stop
+ * or start the monitoring, to change the checkInterval directly, or to have access to its values.</li>
+ * </ul>
+ */
+public abstract class AbstractTrafficShapingHandler extends ChannelHandlerAdapter
+        implements ChannelInboundByteHandler, ChannelOutboundByteHandler {
+
+    /**
+     * Default delay between two checks: 1s
+     */
+    public static final long DEFAULT_CHECK_INTERVAL = 1000;
+
+    /**
+     * AttributeKey to indicate whether this handler has suspended read
+     * operations on the channel
+     */
+    protected AttributeKey<Boolean> readSuspended;
+
+    /**
+     * Default minimal time to wait
+     */
+    private static final long MINIMAL_WAIT = 10;
+
+    /**
+     * Traffic Counter
+     */
+    protected TrafficCounter trafficCounter;
+
+    /**
+     * used in releaseExternalResources() to cancel the timer
+     */
+    private volatile ScheduledFuture<?> scheduledFuture;
+
+    /**
+     * Limit in B/s to apply to write
+     */
+    private long writeLimit;
+
+    /**
+     * Limit in B/s to apply to read
+     */
+    private long readLimit;
+
+    /**
+     * Delay between two performance snapshots
+     */
+    protected long checkInterval = DEFAULT_CHECK_INTERVAL; // default 1 s
+
+    /**
+     * Boolean associated with the release of this TrafficShapingHandler.
+     * It will be true only once when the releaseExternalRessources is called
+     * to prevent waiting when shutdown.
+     */
+    final AtomicBoolean release = new AtomicBoolean(false);
+
+    /**
+     *
+     * @param newTrafficCounter the TrafficCounter to set
+     */
+    void setTrafficCounter(TrafficCounter newTrafficCounter) {
+        trafficCounter = newTrafficCounter;
+    }
+
+    /**
+     * @param writeLimit
+     *          0 or a limit in bytes/s
+     * @param readLimit
+     *          0 or a limit in bytes/s
+     * @param checkInterval
+     *          The delay between two computations of performances for
+     *            channels or 0 if no stats are to be computed
+     */
+    protected AbstractTrafficShapingHandler(long writeLimit, long readLimit,
+                                            long checkInterval) {
+        this.writeLimit = writeLimit;
+        this.readLimit = readLimit;
+        this.checkInterval = checkInterval;
+    }
+
+    /**
+     * Constructor using default Check Interval
+     *
+     * @param writeLimit
+     *          0 or a limit in bytes/s
+     * @param readLimit
+     *          0 or a limit in bytes/s
+     */
+    protected AbstractTrafficShapingHandler(long writeLimit, long readLimit) {
+        this(writeLimit, readLimit, DEFAULT_CHECK_INTERVAL);
+    }
+
+    /**
+     * Constructor using NO LIMIT and default Check Interval
+     */
+    protected AbstractTrafficShapingHandler() {
+        this(0, 0, DEFAULT_CHECK_INTERVAL);
+    }
+
+    /**
+     * Constructor using NO LIMIT
+     *
+     * @param checkInterval
+     *          The delay between two computations of performances for
+     *            channels or 0 if no stats are to be computed
+     */
+    protected AbstractTrafficShapingHandler(long checkInterval) {
+        this(0, 0, checkInterval);
+    }
+
+    /**
+     * Change the underlying limitations and check interval.
+     *
+     * @param newWriteLimit The new write limit (in bytes)
+     * @param newReadLimit The new read limit (in bytes)
+     * @param newCheckInterval The new check interval (in milliseconds)
+     */
+    public void configure(long newWriteLimit, long newReadLimit,
+            long newCheckInterval) {
+        configure(newWriteLimit, newReadLimit);
+        configure(newCheckInterval);
+    }
+
+    /**
+     * Change the underlying limitations.
+     *
+     * @param newWriteLimit The new write limit (in bytes)
+     * @param newReadLimit The new read limit (in bytes)
+     */
+    public void configure(long newWriteLimit, long newReadLimit) {
+        writeLimit = newWriteLimit;
+        readLimit = newReadLimit;
+        if (trafficCounter != null) {
+            trafficCounter.resetAccounting(System.currentTimeMillis() + 1);
+        }
+    }
+
+    /**
+     * Change the check interval.
+     *
+     * @param newCheckInterval The new check interval (in milliseconds)
+     */
+    public void configure(long newCheckInterval) {
+        checkInterval = newCheckInterval;
+        if (trafficCounter != null) {
+            trafficCounter.configure(checkInterval);
+        }
+    }
+
+    /**
+     * Called each time the accounting is computed from the TrafficCounters.
+     * This method could be used for instance to implement almost real time accounting.
+     *
+     * @param counter
+     *            the TrafficCounter that computes its performance
+     */
+    protected void doAccounting(TrafficCounter counter) {
+        // NOOP by default
+    }
+
+    /**
+     * Class to implement setReadable at fix time
+     */
+    private class ReopenReadTimerTask implements Runnable {
+        final ChannelHandlerContext ctx;
+        ReopenReadTimerTask(ChannelHandlerContext ctx) {
+            this.ctx = ctx;
+        }
+
+        @Override
+        public void run() {
+            if (release.get()) {
+                return;
+            }
+
+            if (ctx != null && ctx.channel() != null &&
+                    ctx.channel().isActive()) {
+                ctx.attr(readSuspended).set(null);
+                ctx.readable(true);
+            }
+        }
+    }
+
+    /**
+    *
+    * @return the time that should be necessary to wait to respect limit. Can
+    *         be negative time
+    */
+    private static long getTimeToWait(long limit, long bytes, long lastTime,
+            long curtime) {
+        long interval = curtime - lastTime;
+        if (interval == 0) {
+            // Time is too short, so just lets continue
+            return 0;
+        }
+        return ((bytes * 1000 / limit - interval) / 10) * 10;
+    }
+
+    @Override
+    public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
+        return ctx.alloc().buffer();
+    }
+
+    @Override
+    public void freeInboundBuffer(ChannelHandlerContext ctx, Buf buf) throws Exception {
+        buf.free();
+    }
+
+    @Override
+    public ByteBuf newOutboundBuffer(ChannelHandlerContext ctx) throws Exception {
+        return ctx.alloc().buffer();
+    }
+
+    @Override
+    public void freeOutboundBuffer(ChannelHandlerContext ctx, Buf buf) throws Exception {
+        buf.free();
+    }
+
+    @Override
+    public void inboundBufferUpdated(ChannelHandlerContext ctx) throws Exception {
+        ByteBuf buf = ctx.inboundByteBuffer();
+        long curtime = System.currentTimeMillis();
+        long size = buf.readableBytes();
+
+        if (trafficCounter != null) {
+            trafficCounter.bytesRecvFlowControl(size);
+            if (readLimit == 0) {
+                // no action
+                return;
+            }
+
+            // compute the number of ms to wait before reopening the channel
+            long wait = getTimeToWait(readLimit,
+                                      trafficCounter.getCurrentReadBytes(),
+                                      trafficCounter.getLastTime(), curtime);
+            if (wait >= MINIMAL_WAIT) { // At least 10ms seems a minimal
+                // time in order to
+                Channel channel = ctx.channel();
+                // try to limit the traffic
+                if (channel != null && channel.isActive()) {
+                    if (ctx.attr(readSuspended).get() == null) {
+                        ctx.attr(readSuspended).set(Boolean.TRUE);
+                        ctx.readable(false);
+                        Runnable timerTask = new ReopenReadTimerTask(ctx);
+                        scheduledFuture = ctx.executor().schedule(timerTask, wait,
+                                                   TimeUnit.MILLISECONDS);
+                    } else {
+                        // should be waiting: but can occurs sometime so as
+                        // a FIX
+                        if (release.get()) {
+                            return;
+                        }
+                        Thread.sleep(wait);
+                    }
+                } else {
+                    // Not connected or no channel
+                    if (release.get()) {
+                        return;
+                    }
+                    Thread.sleep(wait);
+                }
+            }
+        }
+    }
+
+    @Override
+    public void flush(ChannelHandlerContext ctx, ChannelFuture future) throws Exception {
+        long curtime = System.currentTimeMillis();
+        long size = ctx.outboundByteBuffer().readableBytes();
+
+        try {
+            if (trafficCounter != null) {
+                trafficCounter.bytesWriteFlowControl(size);
+                if (writeLimit == 0) {
+                    return;
+                }
+                // compute the number of ms to wait before continue with the
+                // channel
+                long wait = getTimeToWait(writeLimit,
+                        trafficCounter.getCurrentWrittenBytes(),
+                        trafficCounter.getLastTime(), curtime);
+                if (wait >= MINIMAL_WAIT) {
+                    // Global or Channel
+                    if (release.get()) {
+                        return;
+                    }
+                    Thread.sleep(wait);
+                }
+            }
+        } finally {
+            ctx.flush(future);
+        }
+    }
+
+    /**
+     *
+     * @return the current TrafficCounter (if
+     *         channel is still connected)
+     */
+    public TrafficCounter getTrafficCounter() {
+        return trafficCounter;
+    }
+
+    @Override
+    public void beforeRemove(ChannelHandlerContext ctx) {
+        if (trafficCounter != null) {
+            trafficCounter.stop();
+        }
+        release.set(true);
+        if (scheduledFuture != null) {
+            scheduledFuture.cancel(true);
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "TrafficShaping with Write Limit: " + writeLimit +
+                " Read Limit: " + readLimit + " and Counter: " +
+                (trafficCounter != null? trafficCounter.toString() : "none");
+    }
+}
diff --git a/handler/src/main/java/io/netty/handler/traffic/ChannelTrafficShapingHandler.java b/handler/src/main/java/io/netty/handler/traffic/ChannelTrafficShapingHandler.java
new file mode 100644
index 00000000000..4158d867488
--- /dev/null
+++ b/handler/src/main/java/io/netty/handler/traffic/ChannelTrafficShapingHandler.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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 io.netty.handler.traffic;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.AttributeKey;
+
+/**
+ * This implementation of the {@link AbstractTrafficShapingHandler} is for channel
+ * traffic shaping, that is to say a per channel limitation of the bandwidth.<br><br>
+ *
+ * The general use should be as follow:<br>
+ * <ul>
+ * <li>Add in your pipeline a new ChannelTrafficShapingHandler.<br>
+ * <tt>ChannelTrafficShapingHandler myHandler = new ChannelTrafficShapingHandler();</tt><br>
+ * <tt>pipeline.addLast(myHandler);</tt><br><br>
+ *
+ * <b>Note that this handler has a Pipeline Coverage of "one" which means a new handler must be created
+ * for each new channel as the counter cannot be shared among all channels.</b>.<br><br>
+ *
+ * Other arguments can be passed like write or read limitation (in bytes/s where 0 means no limitation)
+ * or the check interval (in millisecond) that represents the delay between two computations of the
+ * bandwidth and so the call back of the doAccounting method (0 means no accounting at all).<br><br>
+ *
+ * A value of 0 means no accounting for checkInterval. If you need traffic shaping but no such accounting,
+ * it is recommended to set a positive value, even if it is high since the precision of the
+ * Traffic Shaping depends on the period where the traffic is computed. The highest the interval,
+ * the less precise the traffic shaping will be. It is suggested as higher value something close
+ * to 5 or 10 minutes.<br>
+ * </li>
+ * </ul><br>
+ */
+public class ChannelTrafficShapingHandler extends AbstractTrafficShapingHandler {
+
+    public ChannelTrafficShapingHandler(long writeLimit,
+            long readLimit, long checkInterval) {
+        super(writeLimit, readLimit, checkInterval);
+    }
+
+    public ChannelTrafficShapingHandler(long writeLimit,
+            long readLimit) {
+        super(writeLimit, readLimit);
+    }
+
+    public ChannelTrafficShapingHandler(long checkInterval) {
+        super(checkInterval);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx)
+            throws Exception {
+        if (trafficCounter != null) {
+            trafficCounter.stop();
+        }
+
+        super.channelInactive(ctx);
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        readSuspended = new AttributeKey<Boolean>("readSuspended-channel");
+        ctx.attr(readSuspended).set(Boolean.TRUE);
+        ctx.readable(false);
+        if (trafficCounter == null) {
+            // create a new counter now
+            if (ctx.executor() != null) {
+                trafficCounter = new TrafficCounter(this, ctx.executor(), "ChannelTC" +
+                        ctx.channel().id(), checkInterval);
+            }
+        }
+        if (trafficCounter != null) {
+            trafficCounter.start();
+        }
+        super.channelActive(ctx);
+
+        ctx.attr(readSuspended).set(null);
+        ctx.readable(true);
+    }
+
+}
diff --git a/handler/src/main/java/io/netty/handler/traffic/GlobalTrafficShapingHandler.java b/handler/src/main/java/io/netty/handler/traffic/GlobalTrafficShapingHandler.java
new file mode 100644
index 00000000000..4b817fc3a45
--- /dev/null
+++ b/handler/src/main/java/io/netty/handler/traffic/GlobalTrafficShapingHandler.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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 io.netty.handler.traffic;
+
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.EventExecutor;
+import io.netty.util.AttributeKey;
+
+/**
+ * This implementation of the {@link AbstractTrafficShapingHandler} is for global
+ * traffic shaping, that is to say a global limitation of the bandwidth, whatever
+ * the number of opened channels.<br><br>
+ *
+ * The general use should be as follow:<br>
+ * <ul>
+ * <li>Create your unique GlobalTrafficShapingHandler like:<br><br>
+ * <tt>GlobalTrafficShapingHandler myHandler = new GlobalTrafficShapingHandler(executor);</tt><br><br>
+ * The executor could be the underlying IO worker pool<br>
+ * <tt>pipeline.addLast(myHandler);</tt><br><br>
+ *
+ * <b>Note that this handler has a Pipeline Coverage of "all" which means only one such handler must be created
+ * and shared among all channels as the counter must be shared among all channels.</b><br><br>
+ *
+ * Other arguments can be passed like write or read limitation (in bytes/s where 0 means no limitation)
+ * or the check interval (in millisecond) that represents the delay between two computations of the
+ * bandwidth and so the call back of the doAccounting method (0 means no accounting at all).<br><br>
+ *
+ * A value of 0 means no accounting for checkInterval. If you need traffic shaping but no such accounting,
+ * it is recommended to set a positive value, even if it is high since the precision of the
+ * Traffic Shaping depends on the period where the traffic is computed. The highest the interval,
+ * the less precise the traffic shaping will be. It is suggested as higher value something close
+ * to 5 or 10 minutes.<br>
+ * </li>
+ * </ul><br>
+ */
+@Sharable
+public class GlobalTrafficShapingHandler extends AbstractTrafficShapingHandler {
+    /**
+     * Create the global TrafficCounter
+     */
+    void createGlobalTrafficCounter(EventExecutor executor) {
+        readSuspended = new AttributeKey<Boolean>("readSuspended-global");
+        TrafficCounter tc;
+        if (executor != null) {
+            tc = new TrafficCounter(this, executor, "GlobalTC",
+                    checkInterval);
+            setTrafficCounter(tc);
+            tc.start();
+        }
+    }
+
+    public GlobalTrafficShapingHandler(EventExecutor executor, long writeLimit,
+            long readLimit, long checkInterval) {
+        super(writeLimit, readLimit, checkInterval);
+        createGlobalTrafficCounter(executor);
+    }
+
+    public GlobalTrafficShapingHandler(EventExecutor executor, long writeLimit,
+            long readLimit) {
+        super(writeLimit, readLimit);
+        createGlobalTrafficCounter(executor);
+    }
+
+    public GlobalTrafficShapingHandler(EventExecutor executor, long checkInterval) {
+        super(checkInterval);
+        createGlobalTrafficCounter(executor);
+    }
+
+    public GlobalTrafficShapingHandler(EventExecutor executor) {
+        createGlobalTrafficCounter(executor);
+    }
+}
diff --git a/handler/src/main/java/io/netty/handler/traffic/TrafficCounter.java b/handler/src/main/java/io/netty/handler/traffic/TrafficCounter.java
new file mode 100644
index 00000000000..95b47e96da2
--- /dev/null
+++ b/handler/src/main/java/io/netty/handler/traffic/TrafficCounter.java
@@ -0,0 +1,394 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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 io.netty.handler.traffic;
+
+import io.netty.channel.EventExecutor;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * TrafficCounter is associated with {@link AbstractTrafficShapingHandler}.
+ *
+ * <p>A <tt>TrafficCounter</tt> counts the read and written bytes such that the
+ * {@link AbstractTrafficShapingHandler} can limit the traffic, globally or per channel.</p>
+ *
+ * <p>It computes the statistics for both read and written every {@link #checkInterval}, and calls
+ * back to its parent {@link AbstractTrafficShapingHandler#doAccounting} method.  If the checkInterval
+ * is set to 0, no accounting will be done and statistics will only be computed at each receive or
+ * write operation.</p>
+ */
+public class TrafficCounter {
+    /**
+     * Current written bytes
+     */
+    private final AtomicLong currentWrittenBytes = new AtomicLong();
+
+    /**
+     * Current read bytes
+     */
+    private final AtomicLong currentReadBytes = new AtomicLong();
+
+    /**
+     * Long life written bytes
+     */
+    private final AtomicLong cumulativeWrittenBytes = new AtomicLong();
+
+    /**
+     * Long life read bytes
+     */
+    private final AtomicLong cumulativeReadBytes = new AtomicLong();
+
+    /**
+     * Last Time where cumulative bytes where reset to zero
+     */
+    private long lastCumulativeTime;
+
+    /**
+     * Last writing bandwidth
+     */
+    private long lastWriteThroughput;
+
+    /**
+     * Last reading bandwidth
+     */
+    private long lastReadThroughput;
+
+    /**
+     * Last Time Check taken
+     */
+    private final AtomicLong lastTime = new AtomicLong();
+
+    /**
+     * Last written bytes number during last check interval
+     */
+    private long lastWrittenBytes;
+
+    /**
+     * Last read bytes number during last check interval
+     */
+    private long lastReadBytes;
+
+    /**
+     * Delay between two captures
+     */
+    AtomicLong checkInterval = new AtomicLong(
+            AbstractTrafficShapingHandler.DEFAULT_CHECK_INTERVAL);
+
+    // default 1 s
+
+    /**
+     * Name of this Monitor
+     */
+    final String name;
+
+    /**
+     * The associated TrafficShapingHandler
+     */
+    private final AbstractTrafficShapingHandler trafficShapingHandler;
+
+    /**
+     * Executor that will run the monitor
+     */
+    private final EventExecutor executor;
+    /**
+     * Monitor created once in start()
+     */
+    private Runnable monitor;
+    /**
+     * used in stop() to cancel the timer
+     */
+    private volatile ScheduledFuture<?> scheduledFuture;
+
+    /**
+     * Is Monitor active
+     */
+    AtomicBoolean monitorActive = new AtomicBoolean();
+
+    /**
+     * Class to implement monitoring at fix delay
+     *
+     */
+    private static class TrafficMonitoringTask implements Runnable {
+        /**
+         * The associated TrafficShapingHandler
+         */
+        private final AbstractTrafficShapingHandler trafficShapingHandler1;
+
+        /**
+         * The associated TrafficCounter
+         */
+        private final TrafficCounter counter;
+
+        /**
+         * @param trafficShapingHandler The parent handler to which this task needs to callback to for accounting
+         * @param counter The parent TrafficCounter that we need to reset the statistics for
+         */
+        protected TrafficMonitoringTask(
+                AbstractTrafficShapingHandler trafficShapingHandler,
+                TrafficCounter counter) {
+            trafficShapingHandler1 = trafficShapingHandler;
+            this.counter = counter;
+        }
+
+        @Override
+        public void run() {
+            if (!counter.monitorActive.get()) {
+                return;
+            }
+            long endTime = System.currentTimeMillis();
+            counter.resetAccounting(endTime);
+            if (trafficShapingHandler1 != null) {
+                trafficShapingHandler1.doAccounting(counter);
+            }
+            counter.scheduledFuture = counter.executor.schedule(this, counter.checkInterval.get(),
+                                                                TimeUnit.MILLISECONDS);
+        }
+    }
+
+    /**
+     * Start the monitoring process
+     */
+    public void start() {
+        synchronized (lastTime) {
+            if (monitorActive.get()) {
+                return;
+            }
+            lastTime.set(System.currentTimeMillis());
+            if (checkInterval.get() > 0) {
+                monitorActive.set(true);
+                monitor = new TrafficMonitoringTask(trafficShapingHandler, this);
+                scheduledFuture =
+                    executor.schedule(monitor, checkInterval.get(), TimeUnit.MILLISECONDS);
+            }
+        }
+    }
+
+    /**
+     * Stop the monitoring process
+     */
+    public void stop() {
+        synchronized (lastTime) {
+            if (!monitorActive.get()) {
+                return;
+            }
+            monitorActive.set(false);
+            resetAccounting(System.currentTimeMillis());
+            if (trafficShapingHandler != null) {
+                trafficShapingHandler.doAccounting(this);
+            }
+            if (scheduledFuture != null) {
+                scheduledFuture.cancel(true);
+            }
+        }
+    }
+
+    /**
+     * Reset the accounting on Read and Write
+     *
+     * @param newLastTime the millisecond unix timestamp that we should be considered up-to-date for
+     */
+    void resetAccounting(long newLastTime) {
+        synchronized (lastTime) {
+            long interval = newLastTime - lastTime.getAndSet(newLastTime);
+            if (interval == 0) {
+                // nothing to do
+                return;
+            }
+            lastReadBytes = currentReadBytes.getAndSet(0);
+            lastWrittenBytes = currentWrittenBytes.getAndSet(0);
+            lastReadThroughput = lastReadBytes / interval * 1000;
+            // nb byte / checkInterval in ms * 1000 (1s)
+            lastWriteThroughput = lastWrittenBytes / interval * 1000;
+            // nb byte / checkInterval in ms * 1000 (1s)
+        }
+    }
+
+    /**
+     * Constructor with the {@link AbstractTrafficShapingHandler} that hosts it, the Timer to use, its
+     * name, the checkInterval between two computations in millisecond
+     * @param trafficShapingHandler the associated AbstractTrafficShapingHandler
+     * @param executor the underlying executor service for scheduling checks
+     * @param name  the name given to this monitor
+     * @param checkInterval the checkInterval in millisecond between two computations
+     */
+    public TrafficCounter(AbstractTrafficShapingHandler trafficShapingHandler,
+            EventExecutor executor, String name, long checkInterval) {
+        this.trafficShapingHandler = trafficShapingHandler;
+        this.executor = executor;
+        this.name = name;
+        lastCumulativeTime = System.currentTimeMillis();
+        configure(checkInterval);
+    }
+
+    /**
+     * Change checkInterval between two computations in millisecond
+     *
+     * @param newcheckInterval The new check interval (in milliseconds)
+     */
+    public void configure(long newcheckInterval) {
+        long newInterval = (newcheckInterval / 10) * 10;
+        if (checkInterval.get() != newInterval) {
+            checkInterval.set(newInterval);
+            if (newInterval <= 0) {
+                stop();
+                // No more active monitoring
+                lastTime.set(System.currentTimeMillis());
+            } else {
+                // Start if necessary
+                start();
+            }
+        }
+    }
+
+    /**
+     * Computes counters for Read.
+     *
+     * @param recv
+     *            the size in bytes to read
+     */
+    void bytesRecvFlowControl(long recv) {
+        currentReadBytes.addAndGet(recv);
+        cumulativeReadBytes.addAndGet(recv);
+    }
+
+    /**
+     * Computes counters for Write.
+     *
+     * @param write
+     *            the size in bytes to write
+     */
+    void bytesWriteFlowControl(long write) {
+        currentWrittenBytes.addAndGet(write);
+        cumulativeWrittenBytes.addAndGet(write);
+    }
+
+    /**
+     *
+     * @return the current checkInterval between two computations of traffic counter
+     *         in millisecond
+     */
+    public long getCheckInterval() {
+        return checkInterval.get();
+    }
+
+    /**
+     *
+     * @return the Read Throughput in bytes/s computes in the last check interval
+     */
+    public long getLastReadThroughput() {
+        return lastReadThroughput;
+    }
+
+    /**
+     *
+     * @return the Write Throughput in bytes/s computes in the last check interval
+     */
+    public long getLastWriteThroughput() {
+        return lastWriteThroughput;
+    }
+
+    /**
+     *
+     * @return the number of bytes read during the last check Interval
+     */
+    public long getLastReadBytes() {
+        return lastReadBytes;
+    }
+
+    /**
+     *
+     * @return the number of bytes written during the last check Interval
+     */
+    public long getLastWrittenBytes() {
+        return lastWrittenBytes;
+    }
+
+    /**
+    *
+    * @return the current number of bytes read since the last checkInterval
+    */
+    public long getCurrentReadBytes() {
+        return currentReadBytes.get();
+    }
+
+    /**
+     *
+     * @return the current number of bytes written since the last check Interval
+     */
+    public long getCurrentWrittenBytes() {
+        return currentWrittenBytes.get();
+    }
+
+    /**
+     * @return the Time in millisecond of the last check as of System.currentTimeMillis()
+     */
+    public long getLastTime() {
+        return lastTime.get();
+    }
+
+    /**
+     * @return the cumulativeWrittenBytes
+     */
+    public long getCumulativeWrittenBytes() {
+        return cumulativeWrittenBytes.get();
+    }
+
+    /**
+     * @return the cumulativeReadBytes
+     */
+    public long getCumulativeReadBytes() {
+        return cumulativeReadBytes.get();
+    }
+
+    /**
+     * @return the lastCumulativeTime in millisecond as of System.currentTimeMillis()
+     * when the cumulative counters were reset to 0.
+     */
+    public long getLastCumulativeTime() {
+        return lastCumulativeTime;
+    }
+
+    /**
+     * Reset both read and written cumulative bytes counters and the associated time.
+     */
+    public void resetCumulativeTime() {
+        lastCumulativeTime = System.currentTimeMillis();
+        cumulativeReadBytes.set(0);
+        cumulativeWrittenBytes.set(0);
+    }
+
+    /**
+     * @return the name
+     */
+    public String getName() {
+        return name;
+    }
+
+    /**
+     * String information
+     */
+    @Override
+    public String toString() {
+        return "Monitor " + name + " Current Speed Read: " +
+                (lastReadThroughput >> 10) + " KB/s, Write: " +
+                (lastWriteThroughput >> 10) + " KB/s Current Read: " +
+                (currentReadBytes.get() >> 10) + " KB Current Write: " +
+                (currentWrittenBytes.get() >> 10) + " KB";
+    }
+}
diff --git a/handler/src/main/java/io/netty/handler/traffic/package-info.java b/handler/src/main/java/io/netty/handler/traffic/package-info.java
new file mode 100644
index 00000000000..44123fbb1d0
--- /dev/null
+++ b/handler/src/main/java/io/netty/handler/traffic/package-info.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.
+ */
+
+/**
+ * Implementation of a Traffic Shaping Handler and Dynamic Statistics.
+ *
+ * <p>The main goal of this package is to allow you to shape the traffic (bandwidth limitation),
+ * but also to get statistics on how many bytes are read or written. Both functions can
+ * be active or inactive (traffic or statistics).</p>
+ *
+ * <p>Two classes implement this behavior:
+ * <ul>
+ * <li> <tt>{@link TrafficCounter}</tt>: this class implements the counters needed by the handlers.
+ * It can be accessed to get some extra information like the read or write bytes since last check, the read and write
+ * bandwidth from last check...</li>
+ *
+ * <li> <tt>{@link AbstractTrafficShapingHandler}</tt>: this abstract class implements the kernel
+ * of traffic shaping. It could be extended to fit your needs. Two classes are proposed as default
+ * implementations: see {@link ChannelTrafficShapingHandler} and see {@link GlobalTrafficShapingHandler}
+ * respectively for Channel traffic shaping and Global traffic shaping.</li>
+ * </ul></p>
+ *
+ * <p>Both inbound and outbound traffic can be shaped independently.  This is done by either passing in
+ * the desired limiting values to the constructors of both the Channel and Global traffic shaping handlers,
+ * or by calling the <tt>configure</tt> method on the {@link AbstractTrafficShapingHandler}.  A value of
+ * 0 for either parameter indicates that there should be no limitation.  This allows you to monitor the
+ * incoming and outgoing traffic without shaping.</p>
+ *
+ * <p>To activate or deactivate the statistics, you can adjust the delay to a low (suggested not less than 200ms
+ * for efficiency reasons) or a high value (let say 24H in millisecond is huge enough to not get the problem)
+ * or even using <tt>0</tt> which means no computation will be done.</p>
+ *
+ * <p>If you want to do anything with these statistics, just override the <tt>doAccounting</tt> method.<br>
+ * This interval can be changed either from the method <tt>configure</tt> in {@link AbstractTrafficShapingHandler}
+ * or directly using the method <tt>configure</tt> of {@link TrafficCounter}.</p>
+ *
+ * <p>Note that a new {@link ChannelTrafficShapingHandler} must be created for each new channel,
+ * but only one {@link GlobalTrafficShapingHandler} must be created for all channels.</p>
+ *
+ * <p>Note also that you can create different GlobalTrafficShapingHandler if you want to separate classes of
+ * channels (for instance either from business point of view or from bind address point of view).</p>
+ *
+ * @apiviz.exclude ^java\.lang\.
+ */
+package io.netty.handler.traffic;
+


Test patch: