interprocess.h

Go to the documentation of this file.

// Copyright 2016-2024:
//   GobySoft, LLC (2013-)
//   Community contributors (see AUTHORS file)
// File authors:
//   Toby Schneider <toby@gobysoft.org>
//   Ryan Govostes <rgovostes+git@gmail.com>
//
//
// 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_MIDDLEWARE_TRANSPORT_INTERPROCESS_H
#define GOBY_MIDDLEWARE_TRANSPORT_INTERPROCESS_H
 
#include <atomic>
#include <functional>
#include <sys/types.h>
#include <thread>
#include <tuple>
#include <unistd.h>
 
#include "goby/middleware/group.h"
 
#include "goby/middleware/marshalling/interface.h"
#include "goby/middleware/transport/null.h"
#include "goby/middleware/transport/poller.h"
#include "goby/middleware/transport/serialization_handlers.h"
 
namespace goby
{
namespace middleware
{
template <typename Derived, typename InnerTransporter>
class InterProcessTransporterBase
    : public StaticTransporterInterface<InterProcessTransporterBase<Derived, InnerTransporter>,
                                        InnerTransporter>,
      public Poller<InterProcessTransporterBase<Derived, InnerTransporter>>
{
    using InterfaceType =
        StaticTransporterInterface<InterProcessTransporterBase<Derived, InnerTransporter>,
                                   InnerTransporter>;
 
    using PollerType = Poller<InterProcessTransporterBase<Derived, InnerTransporter>>;
 
  public:
    InterProcessTransporterBase(InnerTransporter& inner)
        : InterfaceType(inner), PollerType(&this->inner())
    {
    }
    InterProcessTransporterBase() : PollerType(&this->inner()) {}
 
    virtual ~InterProcessTransporterBase() {}
 
    template <typename Data> static constexpr int scheme()
    {
        int scheme = goby::middleware::scheme<Data>();
        // if default returns DCCL, use PROTOBUF instead
        if (scheme == MarshallingScheme::DCCL)
            scheme = MarshallingScheme::PROTOBUF;
        return scheme;
    }
 
    template <typename Data, int scheme = InterProcessTransporterBase::scheme<Data>()>
    void publish_dynamic(const Data& data, const Group& group,
                         const Publisher<Data>& publisher = Publisher<Data>())
    {
        check_validity_runtime(group);
        static_cast<Derived*>(this)->template _publish<Data, scheme>(data, group, publisher);
        this->inner().template publish_dynamic<Data, scheme>(data, group, publisher);
    }
 
    template <typename Data, int scheme = InterProcessTransporterBase::scheme<Data>()>
    void publish_dynamic(std::shared_ptr<const Data> data, const Group& group,
                         const Publisher<Data>& publisher = Publisher<Data>())
    {
        if (data)
        {
            check_validity_runtime(group);
            static_cast<Derived*>(this)->template _publish<Data, scheme>(*data, group, publisher);
            this->inner().template publish_dynamic<Data, scheme>(data, group, publisher);
        }
    }
 
    template <typename Data, int scheme = InterProcessTransporterBase::scheme<Data>()>
    void publish_dynamic(std::shared_ptr<Data> data, const Group& group,
                         const Publisher<Data>& publisher = Publisher<Data>())
    {
        publish_dynamic<Data, scheme>(std::shared_ptr<const Data>(data), group, publisher);
    }
 
    void publish_serialized(std::string type_name, int scheme, const std::vector<char>& bytes,
                            const goby::middleware::Group& group)
    {
        check_validity_runtime(group);
        static_cast<Derived*>(this)->_publish_serialized(type_name, scheme, bytes, group);
    }
 
    template <typename Data, int scheme = InterProcessTransporterBase::scheme<Data>()>
    void subscribe_dynamic(std::function<void(const Data&)> f, const Group& group,
                           const Subscriber<Data>& subscriber = Subscriber<Data>())
    {
        check_validity_runtime(group);
        static_cast<Derived*>(this)->template _subscribe<Data, scheme>(
            [=](std::shared_ptr<const Data> d) { f(*d); }, group, subscriber);
    }
 
    template <typename Data, int scheme = InterProcessTransporterBase::scheme<Data>()>
    void subscribe_dynamic(std::function<void(std::shared_ptr<const Data>)> f, const Group& group,
                           const Subscriber<Data>& subscriber = Subscriber<Data>())
    {
        check_validity_runtime(group);
        static_cast<Derived*>(this)->template _subscribe<Data, scheme>(f, group, subscriber);
    }
 
    template <typename Data, int scheme = InterProcessTransporterBase::scheme<Data>()>
    void unsubscribe_dynamic(const Group& group,
                             const Subscriber<Data>& subscriber = Subscriber<Data>())
    {
        check_validity_runtime(group);
        static_cast<Derived*>(this)->template _unsubscribe<Data, scheme>(group, subscriber);
    }
 
    void unsubscribe_all() { static_cast<Derived*>(this)->_unsubscribe_all(); }
 
    std::shared_ptr<SerializationSubscriptionRegex>
    subscribe_regex(std::function<void(const std::vector<unsigned char>&, int scheme,
                                       const std::string& type, const Group& group)>
                        f,
                    const std::set<int>& schemes, const std::string& type_regex = ".*",
                    const std::string& group_regex = ".*")
    {
        return static_cast<Derived*>(this)->_subscribe_regex(f, schemes, type_regex, group_regex);
    }
 
    template <typename Data, int scheme = InterProcessTransporterBase::scheme<Data>()>
    std::shared_ptr<SerializationSubscriptionRegex> subscribe_type_regex(
        std::function<void(std::shared_ptr<const Data>, const std::string& type)> f,
        const Group& group, const std::string& type_regex = ".*")
    {
        std::regex special_chars{R"([-[\]{}()*+?.,\^$|#\s])"};
        std::string sanitized_group =
            std::regex_replace(std::string(group), special_chars, R"(\$&)");
 
        auto regex_lambda = [=](const std::vector<unsigned char>& data, int schm,
                                const std::string& type, const Group& grp) {
            auto data_begin = data.begin(), data_end = data.end(), actual_end = data.end();
            auto msg =
                SerializerParserHelper<Data, scheme>::parse(data_begin, data_end, actual_end, type);
            f(msg, type);
        };
 
        return static_cast<Derived*>(this)->_subscribe_regex(regex_lambda, {scheme}, type_regex,
                                                             "^" + sanitized_group + "$");
    }
 
    template <const Group& group, typename Data,
              int scheme = InterProcessTransporterBase::scheme<Data>()>
    void subscribe_type_regex(
        std::function<void(std::shared_ptr<const Data>, const std::string& type)> f,
        const std::string& type_regex = ".*")
    {
        subscribe_type_regex(f, group, type_regex);
    }
 
    template <const Group& group> void check_validity()
    {
        static_assert((group.c_str() != nullptr) && (group.c_str()[0] != '\0'),
                      "goby::middleware::Group must have non-zero length string to publish on the "
                      "InterProcess layer");
    }
 
    void check_validity_runtime(const Group& group)
    {
        if ((group.c_str() == nullptr) || (group.c_str()[0] == '\0'))
            throw(goby::Exception("Group must have a non-empty string for use on InterProcess"));
    }
 
  protected:
    static constexpr Group to_portal_group_{"goby::middleware::interprocess::to_portal"};
    static constexpr Group regex_group_{"goby::middleware::interprocess::regex"};
    static constexpr Group from_portal_group_{"goby::middleware::interprocess::from_portal"};
 
  private:
    friend PollerType;
    int _poll(std::unique_ptr<std::unique_lock<std::timed_mutex>>& lock)
    {
        return static_cast<Derived*>(this)->_poll(lock);
    }
};
 
template <typename Derived, typename InnerTransporter>
constexpr goby::middleware::Group
    InterProcessTransporterBase<Derived, InnerTransporter>::to_portal_group_;
template <typename Derived, typename InnerTransporter>
constexpr goby::middleware::Group
    InterProcessTransporterBase<Derived, InnerTransporter>::regex_group_;
template <typename Derived, typename InnerTransporter>
constexpr goby::middleware::Group
    InterProcessTransporterBase<Derived, InnerTransporter>::from_portal_group_;
 
template <typename InnerTransporter>
class InterProcessForwarder
    : public InterProcessTransporterBase<InterProcessForwarder<InnerTransporter>, InnerTransporter>
{
  public:
    using Base =
        InterProcessTransporterBase<InterProcessForwarder<InnerTransporter>, InnerTransporter>;
 
    InterProcessForwarder(InnerTransporter& inner) : Base(inner)
    {
        this->inner()
            .template subscribe<Base::regex_group_,
                                goby::middleware::protobuf::SerializerTransporterMessage>(
                [this](
                    std::shared_ptr<const goby::middleware::protobuf::SerializerTransporterMessage>
                        msg) { _receive_regex_data_forwarded(msg); });
    }
    virtual ~InterProcessForwarder()
    {
        this->unsubscribe_all();
 
        // TODO - remove by adding in an explicit handshake with the unsubscribe_all publication so that we don't delete ourself (and thus our inner()) before the Portal has deleted all the subscriptions
        usleep(1e5);
    }
 
    friend Base;
 
  private:
    template <typename Data, int scheme>
    void _publish(const Data& d, const Group& group, const Publisher<Data>& publisher)
    {
        // create and forward publication to edge
        std::vector<char> bytes(SerializerParserHelper<Data, scheme>::serialize(d));
        std::string* sbytes = new std::string(bytes.begin(), bytes.end());
        auto msg = std::make_shared<goby::middleware::protobuf::SerializerTransporterMessage>();
        auto* key = msg->mutable_key();
 
        key->set_marshalling_scheme(scheme);
        key->set_type(SerializerParserHelper<Data, scheme>::type_name(d));
        key->set_group(std::string(group));
        msg->set_allocated_data(sbytes);
 
        *key->mutable_cfg() = publisher.cfg();
 
        this->inner().template publish<Base::to_portal_group_>(msg);
    }
 
    void _publish_serialized(std::string type_name, int scheme, const std::vector<char>& bytes,
                             const goby::middleware::Group& group)
    {
        auto msg = std::make_shared<goby::middleware::protobuf::SerializerTransporterMessage>();
        auto* key = msg->mutable_key();
 
        key->set_marshalling_scheme(scheme);
        key->set_type(type_name);
        key->set_group(std::string(group));
        msg->set_data(std::string(bytes.begin(), bytes.end()));
 
        this->inner().template publish<Base::to_portal_group_>(msg);
    }
 
    template <typename Data, int scheme>
    void _subscribe(std::function<void(std::shared_ptr<const Data> d)> f, const Group& group,
                    const Subscriber<Data>& subscriber)
    {
        this->inner().template subscribe_dynamic<Data, scheme>(f, group);
 
        // forward subscription to edge
        auto inner_publication_lambda = [=](std::shared_ptr<const Data> d) {
            this->inner().template publish_dynamic<Data, scheme>(d, group);
        };
 
        auto subscription = std::make_shared<SerializationSubscription<Data, scheme>>(
            inner_publication_lambda, group,
            middleware::Subscriber<Data>(goby::middleware::protobuf::TransporterConfig(),
                                         [=](const Data& d) { return group; }));
 
        this->inner().template publish<Base::to_portal_group_, SerializationHandlerBase<>>(
            subscription);
    }
 
    template <typename Data, int scheme>
    void _unsubscribe(const Group& group, const Subscriber<Data>& subscriber)
    {
        this->inner().template unsubscribe_dynamic<Data, scheme>(group, subscriber);
 
        auto unsubscription = std::shared_ptr<SerializationHandlerBase<>>(
            new SerializationUnSubscription<Data, scheme>(group));
 
        this->inner().template publish<Base::to_portal_group_, SerializationHandlerBase<>>(
            unsubscription);
    }
 
    void _unsubscribe_all()
    {
        regex_subscriptions_.clear();
        auto all = std::make_shared<SerializationUnSubscribeAll>();
        this->inner().template publish<Base::to_portal_group_, SerializationUnSubscribeAll>(all);
    }
 
    std::shared_ptr<SerializationSubscriptionRegex>
    _subscribe_regex(std::function<void(const std::vector<unsigned char>&, int scheme,
                                        const std::string& type, const Group& group)>
                         f,
                     const std::set<int>& schemes, const std::string& type_regex = ".*",
                     const std::string& group_regex = ".*")
    {
        auto inner_publication_lambda = [=](const std::vector<unsigned char>& data, int scheme,
                                            const std::string& type, const Group& group) {
            std::shared_ptr<goby::middleware::protobuf::SerializerTransporterMessage>
                forwarded_data(new goby::middleware::protobuf::SerializerTransporterMessage);
            forwarded_data->mutable_key()->set_marshalling_scheme(scheme);
            forwarded_data->mutable_key()->set_type(type);
            forwarded_data->mutable_key()->set_group(group);
            forwarded_data->set_data(std::string(data.begin(), data.end()));
            this->inner().template publish<Base::regex_group_>(forwarded_data);
        };
 
        auto portal_subscription = std::make_shared<SerializationSubscriptionRegex>(
            inner_publication_lambda, schemes, type_regex, group_regex);
        this->inner().template publish<Base::to_portal_group_, SerializationSubscriptionRegex>(
            portal_subscription);
 
        auto local_subscription = std::shared_ptr<SerializationSubscriptionRegex>(
            new SerializationSubscriptionRegex(f, schemes, type_regex, group_regex));
        regex_subscriptions_.insert(local_subscription);
        return local_subscription;
    }
 
    void _receive_regex_data_forwarded(
        std::shared_ptr<const goby::middleware::protobuf::SerializerTransporterMessage> msg)
    {
        const auto& bytes = msg->data();
        for (auto& sub : regex_subscriptions_)
            sub->post(bytes.begin(), bytes.end(), msg->key().marshalling_scheme(),
                      msg->key().type(), msg->key().group());
    }
 
    int _poll(std::unique_ptr<std::unique_lock<std::timed_mutex>>& lock)
    {
        return 0;
    } // A forwarder is a shell, only the inner Transporter has data
 
  private:
    std::set<std::shared_ptr<const SerializationSubscriptionRegex>> regex_subscriptions_;
};
 
template <typename Derived, typename InnerTransporter>
class InterProcessPortalBase : public InterProcessTransporterBase<Derived, InnerTransporter>
{
  public:
    using Base = InterProcessTransporterBase<Derived, InnerTransporter>;
 
    InterProcessPortalBase(InnerTransporter& inner) : Base(inner) { _init(); }
    InterProcessPortalBase() { _init(); }
 
    virtual ~InterProcessPortalBase() {}
 
  private:
    void _init()
    {
        using goby::middleware::protobuf::SerializerTransporterMessage;
        this->inner().template subscribe<Base::to_portal_group_, SerializerTransporterMessage>(
            [this](std::shared_ptr<const SerializerTransporterMessage> d) {
                static_cast<Derived*>(this)->_receive_publication_forwarded(*d);
            });
 
        this->inner().template subscribe<Base::to_portal_group_, SerializationHandlerBase<>>(
            [this](std::shared_ptr<const middleware::SerializationHandlerBase<>> s) {
                static_cast<Derived*>(this)->_receive_subscription_forwarded(s);
            });
 
        this->inner().template subscribe<Base::to_portal_group_, SerializationSubscriptionRegex>(
            [this](std::shared_ptr<const middleware::SerializationSubscriptionRegex> s) {
                static_cast<Derived*>(this)->_receive_regex_subscription_forwarded(s);
            });
 
        this->inner().template subscribe<Base::to_portal_group_, SerializationUnSubscribeAll>(
            [this](std::shared_ptr<const middleware::SerializationUnSubscribeAll> s) {
                static_cast<Derived*>(this)->_unsubscribe_all(s->subscriber_id());
            });
    }
};
 
} // namespace middleware
} // namespace goby
 
#endif