combiner.h

Go to the documentation of this file.

/****************************************************************************
 *
 * fkie_message_filters
 * Copyright © 2018-2020 Fraunhofer FKIE
 * Author: Timo Röhling
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 ****************************************************************************/
#ifndef INCLUDE_FKIE_MESSAGE_FILTERS_COMBINER_H_
#define INCLUDE_FKIE_MESSAGE_FILTERS_COMBINER_H_

#include "source.h"
#include "sink.h"

namespace fkie_message_filters
{

namespace combiner_policies
{
}

#ifndef DOXYGEN
template<template<typename...> class PolicyTmpl, class... IOs>
class Combiner : public helpers::io_rewrap_t<helpers::io_concat_t<IOs...>, Source>
#else
template<template<typename...> class PolicyTmpl, class... IOs>
class Combiner : public Source<IOs...>
#endif
{
public:
    static constexpr std::size_t NUM_SINKS = sizeof...(IOs);
    using Connections = std::array<Connection, NUM_SINKS>;
    template<std::size_t N>
    using SinkType = helpers::io_rewrap_t<helpers::select_nth<N, IOs...>, Sink>;
    using Policy = PolicyTmpl<IOs...>;
    explicit Combiner (const Policy& policy = Policy()) noexcept;
    template<std::size_t N>
    SinkType<N>& sink() noexcept;
    template<std::size_t N>
    const SinkType<N>& sink() const noexcept;
    const Policy& policy() const noexcept;
    void set_policy(const Policy& policy) noexcept;
    Connections connect_to_sources (helpers::io_rewrap_t<IOs, Source>&... sources) noexcept;
    void disconnect_from_all_sources() noexcept;
    void reset() noexcept override;
    void disconnect() noexcept override;
private:
    using IncomingTuples = std::tuple<typename helpers::io_tuple_t<IOs>...>;
    using OutgoingTuple = helpers::io_tuple_t<helpers::io_concat_t<IOs...>>;
    template<typename... Inputs>
    class CombinerSink : public Sink<Inputs...>
    {
    public:
        using Tuple = std::tuple<Inputs...>;
        using PolicyInFunc = std::function<void(std::unique_lock<std::mutex>&, const Tuple&)>;
        void set_parent(Combiner* parent) noexcept;
        void set_policy_input(const PolicyInFunc& f) noexcept;
    protected:
        void receive(const Inputs&... in) override;
    private:
        PolicyInFunc forward_;
        Combiner* parent_{nullptr};
    };
    std::mutex combiner_mutex_;
    Policy policy_;
    std::tuple<typename IOs::template Rewrap<CombinerSink>...> sinks_;
    void connect_policy() noexcept;
    template<std::size_t N, typename ThisSource, typename... OtherSources>
    void connect_to_sources_impl(Connections& conn, ThisSource& src, OtherSources&... sources) noexcept;
    template<std::size_t N>
    void connect_to_sources_impl(Connections& conn) noexcept;
};

} // namespace fkie_message_filters

#include "combiner_impl.h"

#endif /* INCLUDE_FKIE_MESSAGE_FILTERS_COMBINER_H_ */