/*
 * 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 org.jboss.netty.handler.codec.redis;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelDownstreamHandler;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

public class RedisEncoder extends SimpleChannelDownstreamHandler {

    private Queue<ChannelBuffer> pool = new ConcurrentLinkedQueue<ChannelBuffer>();

    @Override
    public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        Object o = e.getMessage();
        if (o instanceof Command) {
            Command command = (Command) o;
            ChannelBuffer cb = pool.poll();
            if (cb == null) {
                cb = ChannelBuffers.dynamicBuffer();
            }
            command.write(cb);
            ChannelFuture future = e.getFuture();
            final ChannelBuffer finalCb = cb;
            future.addListener(new ChannelFutureListener() {
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    finalCb.clear();
                    pool.add(finalCb);
                }
            });
            Channels.write(ctx, future, cb);
        } else {
            super.writeRequested(ctx, e);
        }
    }
}

@@ -1065,16 +1065,72 @@
 Handler;
+%0Aimport org.jboss.netty.channel.ChannelHandler.Sharable;
 %0A%0Aimport
@@ -1199,16 +1199,26 @@
 Queue;%0A%0A
+@Sharable%0A
 public c
@@ -1289,16 +1289,22 @@
 private 
+final 
 Queue%3CCh
@@ -1324,55 +1324,712 @@
 pool
- = new ConcurrentLinkedQueue%3CChannelBuffer%3E();%0A
+;%0A%0A    /**%0A     * Calls %7B@link #RedisEncoder(boolean)%7D with %3Ccode%3Efalse%3C/code%3E%0A     */%0A    public RedisEncoder() %7B%0A        this(false);%0A    %7D%0A    %0A    /**%0A     * Create a new %7B@link RedisEncoder%7D instance %0A     * %0A     * @param poolBuffers %3Ccode%3Etrue%3C/code%3E if the %7B@link ChannelBuffer%7D's should be pooled. This should be used with caution as this %0A     *                    can lead to unnecessary big memory consummation if one of the written values is very big and the rest is very small.%0A     */%0A    public RedisEncoder(boolean poolBuffers) %7B%0A        if (poolBuffers) %7B%0A            pool = new ConcurrentLinkedQueue%3CChannelBuffer%3E();%0A        %7D else %7B%0A            pool = null;%0A        %7D%0A    %7D%0A    %0A    
 %0A   
@@ -2281,20 +2281,93 @@
 b = 
-pool.poll();
+null;%0A            if (pool != null) %7B%0A                cb = pool.poll();%0A            %7D
 %0A   
@@ -2537,24 +2537,61 @@
 etFuture();%0A
+%0A            if (pool != null) %7B%0A    
             
@@ -2636,16 +2636,20 @@
         
+ 
+   
 future.a
@@ -2685,24 +2685,28 @@
 istener() %7B%0A
+    
             
@@ -2803,24 +2803,28 @@
             
+   
+ 
 finalCb.clea
@@ -2824,24 +2824,28 @@
 Cb.clear();%0A
+    
             
@@ -2867,24 +2867,28 @@
 d(finalCb);%0A
+    
             
@@ -2909,11 +2909,29 @@
     
-%7D);
+    %7D);%0A            %7D
 %0A   

Make pooling configurable