acomms/libqueue/queue_manager.h

// copyright 2009 t. schneider tes@mit.edu 
//
// this file is part of the Queue Library (libqueue),
// the goby-acomms message queue manager. goby-acomms is a collection of 
// libraries for acoustic underwater networking
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This software is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this software.  If not, see <http://www.gnu.org/licenses/>.


#ifndef QueueManager20091204_H
#define QueueManager20091204_H

#include <limits>
#include <set>
#include <boost/bind.hpp>
#include <boost/signal.hpp>

#include "goby/acomms/dccl.h"
#include "goby/protobuf/queue.pb.h"
#include "goby/protobuf/acomms_proto_helpers.h"

#include <map>
#include <deque>

#include "queue_exception.h"
#include "queue.h"


namespace goby
{
    namespace util
    {
        class FlexOstream;
    }
        
    
    namespace acomms
    {
        class QueueManager
        {
          public:




            QueueManager(std::ostream* log = 0);

            ~QueueManager() { }
        
        


            void set_cfg(const protobuf::QueueManagerConfig& cfg);

            void merge_cfg(const protobuf::QueueManagerConfig& cfg);

            


            static void add_flex_groups(util::FlexOstream* tout);
            
            



            void push_message(const protobuf::ModemDataTransmission& new_message);        

            void flush_queue(const protobuf::QueueFlush& flush);
        


            void handle_modem_data_request(const protobuf::ModemDataRequest& msg_request, protobuf::ModemDataTransmission* msg_data);

            void handle_modem_receive(const protobuf::ModemDataTransmission& message);
        
            void handle_modem_ack(const protobuf::ModemDataAck& message);
        




            void do_work();

        


            std::string summary() const;
            const ManipulatorManager& manip_manager() const { return manip_manager_; }

            
            
        




            boost::signal<void (const protobuf::ModemDataAck& ack_msg)> signal_ack;
            boost::signal<void (const protobuf::ModemDataTransmission& msg)> signal_receive;

            boost::signal<void (const protobuf::ModemDataTransmission& msg)> signal_receive_ccl;
            
            boost::signal<void (const protobuf::ModemDataExpire& expire_msg)> signal_expire;
            
            boost::signal<void (const protobuf::ModemDataRequest& request_msg,
                                protobuf::ModemDataTransmission* data_msg)> signal_data_on_demand;

            boost::signal<void (protobuf::QueueSize size)> signal_queue_size_change;





            
          private:
            friend class Queue;
            static int modem_id_;
            

            std::set<unsigned> add_xml_queue_file(const std::string& xml_file);

            void add_queue(const protobuf::QueueConfig& cfg);
        
        
            
            void qsize(Queue* q);
        
            // finds the %queue with the highest priority
            Queue* find_next_sender(const protobuf::ModemDataRequest& message, const protobuf::ModemDataTransmission& data_msg, bool first_user_frame);
        
            // combine multiple "user" frames into a single "modem" frame
            bool stitch_recursive(const protobuf::ModemDataRequest& request_msg, protobuf::ModemDataTransmission* data_msg, Queue* winning_var);
            bool unstitch_recursive(std::string* data, protobuf::ModemDataTransmission* message);

            void replace_header(bool is_last_user_frame, protobuf::ModemDataTransmission* data_msg, const protobuf::ModemDataTransmission& next_data_msg, const std::string& new_data);
       
            // clears the destination and ack values for the packet to reset for next $CADRQ
            void clear_packet();
            
            // slave function to receive_incoming_modem_data that actually writes a piece of the message (called for each user-frame)
            bool publish_incoming_piece(protobuf::ModemDataTransmission* message, const unsigned incoming_var_id);
            
        
            void process_cfg();
            
          private:
            std::map<goby::acomms::protobuf::QueueKey, Queue> queues_;
            
            std::ostream* log_;

            // map frame number onto %queue pointer that contains
            // the data for this ack
            std::multimap<unsigned, Queue*> waiting_for_ack_;

            // the first *user* frame sets the tone (dest & ack) for the entire packet (all %modem frames)
            unsigned packet_ack_;
            int packet_dest_;
 
            protobuf::QueueManagerConfig cfg_;
            
            ManipulatorManager manip_manager_;
        };

        std::ostream& operator<< (std::ostream& out, const QueueManager& d);

        
    }

}


#endif