soa/service/connection_handler.h

/* connection_handler.h                                            -*- C++ -*-
   Jeremy Barnes, 27 February 2011
   Copyright (c) 2011 Datacratic.  All rights reserved.

   Connection handler logic.
*/

#ifndef __rtb__connection_handler_h__
#define __rtb__connection_handler_h__

#include "transport.h"
#include <iostream>
#include <boost/function.hpp>
#include <list>
#include "jml/arch/format.h"
#include "jml/arch/demangle.h"
#include "jml/arch/atomic_ops.h"

namespace Datacratic {


/*****************************************************************************/
/* CONNECTION HANDLER                                                        */
/*****************************************************************************/

struct ConnectionHandler {

    static uint32_t created, destroyed;

    ConnectionHandler()
        : transport_(0), magic(0x1234)
    {
        ML::atomic_add(created, 1);
    }

    virtual ~ConnectionHandler()
    {
        ML::atomic_add(destroyed, 1);

        if (magic != 0x1234)
            throw ML::Exception("Attempt to double free connection handler");
        magic = 0;
    }

    virtual void onGotTransport()
    {
    }

    virtual void onDisassociate()
    {
    }

    virtual void onHandlerException(const std::string & handler,
                                    const std::exception & exc)
    {
        //using namespace std;
        //cerr << "handler " << handler << " threw exception "
        //     << exc.what() << endl;
        doError("handler " + handler + " had exception " + exc.what());
    }

    virtual void onCleanup()
    {
    }

    virtual void handleError(const std::string & message)
    {
        doError(message);
        closeWhenHandlerFinished();
    }

    virtual void handleDisconnect()
    {
        closeWhenHandlerFinished();
    }

    virtual std::string status() const
    {
        return ML::format("%p of type %s", this, ML::type_name(*this).c_str());
    }

    virtual void doError(const std::string & error) = 0;

    void closeConnection();

    ssize_t send(const char * buf, size_t len, int flags)
    {
        return transport().send(buf, len, flags);
    }

    ssize_t recv(char * buf, size_t buf_size, int flags)
    {
        return transport().recv(buf, buf_size, flags);
    }

    int getHandle() const
    {
        return transport().getHandle();
    }

    void startReading();

    void stopReading();

    void startWriting();

    void stopWriting();

    void scheduleTimerAbsolute(Date timeout,
                               size_t cookie = 0,
                               void (*freecookie) (size_t) = 0)
    {
        transport().scheduleTimerAbsolute(timeout, cookie);
    }

    void scheduleTimerRelative(double secondsFromNow,
                               size_t cookie = 0,
                               void (*freecookie) (size_t) = 0)
    {
        transport().scheduleTimerRelative(secondsFromNow, cookie);
    }
    
    void cancelTimer()
    {
        transport().cancelTimer();
    }

    void closeWhenHandlerFinished()
    {
        transport().closeWhenHandlerFinished();
    }

    void recycleWhenHandlerFinished()
    {
        transport().recycleWhenHandlerFinished();
    }

    /* Event callbacks.  The default implementations throw an exception.
    */
    virtual void handleInput();
    virtual void handleOutput();
    virtual void handlePeerShutdown();
    virtual void handleTimeout(Date time, size_t cookie);

    bool hasTransport() const
    {
        return transport_;
    }

    TransportBase & transport()
    {
        if (!transport_)
            throw ML::Exception("connection asked for transport with none "
                                "set");
        return *transport_;
    }

    const TransportBase & transport() const
    {
        if (!transport_)
            throw ML::Exception("connection asked for transport with none "
                                "set");
        return *transport_;
    }

    EndpointBase * get_endpoint() { return transport().get_endpoint(); }

    //virtual int handlerReturnCode() const = 0;

    void addActivity(const std::string & activity);

    void addActivityS(const char * activity);

    void addActivity(const char * fmt, ...);

    void checkMagic() const;

    void doAsync(const boost::function<void ()> & callback,
                 const char * name)
    {
        transport().doAsync(callback, name);
    }

private:
    void setTransport(TransportBase * transport);
    TransportBase * transport_;
    friend class TransportBase;

    int closePeer()
    {
        return transport().closePeer();
    }

    int magic;
};


/*****************************************************************************/
/* PASSIVE CONNECTION HANDLER                                                */
/*****************************************************************************/

struct PassiveConnectionHandler: public ConnectionHandler {

    PassiveConnectionHandler()
        : inSend(false)
    {
    }

    std::string error;

    int done;
    bool inSend;

    enum NextAction {
        NEXT_CLOSE,
        NEXT_RECYCLE,
        NEXT_CONTINUE
    };

    typedef boost::function<void ()> OnWriteFinished;

    struct WriteEntry {
        Date date;
        std::string data;
        OnWriteFinished onWriteFinished;
        NextAction next;
    };

    std::list<WriteEntry> toWrite;
    
    void send(const std::string & str,
              NextAction action = NEXT_CONTINUE,
              OnWriteFinished onWriteFinished = OnWriteFinished());
    
    virtual void handleData(const std::string & data) = 0;
    
    virtual void handleError(const std::string & message) = 0;

    virtual void doError(const std::string & error);

    /* ACE_Event_Handler callbacks */
    virtual void handleInput();
    virtual void handleOutput();
    virtual void handleTimeout(Date time, size_t cookie);

    friend class TransportBase;
};

} // namespace Datacratic

#endif /* __rtb__connection_handler_h__ */