@@ -0,0 +1,2534 @@
+package com.tacitknowledge.performance.degrade;%0A%0Aimport java.util.concurrent.TimeUnit;%0A%0Aimport com.tacitknowledge.performance.Component;%0A%0Aimport io.netty.buffer.ByteBuf;%0Aimport io.netty.buffer.Unpooled;%0Aimport io.netty.channel.ChannelDuplexHandler;%0Aimport io.netty.channel.ChannelHandlerContext;%0Aimport io.netty.channel.ChannelPromise;%0Aimport io.netty.channel.socket.SocketChannel;%0A%0A/**%0A * TODO: refactor this. For higher traffic rates use buffers larger then 1 byte size and calculate appropriate delays%0A * TODO: separate configs for up and down traffics%0A *%0A * @author Ivan Frolov (ifrolov@tacitknowledge.com) */%0Apublic class SlowDown implements Component%0A%7B%0A    private final long millisecondsPerByte;%0A%0A    public SlowDown(final long millisecondsPerByte)%0A    %7B%0A        this.millisecondsPerByte = millisecondsPerByte;%0A    %7D%0A%0A    @Override%0A    public void initChannel(final SocketChannel ch)%0A    %7B%0A        ch.config().setReceiveBufferSize(1);%0A        ch.pipeline().addLast(new SlowDownHandler(millisecondsPerByte));%0A    %7D%0A%0A    public static class SlowDownHandler extends ChannelDuplexHandler%0A    %7B%0A        private final long millisecondsPerByte;%0A%0A        public SlowDownHandler(final long millisecondsPerByte)%0A        %7B%0A            this.millisecondsPerByte = millisecondsPerByte;%0A        %7D%0A%0A        @Override%0A        public void channelRead(final ChannelHandlerContext ctx, final Object msg) throws Exception%0A        %7B%0A            ctx.executor().schedule(new Runnable()%0A            %7B%0A                @Override%0A                public void run()%0A                %7B%0A                    ctx.fireChannelRead(msg);%0A                %7D%0A            %7D, millisecondsPerByte, TimeUnit.MILLISECONDS);%0A        %7D%0A%0A        @Override%0A        public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) throws Exception%0A        %7B%0A            final ByteBuf buf = (ByteBuf) msg;%0A            Runnable task = new Runnable()%0A            %7B%0A                private int ix = buf.readerIndex();%0A                @Override%0A                public void run()%0A                %7B%0A                    if(ix %3C buf.writerIndex()) %7B%0A                        ctx.writeAndFlush(Unpooled.wrappedBuffer(buf.copy(ix, 1)));%0A                        ix++;%0A                        ctx.executor().schedule(this, millisecondsPerByte, TimeUnit.MILLISECONDS);%0A                    %7D else %7B%0A                        promise.setSuccess();%0A                    %7D%0A                %7D%0A            %7D;%0A%0A            ctx.executor().submit(task);%0A        %7D%0A    %7D%0A%7D%0A

Implement traffic slowdown degradation handler.