jaus++/2.0/include/jaus/core/transport/transport.h

Go to the documentation of this file.

#ifndef __JAUS_CORE_TRANSPORT__H
#define __JAUS_CORE_TRANSPORT__H

#include "jaus/core/service.h"
#include "jaus/core/time.h"
#include <set>
#include <map>
#include <cxutils/circulararray.h>
#include <fstream>

namespace JAUS
{
    class JAUS_CORE_DLL Transport : public Service
    {
    public:        
        class JAUS_CORE_DLL Callback
        {
        public:
            Callback() {}
            virtual ~Callback() {}
            typedef std::set<Callback* > Set;
            typedef std::map<UShort, Set > Map;
            virtual void ProcessMessage(const Message* message) {};
        };
        static const std::string Name;                  
        static const unsigned int MaxQueueSize = 1000;  
        Transport();
        virtual ~Transport();
        // Closes a connection to a given component ID.
        virtual bool CloseConnection(const Address& id) { return false; }
        // Transport Service does not response to any messages.
        virtual bool IsDiscoverable() const { return false; }
        // Sets the maximum number of threads for message processing.
        void SetMaxMessageProcessingThreads(const unsigned int maxThreads = 2);
        // Add a message type to be processed within its own thread.
        void AddPriorityMessage(const UShort messageCode);
        // Sets the delay time in ms for message polling loop.
        virtual void SetMessagePollingDelayMs(const unsigned int delayTimeMs = 1);
        // Initializes the transport with a given ID for a component.
        virtual bool Initialize(const Address& componentID) = 0;
        // Returns true if transport has been initialized.
        virtual bool IsInitialized() const = 0;
        // Shutsdown the transport service.
        virtual void Shutdown() = 0;
        // Gets the list of connections available.
        virtual Address::List GetConnections() const = 0;
        // Returns true if connection is present.
        virtual bool HaveConnection(const Address& id) const = 0;
        // Sends to all children.
        virtual void Receive(const Message* message) { PushMessageToChildren(message); }
        // Send a serialized message.
        virtual bool SendPacket(const Packet& packet, 
                                const Header& header, 
                                const int broadcastFlags = NoBroadcast) const = 0;
        // Serialize the JAUS message (add transport headers, serialize payload, etc.) 
        virtual bool SerializeMessage(const Message* message, 
                                      Packet::List& stream,
                                      Header::List& streamHeaders,
                                      const UShort startingSequenceNumber,
                                      const int broadcastFlags) const = 0;
        // Get a copy of the transport header.
        virtual Packet GetTransportHeader() const = 0;
        // Gets a list of all manually created connections.
        virtual Address::List GetManualConnections() const = 0;
        // Searches inheriting Services (child Services) to create a message for processing.
        virtual Message* CreateMessage(const UShort messageCode) const;
        // Sends a message.
        virtual bool Send(const Message* message, const int broadcastFlags = Service::NoBroadcast) const;
        // Sends a message, then waits for a response.
        virtual bool Send(const Message* message, 
                          Message* response, 
                          const unsigned int waitTimeMs = Service::DefaultWaitMs) const;
        // Sends a message and waits for one of multiple responses.
        virtual bool Send(const Message* message,
                          Message::List& possibleResponses,
                          const unsigned int waitTimeMs = Service::DefaultWaitMs) const;
        // Register to receive copies of messages when received by Transport.
        void RegisterCallback(const UShort messageCode, Callback* callback);
        // Turns logging on/off.
        void EnableLogging(const bool flag = true);
        // Recursively try to create the message.
        Message* CreateMessageFromService(const UShort messageCode, const Service* service) const;
        // Reads the payload data which includes a message code, and converts to Message structure.
        Message* CreateMessageFromPacket(const Packet& packet) const;
        // Triggers any callbacks for a message.
        void TriggerMessageCallbacks(const Message* message);
        // Adds new message template.
        void AddMessageTemplate(Message* message);
        // Create a message using templates.
        Message* GetMessageFromTemplate(const UShort messageCode) const;
    protected:
        virtual void ProcessPacket(const Packet& packet, const Header& header);
    private:
        class Receipt
        {
        public:
            typedef std::set<Receipt*> Set; 
            Receipt();
            ~Receipt();
            volatile bool mPendingFlag;     
            const Message* mpMessage;       
            Message::List* mpResponses;     
            Time::Stamp mSendTimeMs;        
            Time::Stamp mUpdateTimeMs;      
        };       
        bool CheckPendingReceipts(const Header& header, const UShort messageCode, const Packet& packet);
        bool CheckPendingReceipts(const LargeDataSet* stream);
        static void MessageProcessingThread(void* args);
        static void PriorityMessageProcessingThread(void* args);
        void WriteToLog(const Packet& packet, const Header& header);
        bool IsPriority(Message** message);
        volatile bool mLoggingFlag;                             
        Mutex mLogFileMutex;                                    
        std::fstream mLogFile;                                  
        std::map<UShort, std::string> mMessageNames;            
        volatile bool mStopMessageProcessingFlag;               
        Thread::Manager mMessageProcessingThreads;              
        Mutex mJausMessageQueueMutex;                           
        CxUtils::CircularArray<Packet> mJausMessageQueue;       
        Mutex mPriorityJausMessageMutex;                        
        std::set<UShort> mPriorityJausMessageTypes;             
        std::map<Address, Message::Map > mPriorityJausMessages; 
        Mutex mMessageTemplatesMutex;                           
        std::map<UShort, Message*> mMessageTemplates;           
        Thread mPriorityJausMessagesThread;                     
        Mutex mLargeDataSetsMutex;                              
        LargeDataSet::Map mLargeDataSets;                       
        unsigned int mNumberOfMessageProcessingThreads;         
        volatile unsigned int mMessageProcessingDelayMs;        
        Mutex mPendingReceiptsMutex;                            
        Receipt::Set mPendingReceipts;                          
        Mutex mCallbacksMutex;                                  
        Callback::Map mMessageCallbacks;                        
        Mutex mSequenceNumberMutex;                             
        UShort mSequenceNumber;                                 
    };
}

#endif
/*  End of File */