queue_manager.h

// Copyright 2009-2018 Toby Schneider (http://gobysoft.org/index.wt/people/toby)
//                     GobySoft, LLC (2013-)
//                     Massachusetts Institute of Technology (2007-2014)
//                     Community contributors (see AUTHORS file)
//
//
// 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 QueueManager20091204_H
#define QueueManager20091204_H

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

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

#include <deque>
#include <map>

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

namespace goby
{
namespace acomms
{
class QueueManager
{
  public:
    QueueManager();
    ~QueueManager() {}



    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&);
    QueueManager& operator=(const QueueManager&);

    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, boost::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_;
    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
        {
            typedef std::multimap<unsigned, goby::acomms::protobuf::Manipulator>::const_iterator
                iterator;
            std::pair<iterator, iterator> p = manips_.equal_range(id);

            for (iterator 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