interface.h

Go to the documentation of this file.

// Copyright 2017-2026:
//   GobySoft, LLC (2013-)
//   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_MIDDLEWARE_TRANSPORT_DETAIL_INTERFACE_H
#define GOBY_MIDDLEWARE_TRANSPORT_DETAIL_INTERFACE_H
 
#include <chrono>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <vector>
 
#include "goby/middleware/group.h"
#include "goby/middleware/marshalling/interface.h"
 
#include "goby/exception.h"
#include "goby/middleware/marshalling/detail/primitive_type.h"
#include "goby/middleware/protobuf/intervehicle.pb.h"
#include "goby/middleware/protobuf/transporter_config.pb.h"
#include "goby/middleware/transport/detail/type_helpers.h"
#include "goby/middleware/transport/publisher.h"
#include "goby/middleware/transport/subscriber.h"
#include "goby/util/debug_logger.h"
 
namespace goby
{
namespace middleware
{
class NullTransporter;
 
template <typename Transporter, typename InnerTransporter, typename Enable = void>
class InnerTransporterInterface
{
  public:
    using InnerTransporterType = InnerTransporter;
    InnerTransporter& inner()
    {
        static_assert(std::is_void<Enable>::value, "InnerTransporterInterface must be specialized");
    }
    auto innermost()
    {
        static_assert(std::is_void<Enable>::value, "InnerTransporterInterface must be specialized");
    }
};
 
template <typename Transporter, typename InnerTransporter>
class InnerTransporterInterface<
    Transporter, InnerTransporter,
    typename std::enable_if_t<!std::is_same<Transporter, NullTransporter>::value &&
                              !std::is_same<InnerTransporter, NullTransporter>::value>>
{
    using Self = InnerTransporterInterface<
        Transporter, InnerTransporter,
        typename std::enable_if_t<!std::is_same<Transporter, NullTransporter>::value &&
                                  !std::is_same<InnerTransporter, NullTransporter>::value>>;
 
  public:
    InnerTransporter& inner() { return inner_; }
    auto& innermost() { return inner_.innermost(); }
 
    using InnerTransporterType = InnerTransporter;
 
  protected:
    InnerTransporterInterface(InnerTransporter& inner) : inner_(inner) {}
    InnerTransporterInterface() : own_inner_(new InnerTransporter), inner_(*own_inner_) {}
 
  private:
    std::shared_ptr<InnerTransporter> own_inner_;
    InnerTransporter& inner_;
};
 
template <typename Transporter, typename InnerTransporter>
class InnerTransporterInterface<
    Transporter, InnerTransporter,
    typename std::enable_if_t<!std::is_same<Transporter, NullTransporter>::value &&
                              std::is_same<InnerTransporter, NullTransporter>::value>>
{
  public:
    using InnerTransporterType = InnerTransporter;
    InnerTransporter& inner() { return inner_; }
    Transporter& innermost() { return *static_cast<Transporter*>(this); }
 
  protected:
    InnerTransporterInterface(InnerTransporter& inner) : inner_(inner) {}
    InnerTransporterInterface() : own_inner_(new InnerTransporter), inner_(*own_inner_) {}
 
  private:
    std::shared_ptr<InnerTransporter> own_inner_;
    InnerTransporter& inner_;
};
 
template <typename Transporter, typename InnerTransporter>
class InnerTransporterInterface<
    Transporter, InnerTransporter,
    typename std::enable_if_t<std::is_same<Transporter, NullTransporter>::value &&
                              std::is_same<InnerTransporter, NullTransporter>::value>>
{
};
 
class PollerInterface
{
  public:
    template <class Clock = std::chrono::system_clock, class Duration = typename Clock::duration>
    int poll(const std::chrono::time_point<Clock, Duration>& timeout =
                 std::chrono::time_point<Clock, Duration>::max());
 
    template <class Clock = std::chrono::system_clock, class Duration = typename Clock::duration>
    int poll(Duration wait_for);
 
    std::shared_ptr<std::mutex> poll_mutex() { return poll_mutex_; }
 
    std::shared_ptr<std::condition_variable> cv() { return cv_; }
 
    void attach(PollerInterface* poller)
    {
        if (!poller)
            throw(goby::Exception("Cannot attach a null PollerInterface"));
 
        if (poller == this)
            throw(goby::Exception("Cannot attach a PollerInterface to itself"));
 
        if (poller->cv() != cv() || poller->poll_mutex() != poll_mutex())
            throw(goby::Exception("Cannot attach PollerInterface with a different cv() and/or "
                                  "poll_mutex(). Make sure the PollerInterface you are trying to "
                                  "attach has the same innermost PollerInterface"));
 
        std::lock_guard<std::mutex> lock(*poll_mutex_);
        if (std::find(attached_pollers_.begin(), attached_pollers_.end(), poller) !=
            attached_pollers_.end())
            throw(goby::Exception("Cannot attach the same PollerInterface more than once"));
        attached_pollers_.push_back(poller);
    }
 
  protected:
    PollerInterface(std::shared_ptr<std::mutex> poll_mutex,
                    std::shared_ptr<std::condition_variable> cv)
        : poll_mutex_(poll_mutex), cv_(cv)
    {
    }
 
  private:
    template <typename Transporter> friend class Poller;
    // poll the transporter for data
    virtual int _transporter_poll(std::unique_ptr<std::unique_lock<std::mutex>>& lock) = 0;
 
  private:
    // poll all the transporters for data, including a timeout (only called by the outside-most Poller)
    template <class Clock = std::chrono::system_clock, class Duration = typename Clock::duration>
    int _poll_all(const std::chrono::time_point<Clock, Duration>& timeout);
 
    std::shared_ptr<std::mutex> poll_mutex_;
    // signaled when there's no data for this thread to read during _poll()
    std::shared_ptr<std::condition_variable> cv_;
    // non-owning pointers to additional PollerInterface instances to poll alongside this one;
    // attached pollers must remain valid for the lifetime of this instance
    std::vector<PollerInterface*> attached_pollers_;
};
 
enum class Necessity
{
    REQUIRED,
    RECOMMENDED,
    OPTIONAL
};
 
template <typename Transporter, typename InnerTransporter>
class StaticTransporterInterface : public InnerTransporterInterface<Transporter, InnerTransporter>
{
  public:
    template <const Group& group, typename Data,
              int scheme = transporter_scheme<Data, Transporter>()>
    void publish(const Data& data, const Publisher<Data>& publisher = Publisher<Data>())
    {
        static_cast<Transporter*>(this)->template check_validity<group>();
        static_cast<Transporter*>(this)->template publish_dynamic<Data, scheme>(data, group,
                                                                                publisher);
    }
 
    template <const Group& group, typename Data,
              int scheme = transporter_scheme<Data, Transporter>()>
    void publish(std::shared_ptr<const Data> data,
                 const Publisher<Data>& publisher = Publisher<Data>())
    {
        static_cast<Transporter*>(this)->template check_validity<group>();
        static_cast<Transporter*>(this)->template publish_dynamic<Data, scheme>(data, group,
                                                                                publisher);
    }
 
    template <const Group& group, typename Data,
              int scheme = transporter_scheme<Data, Transporter>()>
    void publish(std::shared_ptr<Data> data, const Publisher<Data>& publisher = Publisher<Data>())
    {
        publish<group, Data, scheme>(std::shared_ptr<const Data>(data), publisher);
    }
 
    template <const Group& group, typename Data,
              int scheme = transporter_scheme<Data, Transporter>(),
              Necessity necessity = Necessity::OPTIONAL>
    void subscribe(std::function<void(const Data&)> f,
                   const Subscriber<Data>& subscriber = Subscriber<Data>())
    {
        static_cast<Transporter*>(this)->template check_validity<group>();
        static_cast<Transporter*>(this)->template subscribe_dynamic<Data, scheme>(f, group,
                                                                                  subscriber);
    }
 
    template <const Group& group, typename Data,
              int scheme = transporter_scheme<Data, Transporter>(),
              Necessity necessity = Necessity::OPTIONAL>
    void subscribe(std::function<void(std::shared_ptr<const Data>)> f,
                   const Subscriber<Data>& subscriber = Subscriber<Data>())
    {
        static_cast<Transporter*>(this)->template check_validity<group>();
        static_cast<Transporter*>(this)->template subscribe_dynamic<Data, scheme>(f, group,
                                                                                  subscriber);
    }
 
    template <const Group& group, Necessity necessity = Necessity::OPTIONAL, typename Func>
    void subscribe(Func f)
    {
        // we want to grab the first argument of "f" and then capture "Data" from "const Data& data" and "std::shared_ptr<const Data>"
        using Data = typename detail::primitive_type<
            typename std::decay<detail::first_argument<Func>>::type>::type;
 
        subscribe<group, Data, transporter_scheme<Data, Transporter>(), necessity>(f);
    }
 
    template <const Group& group, typename Data,
              int scheme = transporter_scheme<Data, Transporter>()>
    void unsubscribe(const Subscriber<Data>& subscriber = Subscriber<Data>())
    {
        static_cast<Transporter*>(this)->template check_validity<group>();
        static_cast<Transporter*>(this)->template unsubscribe_dynamic<Data, scheme>(group,
                                                                                    subscriber);
    }
 
    void unsubscribe_all() { static_cast<Transporter*>(this)->template unsubscribe_all(); }
 
  protected:
    StaticTransporterInterface(InnerTransporter& inner)
        : InnerTransporterInterface<Transporter, InnerTransporter>(inner)
    {
    }
    StaticTransporterInterface() {}
};
 
} // namespace middleware
} // namespace goby
 
template <class Clock, class Duration>
int goby::middleware::PollerInterface::poll(const std::chrono::time_point<Clock, Duration>& timeout)
{
    return _poll_all(timeout);
}
 
template <class Clock, class Duration>
int goby::middleware::PollerInterface::poll(Duration wait_for)
{
    if (wait_for == Duration::max())
        return poll();
    else
        return poll(Clock::now() + wait_for);
}
 
template <class Clock, class Duration>
int goby::middleware::PollerInterface::_poll_all(
    const std::chrono::time_point<Clock, Duration>& timeout)
{
    // hold this lock until either we find a polled item or we wait on the condition variable
    std::unique_ptr<std::unique_lock<std::mutex>> lock(
        new std::unique_lock<std::mutex>(*poll_mutex_));
    //    std::cout << std::this_thread::get_id() <<  " _poll_all locking: " << poll_mutex_.get() << std::endl;
 
    auto poll_all_once = [this, &lock]()
    {
        int poll_items = _transporter_poll(lock);
        for (auto* poller : attached_pollers_) poll_items += poller->_transporter_poll(lock);
        return poll_items;
    };
    int poll_items = poll_all_once();
    while (poll_items == 0)
    {
        if (!lock)
            throw(goby::Exception(
                "Poller lock was released by poll() but no poll items were returned"));
 
        if (timeout == Clock::time_point::max())
        {
            cv_->wait(*lock); // wait_until doesn't work well with time_point::max()
            poll_items = poll_all_once();
 
            // TODO: fix this message now that zeromq::InterProcessPortal can have
            // a condition_variable trigger for REQUEST_HOLD_STATE
            //if (poll_items == 0)
            //                goby::glog.is(goby::util::logger::DEBUG3) &&
            //                    goby::glog << "PollerInterface condition_variable: no data (spurious?) wakeup"
            //                               << std::endl;
        }
        else
        {
            if (cv_->wait_until(*lock, timeout) == std::cv_status::no_timeout)
                poll_items = poll_all_once();
            else
                return poll_items;
        }
    }
 
    return poll_items;
}
 
#endif