messagehandler.hh

Go to the documentation of this file.

#ifndef _DRAMA2_MESSAGEHANDLER_INC
#define _DRAMA2_MESSAGEHANDLER_INC
/*
 * History:
     07-Jan-2014 - TJF - Original version
 
 * The above ID is for Doxygen, this one has the format ACMM is looking for.
 * "@(#) $Id$"
*/
// Define to enable the MessageUserStream class - which does not work
//  on some systems.  Currently disabled all the time until we understand
//  what is wrong in more detail.
 
//#define D2_PROVIDE_MESSUSER_STREAM
 
 
#include "drama/task.hh"
#include "drama/entryinfo.hh"
#include "drama/request.hh"
#include "drama/action.hh"
#include "drama/bulkdata.hh"
#include "status.h"
#include "DitsFix.h"
#include "DitsMsgOut.h"
#include <functional>
namespace drama { 
 
    /*
     * Forward declarations of interest.
     */
    class MessageHandler;
    class Spawnable;
 
    class SdsListToUserObj : public sds::PrintObjectCR {
    private:
        MessageHandler *_handler;
    public:
        SdsListToUserObj(MessageHandler *);
        void Print(const std::string &line) const override;
    };
 
    using MessageHandlerPtr = std::shared_ptr<MessageHandler> ;
 
 
    using MessageReceiveFunction = std::function<Request (MessageHandler *)> ;
 
    class MessageHandler : public ActionHandler  {
        friend Spawnable;  // Will invoke MessageEvent and InvokeActionEnd. 
    private:
 
        /*
         * MessageEvent is invoked when a message is received. Must invoke
         * the sub-class's method.
         */
        Request MessageEvent(std::weak_ptr<Task> task, Action *action) 
            override ;
        /*
         * This is invoked when the action ends.  Invokes the user override-able 
         *   method ActionEnd().
         */
        void InvokeActionEnd(std::weak_ptr<Task> task, Action *action,
                             bool taskExiting, 
                             StatusType actionEndStatus) override ;
 
        /*
         * These three are only valid during a call to MessageReceived() or
         *   ActionEnd().
         */
        std::weak_ptr<Task> _theTask;
        Action *_action; // Detail of action (Should change to weak_ptr?)
        EntryInfo _entryInfo;      // Action entry details.
 
        /*
         * An object which can be used to list SDS structures via
         * MessageUser.
         */
        SdsListToUserObj _sdsListToUserObj = this;
    public:
 
        MessageHandler() : 
            _action(nullptr) {
 
        }
 
        virtual ~MessageHandler();
 
        MessageHandler& operator=(const MessageHandler &rhs) = delete;
        MessageHandler(const MessageHandler &source) = delete;
        
        MessageHandler& operator=( MessageHandler &&rhs) = default;
        MessageHandler( MessageHandler &&source) = default;
    private:
        /* 
         * These can be marked private, as they are not meant to be 
         * called outside the object, just implemented by subclasses.
         * Unfortunately, to get the documentation into our
         * doxoygen pages, we need them to be protected.
         */
#ifdef RUNNING_DOXYGEN /* So that DOXYGEN will document the next two */
    protected:
#endif
 
        virtual Request MessageReceived() = 0;
 
        virtual void ActionEnd(bool taskExiting, StatusType actionEndStatus);
 
    public:
        virtual void MessageUser(const std::string &text) const;
 
 
        template<typename... Types>
        void MessageUser(const char *format, Types... args) {
            
            /*
             * Our approach is to write the output to a string, via
             * SafePrintf(), then output that in one operation.  Since
             * std::ostream devices will then do the output in one operation.
             */
            std::stringstream sstrm;
            drama::SafePrintf(sstrm, format, args...);
            MessageUser(sstrm.str());
            
            
        }
 
 
        virtual const sds::PrintObjectCR &SdsListToUser() const;
 
 
        virtual const EntryInfo & GetEntry()  {
            return _entryInfo;
        }
 
        virtual std::shared_ptr<Task>  GetTask() const {
            return std::shared_ptr<Task>(_theTask);
        }
 
        void SendTrigger(const sds::Id &arg) const;
 
        void SendBulkTrigger(BulkData *arg,
                             DitsTransIdType *transId,
                             bool isSds,
                             unsigned notifyBytes=1024*1024);
 
 
        void SendBulkTrigger(BulkDataSds *arg,
                             DitsTransIdType *transId,
                             unsigned notifyBytes=1024*1024) {
 
            SendBulkTrigger(arg, transId, true, notifyBytes);
        }
 
        std::string GetActionName() const {
            if (!_action)
                DramaTHROW(DRAMA2__INV_CALL_FRM_NONACT,
                           "Tried to  GetActionName when not in action.");
 
            return _action->GetActionName();
        }
 
        void SetReturnArg(const sds::Id &arg, bool copy=true) {
 
            if (!_action)
                DramaTHROW(DRAMA2__INV_CALL_FRM_NONACT,
                           "Tried to  SetReturnArg when not in action.");
 
 
            DitsArgFlagType flag = DITS_ARG_COPY;
            if (copy == false)
                flag = DITS_ARG_NODELETE;
            StatusType status = STATUS__OK;
            DitsPutArgument((SdsIdType)(arg), flag, &status);
            if (status != STATUS__OK)
                DramaTHROW(status,"Error when in call to DitsPutArgument()");
        }
        void SetReturnArg(sds::Id *arg) {
            if (!_action)
                DramaTHROW(DRAMA2__INV_CALL_FRM_NONACT,
                           "Tried to  SetReturnArg when not in action.");
 
            /*
             * We need to work out what sds::Id would have
             * done and use that to set our flag to DitsPutArgument()
             */
            DitsArgFlagType flag = DITS_ARG_NODELETE;
            SdsIdType id;
            bool free;
            bool del;
            bool readfree;
            id = arg->COut(true, &free, &del, &readfree);
            if (del)
                flag = DITS_ARG_DELETE;
            else if (readfree)
                flag = DITS_ARG_READFREE;
            else if (free)
                flag = DITS_ARG_FREEID;
                    
            StatusType status = STATUS__OK;
            DitsPutArgument(id, flag, &status);
            if (status != STATUS__OK)
                DramaTHROW(status,"Error when in call to DitsPutArgument()");
        }
        virtual void PutObeyHandler(MessageHandlerPtr obj) {
            if (!_action)
                DramaTHROW(DRAMA2__INV_CALL_FRM_NONACT,
                           "Tried to change Obey handler when not in action.");
            _action->PutObeyHandler(obj);
        }
        virtual void PutKickHandler(MessageHandlerPtr obj) {
            if (!_action)
                DramaTHROW(DRAMA2__INV_CALL_FRM_NONACT,
                           "Tried to change Kick handler when not in action.");
            _action->PutKickHandler(obj);
        }
 
        virtual void PutObeyHandler(MessageReceiveFunction func);
        virtual void PutKickHandler(MessageReceiveFunction func);
 
    protected:
        void GrabEntryInfo() {
            _entryInfo.SetFromDits();
        }
 
    };
 
    class MessageHandlerViaFunctor : public MessageHandler {
    private:
        const MessageReceiveFunction _func; // The function.
    public:
        MessageHandlerViaFunctor(const MessageReceiveFunction func) : 
            _func(func) {
        }
        drama::Request MessageReceived() override {
            return _func(this);
        }
    };
 
    extern Request SimpleExitAction (drama::MessageHandler *messageHandler);
 
 
    template <class T> 
    class MessageUserStreamBuf : public std::streambuf {
        
    private:
        T  &_actionObj;
    protected:
        // Buffer for messages to output.
        // Size is the maximum size of a DRAMA MsgOut message plus one
        //  for a null terminator.
        char _buffer[MSG_C_LEN+1];
 
        /*
         * Buffer flush operation.  Ensure the buffer is null terminated
         * before calling MessageUser.  Must return number sent and reset
         * the buffer.
         */
        int flushBuffer()  {
            int num = pptr() - pbase();
            *pptr() = '\0';  // Ensure null termination.
            _actionObj.MessageUser(std::string(_buffer));
            pbump(-num);
            return num;
        }
 
        /*
         * Invoked if adding a character would overflow the buffer.
         * We Add it (since our constructor ensured there is sufficient
         * space and then flush the buffer.
         */
        virtual int_type overflow(int_type c)  override {
            if (c != EOF)
            {
                // insert character into buffer
                *pptr() = c;
                pbump(1);
            }
            // flush the buffer
            if (flushBuffer() == EOF)
            {
                return EOF;
            }
            return c;
            
        }
 
        virtual int sync() override {
            if (flushBuffer() == EOF) {
                return -1;
            }
            return 0;
        }
        
    public:
        MessageUserStreamBuf(T &actionObj) : _actionObj(actionObj) {
 
            /*
             * Need to provide our buffer to the std::Streambuf base
             * class.  One character less to ensure the last character
             * causes an overflow.
             */
            setp(_buffer, _buffer+(MSG_C_LEN-1));
        }
        virtual ~MessageUserStreamBuf() {
            /*
             * We want to output what is left, but any exception now
             * (which may be caused by DRAMA failing to send the message
             * for various reasons), if left uncaught, will cause the
             * program to crash (can't throw from destructors), so
             * catch any DRAMA exception and use stderr.
             */
            try 
            {
                sync();
            }
            catch (drama::Exception &e)
            {
                fprintf(stderr,"DRAMA Exception thrown by drama::~MessageUserStreamBuf()\n");
                fprintf(stderr,"%s\n", e.toString().c_str());
                fprintf(stderr,"DRAMA Status = 0x%lX, %s\n",
                        static_cast<long int>(e.dramaStatus()),
                        e.dramaStatusStr().c_str());
                
            }
        }
 
    };  //class  MessageUserStreamBuf
 
#ifdef D2_PROVIDE_LOG_STREAM // Currently disabled.
// MessageUserStream not working under CLang, later versions.  unclear why, but
//  appears related to copy/assignment.  msgoutstream would
//  attempting to use a MessageStream
 
    template <class T> 
    class MessageUserStream : public std::ostream {
    private:
        MessageUserStreamBuf<T> _buf;
    public:
        MessageUserStream(T &actionObj) : _buf(actionObj) {
            /* 
             *  Set the buffer address for ostream.
             */
            rdbuf(&_buf);
        }
        
    }; //class  MessageUserStream.
 
#endif
 
} // namespace drama
#endif