service.hh

Go to the documentation of this file.

/*
 * service.hh
 *
 *  Created on: Jan 5, 2016
 *      Author: N.S. Oblath
 */
 
#ifndef DRIPLINE_SERVICE_HH_
#define DRIPLINE_SERVICE_HH_
 
#include "core.hh"
#include "endpoint.hh"
#include "heartbeater.hh"
#include "scheduler.hh"
#include "listener.hh"
#include "receiver.hh"
 
#include "dripline_exceptions.hh"
#include "service_config.hh"
#include "uuid.hh"
 
#include <map>
#include <memory>
#include <set>
#include <vector>
 
namespace scarab
{
    class authentication;
}
namespace dripline
{
    class DRIPLINE_API service :
            public core,
            public endpoint,
            public listener_receiver,
            public heartbeater,
            public scheduler<>
    {
        protected:
            enum class status
            {
                nothing = 0,
                channel_created = 10,
                exchange_declared = 20,
                queue_declared = 30,
                queue_bound = 40,
                consuming = 50,
                listening = 60,
                processing = 70
            };
 
        public:
            /* 
               \brief Extracts necessary configuration and authentication information and prepares the service to interact with the RabbitMQ broker. Does not initiate connection to the broker.
               @param a_config Dripline configuration object.  The `name` must be unique for each service.  The `dripline.broker` (and `dripline.broker_port` if needed) should be made appropriate for the mesh.  
                 The other parameters can be left as their defaults, or should be made uniform across the mesh.
                 - *Service parameters*
                   - `name` (string; default: dlcpp_service) -- Name of the service and the queue used by the service
                   - `restart_on_error` (bool; default: true) -- Flag for whether the service attempts to restart itself if an error occurs in communicating with the broker
                   - `enable_scheduling` (bool; default: false) -- Flag for enabling the scheduler
                   - `broadcast_key` (string; default: broadcast) -- Routing key used for broadcasts
                   - `loop_timeout_ms` (int; default: 1000) -- Maximum time used for listening timeouts (e.g. waiting for replies) in ms
                   - `message_wait_ms` (int; default: 1000) -- Maximum time used to wait for another AMQP message before declaring a DL message complete, in ms
                   - `heartbeat_interval_s` (int; default: 60) -- Interval between sending heartbeat messages in s
                 - *Dripline core parameters -- within the `dripline` config object*
                   - `dripline.broker` (string; default: localhost) -- Address of the RabbitMQ broker
                   - `dripline.broker_port` (int; default: 5672) -- Port used by the RabbitMQ broker
                   - `dripline.requests_exchange` (string; default: requests) -- Name of the exchange used for DL requests
                   - `dripline.alerts_exchange` (string; default: alerts) -- Name of the exchange used for DL alerts
                   - `dripline.heartbeat_routing_key` (string; default: heartbeat) -- Routing key used for sending heartbeats
                   - `dripline.max_payload_size` (int; default: DL_MAX_PAYLOAD_SIZE) -- Maximum size of payloads, in bytes
                   - `dripline.max_connection_attempts` (int; default: 10) -- Maximum number of attempts that will be made to connect to the broker
                   - `dripline.return_codes` (string or array of nodes; default: not present) -- Optional specification of additional return codes in the form of an array of nodes: `[{name: "<name>", value: <ret code>} <, ...>]`. 
                          If this is a string, it's treated as a file can be interpreted by the param system (e.g. YAML or JSON) using the previously-mentioned format
               @param a_auth Authentication object (type scarab::authentication); authentication specification should be processed, and the authentication data should include:
               @param a_make_connection Flag for whether or not to contact a broker; if true, this object operates in "dry-run" mode
             */
            service( const scarab::param_node& a_config = service_config(), 
                     const scarab::authentication& a_auth = create_auth_with_dripline(true), 
                     const bool a_make_connection = true );
//            service( const bool a_make_connection, const scarab::param_node& a_config = scarab::param_node(), const scarab::authentication& a_auth = scarab::authentication() );
            service( const service& ) = delete;
            service( service&& a_orig ) = default;
            virtual ~service();
 
            service& operator=( const service& ) = delete;
            service& operator=( service&& a_orig );
 
            mv_referrable( scarab::authentication, auth );
 
            mv_accessible( status, status );
            mv_accessible( bool, restart_on_error );
            mv_accessible( bool, enable_scheduling );
 
        public:
            bool add_child( endpoint_ptr_t a_endpoint_ptr );
 
            bool add_async_child( endpoint_ptr_t a_endpoint_ptr );
 
        public:
            virtual sent_msg_pkg_ptr send( request_ptr_t a_request, amqp_channel_ptr a_channel = amqp_channel_ptr() ) const;
 
            virtual sent_msg_pkg_ptr send( reply_ptr_t a_reply, amqp_channel_ptr a_channel = amqp_channel_ptr() ) const;
 
            virtual sent_msg_pkg_ptr send( alert_ptr_t a_alert, amqp_channel_ptr a_channel = amqp_channel_ptr() ) const;
 
        public:
            virtual void run();
 
            bool start();
 
            bool listen();
 
            bool stop();
 
        protected:
            virtual bool open_channels();
 
            virtual bool setup_queues();
 
            virtual bool bind_keys();
 
            virtual bool start_consuming();
 
            virtual bool stop_consuming();
 
            virtual bool remove_queue();
 
        public:
            virtual bool listen_on_queue();
 
            virtual void send_reply( reply_ptr_t a_reply ) const;
 
            mv_accessible( uuid_t, id );
 
        public:
            typedef std::map< std::string, endpoint_ptr_t > sync_map_t;
            mv_referrable( sync_map_t, sync_children );
 
            typedef std::map< std::string, lr_ptr_t > async_map_t;
            mv_referrable( async_map_t, async_children );
 
            mv_referrable( std::string, broadcast_key );
 
        protected:
            virtual void submit_message( message_ptr_t a_message );
 
            virtual reply_ptr_t on_request_message( const request_ptr_t a_request );
 
        private:
            virtual void do_cancellation( int a_code );
    };
 
    inline sent_msg_pkg_ptr service::send( request_ptr_t a_request, amqp_channel_ptr a_channel ) const
    {
        a_request->sender_service_name() = f_name;
        // we don't use f_channel on this core::send command because a channel can only be used in a single thread, 
        // and f_channel is primarily meant for listening with the listener thread.
        return core::send( a_request, a_channel );
    }
 
    inline sent_msg_pkg_ptr service::send( reply_ptr_t a_reply, amqp_channel_ptr a_channel ) const
    {
        a_reply->sender_service_name() = f_name ;
        // we don't use f_channel on this core::send command because a channel can only be used in a single thread, 
        // and f_channel is primarily meant for listening with the listener thread.
        return core::send( a_reply, a_channel );
    }
 
    inline sent_msg_pkg_ptr service::send( alert_ptr_t a_alert, amqp_channel_ptr a_channel ) const
    {
        a_alert->sender_service_name() = f_name;
        // we don't use f_channel on this core::send command because a channel can only be used in a single thread, 
        // and f_channel is primarily meant for listening with the listener thread.
        return core::send( a_alert, a_channel );
    }
 
} /* namespace dripline */
 
#endif /* DRIPLINE_SERVICE_HH_ */