memcached_server.h

Go to the documentation of this file.

// Evo C++ Library
/* Copyright 2019 Justin Crowell
Distributed under the BSD 2-Clause License -- see included file LICENSE.txt for details.
*/

#pragma once
#ifndef INCL_evo_api_memcached_server_h
#define INCL_evo_api_memcached_server_h

#include "memcached_common.h"
#include "../ioasync_server.h"
#include "../strtok.h"

namespace evo {
namespace async {


struct MemcachedServerHandlerBase : public AsyncServerHandler {
    typedef evo::async::Memcached Memcached;    

    enum Command {
        cUNKNOWN = 0,       
        cADD,               
        cAPPEND,            
        cCAS,               
        cDECREMENT,         
        cDELETE,            
        cGAT,               
        cGATS,              
        cGET,               
        cGETS,              
        cINCREMENT,         
        cPREPEND,           
        cQUIT,              
        cREPLACE,           
        cSET,               
        cSTATS,             
        cTOUCH,             
        cVERSION,           
        cENUM_END           
    };

    EVO_ENUM_MAP_PREFIXED(Command, c,
        "add",
        "append",
        "cas",
        "decr",
        "delete",
        "gat",
        "gats",
        "get",
        "gets",
        "incr",
        "prepend",
        "quit",
        "replace",
        "set",
        "stats",
        "touch",
        "version"
    );

    struct StoreParams {
        String key;         
        uint32 flags;       
        int64  expire;      
        ulong  size;        
        uint64 cas_id;      

        StoreParams() : flags(0), expire(0), size(0), cas_id(0) { }
    };

    enum GetStartResultValue {
        gsrCONTINUE,    
        gsrSKIP         
    };

    enum GeneralResultValue {
        grOK,           
        grNOT_FOUND     
    };

    typedef ResponseResult<Memcached::StoreResult> StoreResult;     
    typedef ResponseResult<GetStartResultValue>    GetStartResult;  
    typedef ResponseResult<UInt64>                 IncrementResult; 
    typedef ResponseResult<GeneralResultValue>     DeleteResult;    
    typedef ResponseResult<GeneralResultValue>     TouchResult;     

    struct GetAdvParams {
        Int64 expire;   
        bool  cas;      
    };

    typedef DeferredContextT<MemcachedServerHandlerBase> DeferredContext;

    struct DeferredReply : DeferredContext::ReplyBase {
        DeferredReply(DeferredContext& context, ulong id) : DeferredContext::ReplyBase(context, id) {
        }

        ~DeferredReply() {
            if (!finished && context.handler != NULL) {
                context.handler->logger.log(LOG_LEVEL_ERROR, "MemcServer DeferredReply left unfinished");
                deferred_reply_error("Internal handler error: DeferredReply left unfinished");
            }
        }

        void deferred_reply_error(const SubString& msg) {
            if (context.handler != NULL) {
                String buf;
                buf.reserve(15 + msg.size());
                buf.set("SERVER_ERROR ", 13);
                buf << msg;
                buf.add("\r\n", 2);
                DeferredContext::Handler& handler = *context.handler;
                handler.reply.deferred_send(id, buf, true);
                if (handler.logger.check(LOG_LEVEL_DEBUG_LOW))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(40 + msg.size()) << "MemcServer deferred reply " << id << " error: " << msg);
                if (handler.reply.deferred_end(context))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, "MemcServer cleanup");
            }
            finished = true;
        }

        void deferred_reply_store(Memcached::StoreResult result) {
            if (context.handler != NULL) {
                String buf;
                switch (result) {
                    case Memcached::srSTORED:     buf.set("STORED\r\n", 8);      break;
                    case Memcached::srNOT_STORED: buf.set("NOT_STORED\r\n", 12); break;
                    case Memcached::srEXISTS:     buf.set("EXISTS\r\n", 8);      break;
                    case Memcached::srNOT_FOUND:  buf.set("NOT_FOUND\r\n", 11);  break;
                    default: buf.set("SERVER_ERROR Backend error\r\n", 28); break;
                }
                DeferredContext::Handler& handler = *context.handler;
                if (handler.logger.check(LOG_LEVEL_DEBUG_LOW))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(60 + buf.size()) << "MemcServer on_store send deferred response " << id << ": " << SubString(buf.data(), buf.size() - 2));
                handler.reply.deferred_send(id, buf, true);
                if (handler.reply.deferred_end(context))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, "MemcServer cleanup");
            }
            finished = true;
        }

        void deferred_reply_increment(UInt64 value) {
            if (context.handler != NULL) {
                String buf;
                if (value.null()) {
                    buf.set("NOT_FOUND\r\n", 11);
                } else {
                    buf.reserve(UInt64::MAXSTRLEN + 2);
                    buf << *value << SubString("\r\n", 2);
                }
                DeferredContext::Handler& handler = *context.handler;
                if (handler.logger.check(LOG_LEVEL_DEBUG_LOW))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(60 + buf.size()) << "MemcServer on_increment send deferred response " << id << ": " << SubString(buf.data(), buf.size() - 2));
                handler.reply.deferred_send(id, buf, true);
                if (handler.reply.deferred_end(context))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, "MemcServer cleanup");
            }
            finished = true;
        }

        void deferred_reply_delete(bool success) {
            if (context.handler != NULL) {
                String buf;
                if (success)
                    buf.set("DELETED\r\n", 9);
                else
                    buf.set("NOT_FOUND\r\n", 11);
                DeferredContext::Handler& handler = *context.handler;
                if (handler.logger.check(LOG_LEVEL_DEBUG_LOW))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(60 + buf.size()) << "MemcServer on_delete send deferred response " << id << ": " << SubString(buf.data(), buf.size() - 2));
                handler.reply.deferred_send(id, buf, true);
                if (handler.reply.deferred_end(context))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, "MemcServer cleanup");
            }
            finished = true;
        }

        void deferred_reply_touch(bool success) {
            if (context.handler != NULL) {
                String buf;
                if (success)
                    buf.set("TOUCHED\r\n", 9);
                else
                    buf.set("NOT_FOUND\r\n", 11);
                DeferredContext::Handler& handler = *context.handler;
                if (handler.logger.check(LOG_LEVEL_DEBUG_LOW))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(60 + buf.size()) << "MemcServer on_touch send deferred response " << id << ": " << SubString(buf.data(), buf.size() - 2));
                handler.reply.deferred_send(id, buf, true);
                if (handler.reply.deferred_end(context))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, "MemcServer cleanup");
            }
            finished = true;
        }

        void deferred_reply_get(const SubString& key, const SubString& value, uint32 flags, uint64* cas_id=NULL) {
            if (context.handler != NULL) {
                DeferredContext::Handler& handler = *context.handler;
                handler.send_value_internal(id, key, value, flags, cas_id, AsyncServerReply::wfDEFERRED);
                if (handler.logger.check(LOG_LEVEL_DEBUG_LOW))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(68 + key.size()) << "MemcServer on_get sent deferred value " << id << ": '" << key << "' (size: " << value.size() << ')');
            }
        }

        void deferred_reply_get_end() {
            if (context.handler != NULL) {
                String buf("END\r\n", 5);
                DeferredContext::Handler& handler = *context.handler;
                handler.reply.deferred_send(id, buf, true);
                if (handler.logger.check(LOG_LEVEL_DEBUG_LOW))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(48) << "MemcServer on_get end deferred response " << id);
                if (handler.reply.deferred_end(context))
                    handler.logger.log_direct(LOG_LEVEL_DEBUG_LOW, "MemcServer cleanup");
            }
            finished = true;
        }
    };

    LoggerPtr<> logger;     
    bool noreply;           
    bool enable_gat;        
    bool enable_cas;        

    MemcachedServerHandlerBase() : noreply(false), enable_gat(false), enable_cas(false) {
    }

    // Config

    static size_t get_max_initial_read() {
        const size_t MAX_INITIAL_READ_VALUE = 524288;  // 512 KB
        return MAX_INITIAL_READ_VALUE;
    }

    // Events

    virtual StoreResult on_store(DeferredContext& context, StoreParams& params, SubString& value, Command command, uint64 cas_id) {
        EVO_PARAM_UNUSED(context);
        EVO_PARAM_UNUSED(params);
        EVO_PARAM_UNUSED(value);
        EVO_PARAM_UNUSED(command);
        EVO_PARAM_UNUSED(cas_id);
        send_error("Not implemented");
        return rtHANDLED;
    }

    virtual IncrementResult on_increment(DeferredContext& context, const SubString& key, uint64 count, bool decrement) {
        EVO_PARAM_UNUSED(context);
        EVO_PARAM_UNUSED(key);
        EVO_PARAM_UNUSED(count);
        EVO_PARAM_UNUSED(decrement);
        send_error("Not implemented");
        return rtHANDLED;
    }

    virtual DeleteResult on_delete(DeferredContext& context, const SubString& key) {
        EVO_PARAM_UNUSED(context);
        EVO_PARAM_UNUSED(key);
        send_error("Not implemented");
        return rtHANDLED;
    }

    virtual TouchResult on_touch(DeferredContext& context, const SubString& key, int64 expire) {
        EVO_PARAM_UNUSED(context);
        EVO_PARAM_UNUSED(key);
        EVO_PARAM_UNUSED(expire);
        send_error("Not implemented");
        return rtHANDLED;
    }

    virtual GetStartResult on_get_start(DeferredContext& context, const SubString& keys, GetAdvParams* adv_params) {
        EVO_PARAM_UNUSED(context);
        EVO_PARAM_UNUSED(keys);
        EVO_PARAM_UNUSED(adv_params);
        return gsrCONTINUE;
    }

    virtual ResponseType on_get(DeferredContext& context, const SubString& key, GetAdvParams* adv_params) {
        EVO_PARAM_UNUSED(context);
        EVO_PARAM_UNUSED(key);
        EVO_PARAM_UNUSED(adv_params);
        return rtNORMAL;
    }

    virtual ResponseType on_get_end(DeferredContext& context) {
        EVO_PARAM_UNUSED(context);
        return rtNORMAL;
    }

    virtual void on_flush_all(ulong delay_sec) {
        EVO_PARAM_UNUSED(delay_sec);
    }

    virtual void on_stats(SubString& params) {
        EVO_PARAM_UNUSED(params);
    }

    virtual void on_version(String& version) {
        version = "Unknown";
    }

    virtual bool on_command(bool& handled, SubString& command_str, SubString& params) {
        EVO_PARAM_UNUSED(handled);
        EVO_PARAM_UNUSED(command_str);
        EVO_PARAM_UNUSED(params);
        return true;
    }

    virtual void on_error(AsyncError err) {
        EVO_PARAM_UNUSED(err);
    }

    // Helpers

    void send_reply(const SubString& msg) {
        if (!noreply) {
            AsyncServerReply::Writer writer(reply, id, msg.size() + 2);
            writer.add(msg.data(), msg.size());
            writer.add("\r\n", 2);
            if (logger.check(LOG_LEVEL_DEBUG_LOW))
                logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(32 + msg.size()) << "MemcServer -- send_reply: " << msg);
        }
    }

    void send_client_error(const SubString& msg) {
        if (!noreply) {
            AsyncServerReply::Writer writer(reply, id, 15 + msg.size());
            writer.add("CLIENT_ERROR ", 13);
            writer.add(msg.data(), msg.size());
            writer.add("\r\n", 2);
            if (logger.check(LOG_LEVEL_DEBUG_LOW))
                logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(40 + msg.size()) << "MemcServer -- send_error: CLIENT_ERROR " << msg);
        }
    }

    void send_error(const SubString& msg) {
        if (!noreply) {
            AsyncServerReply::Writer writer(reply, id, 15 + msg.size());
            writer.add("SERVER_ERROR ", 13);
            writer.add(msg.data(), msg.size());
            writer.add("\r\n", 2);
            if (logger.check(LOG_LEVEL_DEBUG_LOW))
                logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(40 + msg.size()) << "MemcServer -- send_error: SERVER_ERROR " << msg);
        }
    }

    void send_stat(const SubString& name, const SubString& value) {
        AsyncServerReply::Writer writer(reply, id, 8 + name.size() + value.size());
        writer.add("STAT ", 5);
        writer.add(name.data(), name.size());
        writer.add(' ');
        writer.add(value.data(), value.size());
        writer.add("\r\n", 2);
    }

    void send_value(const SubString& key, const SubString& value, uint32 flags=0, uint64* cas_id=NULL) {
        send_value_internal(id, key, value, flags, cas_id, AsyncServerReply::wfNONE);
        if (logger.check(LOG_LEVEL_DEBUG_LOW))
            logger.log_direct(LOG_LEVEL_DEBUG_LOW, String().reserve(42 + key.size()) << "MemcServer -- send_value '" << key << "' (size: " << value.size() << ')');
    }

private:
    void send_value_internal(ulong req_id, const SubString& key, const SubString& value, uint32 flags, uint64* cas_id, AsyncServerReply::WriterFlags writer_flags) {
        StringInt<uint32> flags_str(flags);
        StringInt<StrSizeT> value_size_str(value.size());
        StringInt<uint64> cas_id_str;

        SizeT write_size = 12 + key.size() + flags_str.size() + value_size_str.size() + value.size();
        if (cas_id != NULL) {
            cas_id_str.set(*cas_id);
            write_size += 1 + cas_id_str.size();
        }

        AsyncServerReply::Writer writer(reply, req_id, write_size, writer_flags);
        writer.add("VALUE ", 6);
        writer.add(key.data(), key.size());
        writer.add(' ').add(flags_str.data(), flags_str.size());
        writer.add(' ').add(value_size_str.data(), value_size_str.size());
        if (cas_id != NULL)
            writer.add(' ').add(cas_id_str.data(), cas_id_str.size());
        writer.add("\r\n", 2);
        writer.add(value.data(), value.size());
        writer.add("\r\n", 2);
    }
};


template<class T>
struct MemcachedServer {
    static const size_t MIN_INITIAL_READ = 0;
    static const size_t NEWLINE_LEN = 2;

    typedef T Handler;                          
    typedef MemcachedServer<Handler> This;      
    typedef AsyncServer<This> Server;           
    typedef typename Handler::DeferredContext DeferredContext;  
    typedef typename Handler::Global Global;    
    typedef typename Handler::Shared Shared;    

    typedef MemcachedServerHandlerBase HandlerBase; 

    LoggerPtr<> logger;                     
    Handler handler;                        

    MemcachedServer(Global& global, Shared& shared, LoggerBase* logger) : logger(logger), handler(global, shared), command(HandlerBase::cUNKNOWN) {
        handler.logger.set(logger);
    }

    // Read events

    // Helper for common deferred count checks -- undef'd below
    #define EVO_HELPER_HANDLER_DEFCHECK(EVENT_NAME, EVENT_MSG_SUFFIX) \
        if (handler.noreply) { \
            logger.log(LOG_LEVEL_ERROR, "MemcServer " EVENT_NAME " error: Handler returned rtDEFERRED on 'noreply' request"); \
            return false; /* invalid result, close connection */ \
        } \
        if (++expected_deferred_count != context_ref.count()) { \
            if (expected_deferred_count > context_ref.count()) \
                logger.log(LOG_LEVEL_ERROR, "MemcServer " EVENT_NAME " error: Handler returned rtDEFERRED without creating a DeferredReply"); \
            else \
                logger.log(LOG_LEVEL_ERROR, "MemcServer " EVENT_NAME " error: Handler created more than one DeferredReply, only 1 allowed per handler"); \
            return false; /* invalid result, close connection */ \
        } \
        logger.log(LOG_LEVEL_DEBUG_LOW, "MemcServer -- " EVENT_NAME " response deferred" EVENT_MSG_SUFFIX);

    // Helper for common result.type cases -- undef'd below
    #define EVO_HELPER_HANDLER_CASES(EVENT_NAME) \
        case HandlerBase::rtDEFERRED: \
            EVO_HELPER_HANDLER_DEFCHECK(EVENT_NAME, "") \
            break; \
        case HandlerBase::rtHANDLED: break; \
        default: return false; /* rtCLOSE */

    // Helper for common handler end deferred-check -- undef'd below
    #define EVO_HELPER_HANDLER_END_DEFCHECK(EVENT_NAME) \
        if (expected_deferred_count != context_ref.count()) { \
            logger.log(LOG_LEVEL_ERROR, "MemcServer " EVENT_NAME " error: Handler created DeferredReply without returning rtDEFERRED"); \
            return false; /* invalid result, close connection */ \
        }

    bool on_read_fixed(SizeT& next_size, SubString& data, void* context) {
        // Read storage value
        EVO_PARAM_UNUSED(next_size);
        assert( context != NULL );
        String logstr;
        if (logger.check(LOG_LEVEL_DEBUG)) {
            logstr.set().reserve(96 + storage_params.key.size())
                << "MemcServer on_store " << HandlerBase::CommandEnum::get_string(command)
                << " '" << storage_params.key << "' fl:" << storage_params.flags << " exp:" << storage_params.expire;
            if (command == HandlerBase::cCAS)
                logstr << " id:" << storage_params.cas_id;
            logstr << " (size: " << storage_params.size << ')';
            logger.log_direct(LOG_LEVEL_DEBUG, logstr);
        }
        data.stripr("\r\n", NEWLINE_LEN, 1);

        DeferredContext& context_ref = *(DeferredContext*)context;
        ulong expected_deferred_count = context_ref.count();
        
        HandlerBase::StoreResult result = handler.on_store(context_ref, storage_params, data, command, storage_params.cas_id);
        switch (result.type) {
            case HandlerBase::rtNORMAL:
                switch (result.result) {
                    case Memcached::srSTORED:     reply("STORED\r\n", 8);      break;
                    case Memcached::srNOT_STORED: reply("NOT_STORED\r\n", 12); break;
                    case Memcached::srEXISTS:     reply("EXISTS\r\n", 8);      break;
                    case Memcached::srNOT_FOUND:  reply("NOT_FOUND\r\n", 11);  break;
                    default:
                        if (logger.check(LOG_LEVEL_ERROR))
                            logger.log_direct(LOG_LEVEL_ERROR, logstr.set().reserve(56) << "MemcServer on_store error: Invalid handler result: " << (int)result.result);
                        return false; // invalid result, close connection
                }
                break;
            EVO_HELPER_HANDLER_CASES("on_store");
        }
        EVO_HELPER_HANDLER_END_DEFCHECK("on_store");
        return true;
    }

    bool on_read(SizeT& fixed_size, AsyncBuffers& buffers, void* context) {
        // New command
        assert( context != NULL );

        DeferredContext& context_ref = *(DeferredContext*)context;
        ulong expected_deferred_count = context_ref.count();

        const char DELIM = ' ';
        String logstr;
        SubString line;
        while (buffers.read_line(line)) {
            SubString command_str, params_str;
            line.split(' ', command_str, params_str);
            handler.set_id();
            handler.noreply = false;

            command = Handler::CommandEnum::get_enum(command_str);
            switch (command) {
                // Storage
                case HandlerBase::cCAS:
                    if (!handler.enable_cas) {
                        handler.send_error("Not implemented");
                        break;
                    } // fallthrough
                case HandlerBase::cADD:      // fallthrough
                case HandlerBase::cAPPEND:   // fallthrough
                case HandlerBase::cPREPEND:  // fallthrough
                case HandlerBase::cREPLACE:  // fallthrough
                case HandlerBase::cSET: {
                    storage_params.key.set();
                    storage_params.flags  = 0;
                    storage_params.expire = 0;
                    storage_params.size   = 0;

                    StrTokWord tok(params_str);
                    for (;;) {
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  storage_params.key    = tok.value();
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  storage_params.flags  = tok.value().getnum<uint32>(fDEC);
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  storage_params.expire = tok.value().getnum<longl>(fDEC);
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  storage_params.size   = tok.value().getnum<ulong>(fDEC);
                        if (command == HandlerBase::cCAS) {
                            EVO_TOK_NEXT_OR_BREAK(tok, DELIM); storage_params.cas_id = tok.value().getnum<uint64>();
                        }
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  handler.noreply       = (tok.value() == "noreply");
                        break;
                    }
                    if (storage_params.key.empty()) {
                        handler.send_client_error("Missing parameter, expected key");
                        break;
                    }
                    buffers.read_flush();
                    if (handler.noreply)
                        handler.reply.nosend(handler.id);

                    if (!buffers.read_fixed_helper(*this, fixed_size, storage_params.size + NEWLINE_LEN, 0, context))
                        return false;
                    if (fixed_size > 0)
                        return true;
                    continue; // next command (while loop at top)
                }

                // Increment
                case HandlerBase::cINCREMENT:
                case HandlerBase::cDECREMENT: {
                    SubString key;
                    uint64 count = 0;
                    StrTokWord tok(params_str);
                    for (;;) {
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  key             = tok.value();
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  count           = tok.value().getnum<uint64>();
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  handler.noreply = (tok.value() == "noreply");
                        break;
                    }
                    if (key.empty()) {
                        handler.send_client_error("Missing parameter, expected key");
                        break;
                    }
                    if (logger.check(LOG_LEVEL_DEBUG))
                        logger.log_direct(LOG_LEVEL_DEBUG, logstr.set().reserve(42 + key.size()) << "MemcServer on_increment '" << key << "' " << (command == HandlerBase::cDECREMENT ? '-' : '+') << count);
                    if (handler.noreply)
                        handler.reply.nosend(handler.id);
                    HandlerBase::IncrementResult result = handler.on_increment(context_ref, key, count, (command == HandlerBase::cDECREMENT));
                    switch (result.type) {
                        case HandlerBase::rtNORMAL:
                            if (!result.result.null()) {
                                StringInt<uint64,NEWLINE_LEN> result_str(*result.result, fDEC, false, NEWLINE_LEN);
                                result_str.add("\r\n", NEWLINE_LEN);
                                reply(result_str.data(), result_str.size());
                            } else
                                reply("NOT_FOUND\r\n", 11);
                            break;
                        EVO_HELPER_HANDLER_CASES("on_increment");
                    }
                    EVO_HELPER_HANDLER_END_DEFCHECK("on_increment");
                    break;
                }

                // Delete
                case HandlerBase::cDELETE: {
                    SubString key;
                    StrTokWord tok(params_str);
                    for (;;) {
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  key             = tok.value();
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  handler.noreply = (tok.value() == "noreply");
                        break;
                    }
                    if (key.empty()) {
                        handler.send_client_error("Missing parameter, expected key(s)");
                        break;
                    }
                    if (logger.check(LOG_LEVEL_DEBUG))
                        logger.log_direct(LOG_LEVEL_DEBUG, logstr.set().reserve(24 + key.size()) << "MemcServer on_delete '" << key << '\'');
                    if (handler.noreply)
                        handler.reply.nosend(handler.id);
                    HandlerBase::DeleteResult result = handler.on_delete(context_ref, key);
                    switch (result.type) {
                        case HandlerBase::rtNORMAL:
                            if (result.result == HandlerBase::grOK)
                                reply("DELETED\r\n", 9);
                            else
                                reply("NOT_FOUND\r\n", 11);
                            break;
                        EVO_HELPER_HANDLER_CASES("on_delete");
                    }
                    EVO_HELPER_HANDLER_END_DEFCHECK("on_delete");
                    break;
                }

                // Touch
                case HandlerBase::cTOUCH: {
                    SubString key;
                    int64 expire = 0;
                    StrTokWord tok(params_str);
                    for (;;) {
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  key             = tok.value();
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  expire          = tok.value().getnum<int64>(fDEC);
                        EVO_TOK_NEXT_OR_BREAK(tok, DELIM);  handler.noreply = (tok.value() == "noreply");
                        break;
                    }
                    if (key.empty()) {
                        handler.send_client_error("Missing parameter, expected key(s)");
                        break;
                    }
                    if (logger.check(LOG_LEVEL_DEBUG))
                        logger.log_direct(LOG_LEVEL_DEBUG, logstr.set().reserve(24 + key.size()) << "MemcServer on_touch '" << key << '\'');
                    if (handler.noreply)
                        handler.reply.nosend(handler.id);
                    HandlerBase::TouchResult result = handler.on_touch(context_ref, key, expire);
                    switch (result.type) {
                        case HandlerBase::rtNORMAL:
                            if (result.result == HandlerBase::grOK)
                                reply("TOUCHED\r\n", 9);
                            else
                                reply("NOT_FOUND\r\n", 11);
                            break;
                        EVO_HELPER_HANDLER_CASES("on_touch");
                    }
                    EVO_HELPER_HANDLER_END_DEFCHECK("on_touch");
                    break;
                }

                // Get
                case HandlerBase::cGAT:
                case HandlerBase::cGATS:
                case HandlerBase::cGETS:
                case HandlerBase::cGET: {
                    bool handled  = false;
                    HandlerBase::GetAdvParams adv_params;
                    HandlerBase::GetAdvParams* adv_params_ptr = NULL;
                    switch (command) {
                        case HandlerBase::cGETS:
                            // CAS variant
                            if (!handler.enable_cas) {
                                handler.send_error("Not implemented");
                                handled = true;
                                break;
                            }
                            adv_params.cas = true;
                            adv_params_ptr = &adv_params;
                            break;
                        case HandlerBase::cGATS:
                            // GAT and CAS variant
                            if (!handler.enable_cas) {
                                handler.send_error("Not implemented");
                                handled = true;
                                break;
                            }
                            adv_params.cas = true; // fallthrough
                        case HandlerBase::cGAT: {
                            // GAT variant
                            if (!handler.enable_gat) {
                                handler.send_error("Not implemented");
                                handled = true;
                                break;
                            }
                            SubString expire_str;
                            if (!params_str.token(expire_str, DELIM) || expire_str.empty() || params_str.stripl(DELIM).empty()) {
                                handler.send_client_error("Missing parameter, expected expire value and key(s)");
                                handled = true;
                                break;
                            }
                            adv_params.expire = expire_str.getnum<uint64>();
                            adv_params_ptr = &adv_params;
                            break;
                        }
                        default:
                            if (params_str.empty()) {
                                handler.send_client_error("Missing parameter, expected key(s)");
                                handled = true;
                                break;
                            }
                            break;
                    }
                    if (handled)
                        break;

                    if (adv_params_ptr != NULL && logger.check(LOG_LEVEL_DEBUG_LOW)) {
                        logstr.set().reserve(48) << "MemcServer get adv:" << command_str;
                        if (!adv_params.expire.null())
                            logstr << " exp:" << *adv_params.expire;
                        logger.log_direct(LOG_LEVEL_DEBUG_LOW, logstr);
                    }

                    uint deferred = 0;
                    HandlerBase::GetStartResult result = handler.on_get_start(context_ref, params_str, adv_params_ptr);
                    switch (result.type) {
                        case HandlerBase::rtNORMAL:
                            switch (result.result) {
                                case HandlerBase::gsrCONTINUE: {
                                    // Call on_get() for each key
                                    StrTokWord tok(params_str);
                                    while (tok.nextw(DELIM)) {
                                        if (logger.check(LOG_LEVEL_DEBUG))
                                            logger.log_direct(LOG_LEVEL_DEBUG, logstr.set().reserve(24 + tok.value().size()) << "MemcServer on_get '" << tok.value() << '\'');
                                        switch (handler.on_get(context_ref, tok.value(), adv_params_ptr)) {
                                            case HandlerBase::rtNORMAL:   break;
                                            case HandlerBase::rtHANDLED:  handled = true; break;
                                            case HandlerBase::rtDEFERRED: ++deferred; EVO_HELPER_HANDLER_DEFCHECK("on_get", " from on_get()"); break;
                                            default: return false; // rtCLOSE
                                        }
                                        EVO_HELPER_HANDLER_END_DEFCHECK("on_get");
                                        if (handled)
                                            break;
                                    }
                                }
                                case HandlerBase::gsrSKIP:
                                    break;
                            }
                            break;
                        case HandlerBase::rtDEFERRED: ++deferred; EVO_HELPER_HANDLER_DEFCHECK("on_get", " from on_get_start()"); break;
                        case HandlerBase::rtHANDLED:  handled = true; break;
                        default: return false; // rtCLOSE
                    }
                    EVO_HELPER_HANDLER_END_DEFCHECK("on_get_start");

                    if (!handled) {
                        switch (handler.on_get_end(context_ref)) {
                            case HandlerBase::rtNORMAL:   break;
                            case HandlerBase::rtHANDLED:  handled = true; break;
                            case HandlerBase::rtDEFERRED: ++deferred; EVO_HELPER_HANDLER_DEFCHECK("on_get", " from on_get_end()"); break;
                            default: return false; // rtCLOSE
                        }
                        EVO_HELPER_HANDLER_END_DEFCHECK("on_get_end");

                        if (!handled && deferred == 0) {
                            String buf("END\r\n", 5);
                            handler.reply.send(handler.id, buf);
                            handler.reply.send_end();
                        }
                    }
                    break;
                }

                // Misc
                case HandlerBase::cSTATS: handler.on_stats(params_str); break;
                case HandlerBase::cQUIT:  return false;

                // Unknown
                default: {
                    bool handled = false;
                    if (!handler.on_command(handled, command_str, params_str))
                        return false;
                    if (!handled)
                        reply("ERROR\r\n", 7);
                    break;
                }
            }
            buffers.read_flush();
        }
        return true;
    }

    void on_error(AsyncError err) {
        handler.on_error(err);
    }

    #undef EVO_HELPER_HANDLER_END_DEFCHECK
    #undef EVO_HELPER_HANDLER_CASES
    #undef EVO_HELPER_HANDLER_DEFCHECK

private:
    HandlerBase::Command     command;
    HandlerBase::StoreParams storage_params;

    // String literals only
    void reply(const char* data, StrSizeT size) {
        if (!handler.noreply) {
            String data_str(data, size);
            handler.reply.send(handler.id, data_str);
            handler.reply.send_end();
        }
    }
};


}
}
#endif