soa/service/passive_endpoint.h

/* passive_endpoint.h                                              -*- C++ -*-
   Jeremy Barnes, 29 April 2011
   Copyright (C) 2011 Datacratic.  All rights reserved.

   Base class for a passive endpoint.
*/

#pragma once

#include "soa/service/endpoint.h"
#include "soa/service/port_range_service.h"
#include "jml/arch/wakeup_fd.h"

namespace Datacratic {

enum {
    DEF_BACKLOG = 128
};

/*****************************************************************************/
/* ACCEPTOR                                                                  */
/*****************************************************************************/

struct Acceptor {
    virtual ~Acceptor()
    {
    }

    virtual int
    listen(PortRange const & portRange, const std::string & hostname,
           PassiveEndpoint * endpoint, bool nameLookup, int backlog) = 0;

    virtual void closePeer() = 0;

    virtual std::string hostname() const = 0;

    virtual int port() const = 0;
};


/*****************************************************************************/
/* PASSIVE ENDPOINT                                                          */
/*****************************************************************************/

struct PassiveEndpoint: public EndpointBase {

    PassiveEndpoint(const std::string & name);

    virtual ~PassiveEndpoint();

    int init(PortRange const & portRange = PortRange(), const std::string & hostname = "localhost",
             int threads = 1, bool synchronous = true, bool nameLookup=true,
             int backlog = DEF_BACKLOG);

    virtual int listen(PortRange const & portRange, const std::string & host,bool nameLookup=true,
                       int backlog = DEF_BACKLOG)
    {
        if (!acceptor)
            throw ML::Exception("can't listen without acceptor");

        return acceptor->listen(portRange, host, this, nameLookup, backlog);
    }

    virtual void closePeer()
    {
        return acceptor->closePeer();
    }

    virtual std::string hostname() const
    {
        return acceptor->hostname();
    }

    virtual int port() const
    {
        return acceptor->port();
    }

    boost::function<std::shared_ptr<ConnectionHandler> ()> onMakeNewHandler;

    boost::function<void (std::string)> onAcceptError;
    
protected:

    std::shared_ptr<Acceptor> acceptor;

    virtual void
    associateHandler(const std::shared_ptr<TransportBase> & transport)
    {
        if (!transport)
            throw ML::Exception("no transport");
        transport->hasConnection();
        notifyNewTransport(transport);

        auto finishAccept = [=] ()
            {
                std::shared_ptr<ConnectionHandler> handler
                    = this->makeNewHandler();
                transport->associate(handler);
            };

        transport->doAsync(finishAccept, "finishAccept");
    }

    virtual std::shared_ptr<ConnectionHandler>
    makeNewHandler()
    {
        return onMakeNewHandler();
    }

    virtual void acceptError(const std::string & error)
    {
        if (onAcceptError) onAcceptError(error);
        else {
            using namespace std;
            cerr << "error accepting connection: " << error << endl;
        }
    }

    template<typename Transport> friend struct AcceptorT;
    // whether or not to perform a host name look up
    bool nameLookup_;// whether or not to perform a host name look up
};


/*****************************************************************************/
/* ACCEPTOR TEMPLATE                                                         */
/*****************************************************************************/

template<typename Transport>
struct AcceptorT: public Acceptor {
};

/*****************************************************************************/
/* ACCEPTOR TEMPLATE FOR SOCKET TRANSPORT                                    */
/*****************************************************************************/

template<>
struct AcceptorT<SocketTransport> : public Acceptor {

    AcceptorT();
    virtual ~AcceptorT();

    virtual int listen(PortRange const & portRange,
                       const std::string & hostname,
                       PassiveEndpoint * endpoint,
                       bool nameLookup,
                       int backlog);

    virtual void closePeer();

    virtual std::string hostname() const;

    virtual int port() const;

    void runAcceptThread();


protected:
    std::shared_ptr<boost::thread> acceptThread;
    ML::Wakeup_Fd wakeup;
    ACE_INET_Addr addr;
    int fd;
    PassiveEndpoint * endpoint;
    bool nameLookup;
    bool shutdown;
};


/*****************************************************************************/
/* PASSIVE ENDPOINT TEMPLATE                                                 */
/*****************************************************************************/

template<typename Transport>
struct PassiveEndpointT: public PassiveEndpoint {

    PassiveEndpointT(const std::string & name)
        : PassiveEndpoint(name)
    {
        acceptor.reset(new AcceptorT<Transport>());
    }
    
    virtual ~PassiveEndpointT()
    {
    }
};

} // namespace Datacratic