queue_manager.h

Go to the documentation of this file.

// Copyright 2009-2025:
//   GobySoft, LLC (2013-)
//   Massachusetts Institute of Technology (2007-2014)
//   Community contributors (see AUTHORS file)
// File authors:
//   Toby Schneider <toby@gobysoft.org>
//
//
// This file is part of the Goby Underwater Autonomy Project Libraries
// ("The Goby Libraries").
//
// The Goby Libraries are free software: you can redistribute them and/or modify
// them under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or
// (at your option) any later version.
//
// The Goby Libraries are distributed in the hope that they will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with Goby.  If not, see <http://www.gnu.org/licenses/>.
 
#ifndef GOBY_ACOMMS_QUEUE_QUEUE_MANAGER_H
#define GOBY_ACOMMS_QUEUE_QUEUE_MANAGER_H
 
#include <iosfwd>  // for ostream
#include <list>    // for list
#include <map>     // for allocator, multimap
#include <memory>  // for shared_ptr, __sha...
#include <set>     // for set
#include <string>  // for string, operator+
#include <utility> // for pair, make_pair
 
#include <boost/signals2/signal.hpp>    // for signal
#include <google/protobuf/descriptor.h> // for Descriptor
#include <google/protobuf/message.h>    // for Message
 
#include "goby/acomms/dccl/dccl.h"               // for DCCLCodec
#include "goby/acomms/protobuf/manipulator.pb.h" // for Manipulator
#include "goby/acomms/protobuf/network_ack.pb.h" // for NetworkAck, Netwo...
#include "goby/acomms/protobuf/queue.pb.h"       // for QueuedMessageEntr...
#include "goby/util/as.h"                        // for as
 
#include "queue.h"           // for Queue
#include "queue_exception.h" // for QueueException
 
namespace goby
{
namespace acomms
{
namespace protobuf
{
class ModemTransmission;
} // namespace protobuf
 
class QueueManager
{
  public:
    QueueManager();
    ~QueueManager() = default;
 
 
 
    void set_cfg(const protobuf::QueueManagerConfig& cfg);
 
    void merge_cfg(const protobuf::QueueManagerConfig& cfg);
 
    template <typename ProtobufMessage>
    void add_queue(const protobuf::QueuedMessageEntry& queue_cfg)
    {
        add_queue(ProtobufMessage::descriptor(), queue_cfg);
    }
 
    void add_queue(const google::protobuf::Descriptor* desc,
                   const protobuf::QueuedMessageEntry& queue_cfg); //@}
 
 
 
    void push_message(const google::protobuf::Message& new_message);
    void push_message(const google::protobuf::Message& new_message,
                      const protobuf::QueuedMessageMeta* meta);
 
    void flush_queue(const protobuf::QueueFlush& flush);
 
 
 
    void handle_modem_data_request(protobuf::ModemTransmission* msg);
 
    void handle_modem_receive(const protobuf::ModemTransmission& message);
 
 
 
 
    void do_work();
 
 
 
 
    void info_all(std::ostream* os) const;
 
    template <typename ProtobufMessage> void info(std::ostream* os) const
    {
        info(ProtobufMessage::descriptor(), os);
    }
 
    void info(const google::protobuf::Descriptor* desc, std::ostream* os) const;
 
    const std::string& glog_push_group() { return glog_push_group_; }
    const std::string& glog_pop_group() { return glog_pop_group_; }
    const std::string& glog_priority_group() { return glog_priority_group_; }
    const std::string& glog_out_group() { return glog_out_group_; }
    const std::string& glog_in_group() { return glog_in_group_; }
 
    std::string msg_string(const google::protobuf::Descriptor* desc)
    {
        return desc->full_name() + " (" + goby::util::as<std::string>(codec_->id(desc)) + ")";
    }
 
    int modem_id() { return modem_id_; }
 
    protobuf::QueuedMessageMeta meta_from_msg(const google::protobuf::Message& msg)
    {
        unsigned dccl_id = codec_->id(msg.GetDescriptor());
        if (!queues_.count(dccl_id))
            throw(QueueException("No such queue [[" + msg.GetDescriptor()->full_name() +
                                 "]] loaded"));
 
        return queues_[dccl_id]->meta_from_msg(msg);
    }
 
 
 
 
    boost::signals2::signal<void(const protobuf::ModemTransmission& ack_msg,
                                 const google::protobuf::Message& orig_msg)>
        signal_ack;
 
    boost::signals2::signal<void(const google::protobuf::Message& msg)> signal_receive;
 
    boost::signals2::signal<void(const google::protobuf::Message& orig_msg)> signal_expire;
 
    boost::signals2::signal<void(const protobuf::ModemTransmission& request_msg,
                                 google::protobuf::Message* data_msg)>
        signal_data_on_demand;
 
    boost::signals2::signal<void(protobuf::QueueSize size)> signal_queue_size_change;
 
    boost::signals2::signal<void(protobuf::QueuedMessageMeta* meta,
                                 const google::protobuf::Message& data_msg, int modem_id)>
        signal_out_route;
 
    boost::signals2::signal<void(const protobuf::QueuedMessageMeta& meta,
                                 const google::protobuf::Message& data_msg, int modem_id)>
        signal_in_route;
 
  private:
    QueueManager(const QueueManager&) = delete;
    QueueManager& operator=(const QueueManager&) = delete;
 
    void qsize(Queue* q);
 
    // finds the %queue with the highest priority
    Queue* find_next_sender(const protobuf::ModemTransmission& message, const std::string& data,
                            bool first_user_frame);
 
    // clears the destination and ack values for the packet to reset for next $CADRQ
    void clear_packet(const protobuf::ModemTransmission& message);
    void process_cfg();
 
    void process_modem_ack(const protobuf::ModemTransmission& ack_msg);
    void create_network_ack(int ack_src, const google::protobuf::Message& orig_msg,
                            goby::acomms::protobuf::NetworkAck::AckType ack_type);
 
    // "overload" those from DCCLCodec to allow changing of crypto passphrase
    std::string encode_repeated(const std::list<QueuedMessage>& msgs);
    std::list<QueuedMessage> decode_repeated(const std::string& orig_bytes);
    unsigned size_repeated(const std::list<QueuedMessage>& msgs);
 
  private:
    friend class Queue;
    int modem_id_;
    std::map<unsigned, std::shared_ptr<Queue> > queues_;
 
    // 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_{0};
    int packet_dest_;
 
    typedef int ModemId;
    std::set<ModemId> network_ack_src_ids_;
    std::set<ModemId> route_additional_modem_ids_;
 
    // maps id to crypto passphrase
    std::map<ModemId, std::string> encrypt_rules_;
 
    protobuf::QueueManagerConfig cfg_;
 
    goby::acomms::DCCLCodec* codec_;
 
    std::string glog_push_group_;
    std::string glog_pop_group_;
    std::string glog_priority_group_;
    std::string glog_out_group_;
    std::string glog_in_group_;
 
    static int count_;
 
    class ManipulatorManager
    {
      public:
        void add(unsigned id, goby::acomms::protobuf::Manipulator manip)
        {
            manips_.insert(std::make_pair(id, manip));
        }
 
        bool has(unsigned id, goby::acomms::protobuf::Manipulator manip) const
        {
            using iterator =
                std::multimap<unsigned int, goby::acomms::protobuf::Manipulator>::const_iterator;
            std::pair<iterator, iterator> p = manips_.equal_range(id);
 
            for (auto it = p.first; it != p.second; ++it)
            {
                if (it->second == manip)
                    return true;
            }
 
            return false;
        }
 
        void clear() { manips_.clear(); }
 
      private:
        // manipulator multimap (no_encode, no_decode, etc)
        // maps DCCL ID (unsigned) onto Manipulator enumeration (xml_config.proto)
        std::multimap<unsigned, goby::acomms::protobuf::Manipulator> manips_;
    };
 
    ManipulatorManager manip_manager_;
};
 
std::ostream& operator<<(std::ostream& out, const QueueManager& d);
 
} // namespace acomms
 
} // namespace goby
 
#endif