/**
 *  Copyright 2008 ThimbleWare Inc.
 *
 *  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
 *
 *      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 com.thimbleware.jmemcached;

import com.thimbleware.jmemcached.protocol.text.MemcachedPipelineFactory;
import com.thimbleware.jmemcached.protocol.binary.MemcachedBinaryPipelineFactory;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

/**
 * The actual daemon - responsible for the binding and configuration of the network configuration.
 */
public class MemCacheDaemon {

    final Logger logger = LoggerFactory.getLogger(MemCacheDaemon.class);

    public static String memcachedVersion = "0.7";

    private int receiveBufferSize = 32768 * 1024;
    private int sendBufferSize = 32768;

    private boolean binary = false;
    private boolean verbose;
    private int idleTime;
    private InetSocketAddress addr;
    private Cache cache;

    private boolean running = false;
    private NioServerSocketChannelFactory channelFactory;
    private DefaultChannelGroup allChannels;


    public MemCacheDaemon() {
    }

    public MemCacheDaemon(Cache cache) {
        this.cache = cache;
    }

    /**
     * Bind the network connection and start the network processing threads.
     *
     * @throws IOException
     */
    public void start() throws IOException {
        channelFactory =
                new NioServerSocketChannelFactory(
                        Executors.newCachedThreadPool(),
                        Executors.newCachedThreadPool());

        ServerBootstrap bootstrap = new ServerBootstrap(channelFactory);
        allChannels = new DefaultChannelGroup("jmemcachedChannelGroup");

        ChannelPipelineFactory pipelineFactory;
        if (binary)
            pipelineFactory = new MemcachedBinaryPipelineFactory(cache, memcachedVersion, verbose, idleTime, allChannels);
        else
            pipelineFactory = new MemcachedPipelineFactory(cache, memcachedVersion, verbose, idleTime, receiveBufferSize, allChannels);

        bootstrap.setPipelineFactory(pipelineFactory);

        Channel serverChannel = bootstrap.bind(addr);
        allChannels.add(serverChannel);

        bootstrap.setOption("child.tcpNoDelay", true);
        bootstrap.setOption("child.keepAlive", true);

        logger.info("Listening on " + String.valueOf(addr.getHostName()) + ":" + addr.getPort());
        running = true;
    }

    public void stop() {
        ChannelGroupFuture future = allChannels.close();
        future.awaitUninterruptibly();
        if (!future.isCompleteSuccess()) {
            System.err.println("shit");
        }
        channelFactory.releaseExternalResources();
        running = false;
    }

    public static void setMemcachedVersion(String memcachedVersion) {
        MemCacheDaemon.memcachedVersion = memcachedVersion;
    }

    public void setReceiveBufferSize(int receiveBufferSize) {
        this.receiveBufferSize = receiveBufferSize;
    }

    public void setSendBufferSize(int sendBufferSize) {
        this.sendBufferSize = sendBufferSize;
    }

    public void setVerbose(boolean verbose) {
        this.verbose = verbose;
    }

    public void setIdleTime(int idleTime) {
        this.idleTime = idleTime;
    }

    public void setAddr(InetSocketAddress addr) {
        this.addr = addr;
    }


    public Cache getCache() {
        return cache;
    }

    public void setCache(Cache cache) {
        this.cache = cache;
    }

    public boolean isRunning() {
        return running;
    }

    public boolean isBinary() {
        return binary;
    }

    public void setBinary(boolean binary) {
        this.binary = binary;
    }
}

@@ -1122,32 +1122,171 @@
 ChannelFactory;%0A
+import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;%0Aimport org.jboss.netty.channel.socket.ClientSocketChannelFactory;%0A
 import org.slf4j
@@ -1294,16 +1294,16 @@
 Logger;%0A
-
 import o
@@ -1428,16 +1428,54 @@
 ecutors;
+%0Aimport java.util.concurrent.Executor;
 %0A%0A/**%0A *
@@ -1727,17 +1727,17 @@
 on = %220.
-7
+9
 %22;%0A%0A    
@@ -2599,81 +2599,8 @@
 );%0A%0A
-        ServerBootstrap bootstrap = new ServerBootstrap(channelFactory);%0A
     
@@ -2665,24 +2665,104 @@
 elGroup%22);%0A%0A
+        ServerBootstrap bootstrap = new ServerBootstrap(channelFactory);%0A      %0A
         Chan
@@ -3086,24 +3086,133 @@
 Channels);%0A%0A
+        bootstrap.setOption(%22child.tcpNoDelay%22, true);%0A        bootstrap.setOption(%22child.keepAlive%22, true);%0A
         boot
@@ -3309,16 +3309,16 @@
 (addr);%0A
+
         
@@ -3354,118 +3354,8 @@
 );%0A%0A
-        bootstrap.setOption(%22child.tcpNoDelay%22, true);%0A        bootstrap.setOption(%22child.keepAlive%22, true);%0A%0A
     

Fix to location of tcpNoDelay setting, which was ineffective given it was happening after the bind. Many fold speed up in synchronous operations.