ObservableConnection.h

Go to the documentation of this file.

// Copyright (c) Fraunhofer MEVIS, Germany. All rights reserved.
// **InsertLicense** code
 
#pragma once
 
#include <asio/Connection.h>
#include <asio/Observable.h>
#include <asio/Processable.h>
#include <asio/ExecutionPolicy.h>
#include <asio/SynchronousExecution.h>
#include <asio/ResourceDestructionGuard.h>
 
#include <FMEThirdPartyWarningsDisable.h>
#include <memory>
#include <utility>
#include <tuple>
#include <array>
#include <cstddef>
#include <FMEThirdPartyWarningsRestore.h>
 
 
namespace std14
{
  template<typename T, T... Ints>
  struct integer_sequence
  {
    using value_type = T;
    static constexpr std::size_t size() { return sizeof...(Ints); }
  };
 
  template<std::size_t... Ints>
  using index_sequence = integer_sequence<std::size_t, Ints...>;
 
  template<typename T, std::size_t N, T... Is>
  struct make_integer_sequence : make_integer_sequence<T, N - 1, N - 1, Is...> {};
 
  template<typename T, T... Is>
  struct make_integer_sequence<T, 0, Is...> : integer_sequence<T, Is...> {};
 
  template<std::size_t N>
  using make_index_sequence = make_integer_sequence<std::size_t, N>;
 
  template<typename... T>
  using index_sequence_for = make_index_sequence<sizeof...(T)>;
}
 
 
namespace detail {
 
  template <typename Fnc, typename... Args, std::size_t... Indices>
  void apply_impl(Fnc &&fnc, const std::tuple<Args...> &tuple, std14::index_sequence<Indices...>)
  {
    std::forward<Fnc>(fnc)(std::get<Indices>(tuple)...);
  }
 
  template <typename Fnc, typename... Args>
  void apply(Fnc &&fnc, const std::tuple<Args...> &tuple)
  {
    apply_impl(std::forward<Fnc>(fnc), tuple, std14::index_sequence_for<Args...>());
  }
 
}
 
 
namespace asio {
 
 
  template<typename... Args>
  class ObservableConnectionInput final
  {
  public:
 
    std::shared_ptr<Observable<Args...>> observable;
    bool shouldTriggerProcessing;
    bool shouldBeInvalidatedAfterProcessing;
  };
 
 
  template<typename Result, typename... Args>
  class ObservableConnection : public Connection
  {
    class State final
    {
    public:
      State(
        std::shared_ptr<Observable<Result>> target_,
        std::shared_ptr<Processable<Result, Args...>> processable_,
        std::unique_ptr<ExecutionPolicy> policy_,
        bool muteUntilTargetSignalEmitted_)
        : target(target_)
        , processable(processable_)
        , policy(std::move(policy_))
        , isSynchronousExecutionPolicy(dynamic_cast<asio::SynchronousExecution*>(policy.get()) != nullptr)
        , muteUntilTargetSignalEmitted(muteUntilTargetSignalEmitted_)
        , isCurrentlyMuted(false)
      {
        validities.fill(false);
      }
 
      std::tuple<Args...> values;
      std::array<bool, sizeof...(Args)> validities;
      std::array<bool, sizeof...(Args)> invalidations;
      std::shared_ptr<Observable<Result>> target;
      std::shared_ptr<Processable<Result, Args...>> processable;
      std::unique_ptr<ExecutionPolicy> policy;
      bool isSynchronousExecutionPolicy;
      bool muteUntilTargetSignalEmitted;
      bool isCurrentlyMuted;
    };
 
  public:
 
    ObservableConnection() = default;
 
    ObservableConnection(
      ObservableConnectionInput<Args>... inputs,
      std::shared_ptr<Observable<Result>> target,
      std::shared_ptr<Processable<Result, Args...>> processable,
      std::unique_ptr<ExecutionPolicy> policy,
      bool muteUntilTargetSignalEmitted)
    {
      // We need a state object as we can't use the this pointer
      // as capture inside the connection lambda down below due to
      // vtable issues (because we're inside a constructor).
      // An alternative would be to move the connection into a
      // public method, but this would uglify the API - the connection
      // should be well established once the object is constructed.
      auto state = std::make_shared<State>(target, processable, std::move(policy), muteUntilTargetSignalEmitted);
      observe(state, inputs...);
    }
 
    void suspend() override
    {
      for (std::size_t i = 0; i < blocks.size(); ++i) {
        blocks[i] = SharedSignalConnectionBlock(connections[i]);
      }
    }
 
    void resume() override
    {
      for (std::size_t i = 0; i < blocks.size(); ++i) {
        blocks[i] = SharedSignalConnectionBlock();
      }
    }
 
  private:
 
    template<typename HeadType, typename... TailTypes>
    void observe(std::shared_ptr<State> state, const ObservableConnectionInput<HeadType>& head, const ObservableConnectionInput<TailTypes>&... tail)
    {
      do_observe<HeadType, sizeof...(Args) - (sizeof...(TailTypes) + 1)>(state, head);
      observe(state, tail...);
    }
 
    template<typename LastType>
    void observe(std::shared_ptr<State> state, const ObservableConnectionInput<LastType>& last)
    {
      do_observe<LastType, sizeof...(Args) - 1>(state, last);
    }
 
    template<typename Type, std::size_t Position>
    void do_observe(std::shared_ptr<State> state, const ObservableConnectionInput<Type>& input)
    {
      state->invalidations[Position] = input.shouldBeInvalidatedAfterProcessing;
      auto shouldTrigger = input.shouldTriggerProcessing;
      auto handle = guard.handle();
      connections[Position] = input.observable->signal.connect([handle, state, shouldTrigger](const Type& value) {
        auto lock = handle.lock();
        if (lock) {
          std::get<Position>(state->values) = value;
          state->validities[Position] = true;
          if (shouldTrigger && !state->isCurrentlyMuted) {
            if (std::all_of(state->validities.begin(), state->validities.end(), [](bool v) { return v; })) {
              if (state->isSynchronousExecutionPolicy) {
                detail::apply([&state](const Args&... args) {
                  state->target->signal(state->processable->process(args...));
                }, state->values);
              }
              else {
                if (state->muteUntilTargetSignalEmitted) {
                  state->isCurrentlyMuted = true;
                }
                auto shouldUnmute = state->muteUntilTargetSignalEmitted;
                
                // For non-synchronous execution policies we need to capture
                // the values as they are at this very timepoint in the execution
                // lambda, because following input changes will overwrite them in
                // the shared state. We use a reference here and by-value capture to
                // avoid multiple copies.
                auto &values = state->values;
                state->policy->execute([handle, values, state, shouldUnmute] {
                  detail::apply([&state](const Args&... args) {
                    state->target->signal(state->processable->process(args...));
                  }, values);
                  if (shouldUnmute) {
                    auto innerLock = handle.lock();
                    if (innerLock) {
                      state->isCurrentlyMuted = false;
                    }
                  }
                });
              }
              for (std::size_t i = 0; i < state->invalidations.size(); ++i) {
                if (state->invalidations[i]) {
                  state->validities[i] = false;
                }
              }
            }
          }
        }
      });
    }
 
    std::array<ScopedSignalConnection, sizeof...(Args)> connections;
    std::array<SharedSignalConnectionBlock, sizeof...(Args)> blocks;
 
    asio::ResourceDestructionGuard guard;
  };
 
 
  // Specialization for parameterless processing with a trigger, i.e., asio::Observable<>:
  template<typename Result>
  class ObservableConnection<Result> : public Connection
  {
    class State final
    {
    public:
      State(
        std::shared_ptr<Observable<Result>> target_,
        std::shared_ptr<Processable<Result>> processable_,
        std::unique_ptr<ExecutionPolicy> policy_,
        bool muteUntilTargetSignalEmitted_)
        : target(target_)
        , processable(processable_)
        , policy(std::move(policy_))
        , isSynchronousExecutionPolicy(dynamic_cast<asio::SynchronousExecution*>(policy.get()) != nullptr)
        , muteUntilTargetSignalEmitted(muteUntilTargetSignalEmitted_)
        , isCurrentlyMuted(false)
      {
      }
 
      std::shared_ptr<Observable<Result>> target;
      std::shared_ptr<Processable<Result>> processable;
      std::unique_ptr<ExecutionPolicy> policy;
      bool isSynchronousExecutionPolicy;
      bool muteUntilTargetSignalEmitted;
      bool isCurrentlyMuted;
    };
 
  public:
 
    ObservableConnection() = default;
 
    ObservableConnection(
      ObservableConnectionInput<> input,
      std::shared_ptr<Observable<Result>> target,
      std::shared_ptr<Processable<Result>> processable,
      std::unique_ptr<ExecutionPolicy> policy,
      bool muteUntilTargetSignalEmitted)
    {
      auto state = std::make_shared<State>(target, processable, std::move(policy), muteUntilTargetSignalEmitted);
      auto handle = guard.handle();
      connection = input.observable->signal.connect([handle, state] {
        auto lock = handle.lock();
        if (lock) {
          if (!state->isCurrentlyMuted) {
            if (state->isSynchronousExecutionPolicy) {
              state->target->signal(state->processable->process());
            }
            else {
              if (state->muteUntilTargetSignalEmitted) {
                state->isCurrentlyMuted = true;
              }
              auto shouldUnmute = state->muteUntilTargetSignalEmitted;
              state->policy->execute([handle, state, shouldUnmute]{
                state->target->signal(state->processable->process());
                if (shouldUnmute) {
                  auto innerLock = handle.lock();
                  if (innerLock) {
                    state->isCurrentlyMuted = false;
                  }
                }
              });
            }
          }
        }
      });
    }
 
    void suspend() override
    {
      block = SharedSignalConnectionBlock(connection);
    }
 
    void resume() override
    {
      block = SharedSignalConnectionBlock();
    }
 
  private:
 
    ScopedSignalConnection connection;
    SharedSignalConnectionBlock block;
 
    asio::ResourceDestructionGuard guard;
  };
 
  // Specialization for connecting two triggers with a void processable without parameters
  template<>
  class ObservableConnection<void> : public Connection
  {
    class State final
    {
    public:
      State(
        std::shared_ptr<Observable<>> target_,
        std::shared_ptr<Processable<void>> processable_,
        std::unique_ptr<ExecutionPolicy> policy_,
        bool muteUntilTargetSignalEmitted_)
        : target(target_)
        , processable(processable_)
        , policy(std::move(policy_))
        , isSynchronousExecutionPolicy(dynamic_cast<asio::SynchronousExecution*>(policy.get()) != nullptr)
        , muteUntilTargetSignalEmitted(muteUntilTargetSignalEmitted_)
        , isCurrentlyMuted(false)
      {
      }
 
      std::shared_ptr<Observable<>> target;
      std::shared_ptr<Processable<void>> processable;
      std::unique_ptr<ExecutionPolicy> policy;
      bool isSynchronousExecutionPolicy;
      bool muteUntilTargetSignalEmitted;
      bool isCurrentlyMuted;
    };
 
  public:
 
    ObservableConnection() = default;
 
    ObservableConnection(
      ObservableConnectionInput<> input,
      std::shared_ptr<Observable<>> target,
      std::shared_ptr<Processable<void>> processable,
      std::unique_ptr<ExecutionPolicy> policy,
      bool muteUntilTargetSignalEmitted)
    {
      auto state = std::make_shared<State>(target, processable, std::move(policy), muteUntilTargetSignalEmitted);
      auto handle = guard.handle();
      connection = input.observable->signal.connect([state, handle] {
        auto lock = handle.lock();
        if (lock) {
          if (!state->isCurrentlyMuted) {
            if (state->isSynchronousExecutionPolicy) {
              state->processable->process();
              state->target->signal();
            }
            else {
              if (state->muteUntilTargetSignalEmitted) {
                state->isCurrentlyMuted = true;
              }
              auto shouldUnmute = state->muteUntilTargetSignalEmitted;
              state->policy->execute([state, handle, shouldUnmute] {
                state->processable->process();
                state->target->signal();
                if (shouldUnmute) {
                  auto innerLock = handle.lock();
                  if (innerLock) {
                    state->isCurrentlyMuted = false;
                  }
                }
              });
            }
          }
        }
      });
    }
 
    void suspend() override
    {
      block = SharedSignalConnectionBlock(connection);
    }
 
    void resume() override
    {
      block = SharedSignalConnectionBlock();
    }
 
  private:
 
    ScopedSignalConnection connection;
    SharedSignalConnectionBlock block;
 
    asio::ResourceDestructionGuard guard;
  };
 
 
  // Specialization for void types
  template<typename... Args>
  class ObservableConnection<void, Args...> : public Connection
  {
    class State final
    {
    public:
      State(
        std::shared_ptr<Observable<>> target_,
        std::shared_ptr<Processable<void, Args...>> processable_,
        std::unique_ptr<ExecutionPolicy> policy_,
        bool muteUntilTargetSignalEmitted_)
        : target(target_)
        , processable(processable_)
        , policy(std::move(policy_))
        , isSynchronousExecutionPolicy(dynamic_cast<asio::SynchronousExecution*>(policy.get()) != nullptr)
        , muteUntilTargetSignalEmitted(muteUntilTargetSignalEmitted_)
        , isCurrentlyMuted(false)
      {
        validities.fill(false);
      }
 
      std::tuple<Args...> values;
      std::array<bool, sizeof...(Args)> validities;
      std::array<bool, sizeof...(Args)> invalidations;
      std::shared_ptr<Observable<>> target;
      std::shared_ptr<Processable<void, Args...>> processable;
      std::unique_ptr<ExecutionPolicy> policy;
      bool isSynchronousExecutionPolicy;
      bool muteUntilTargetSignalEmitted;
      bool isCurrentlyMuted;
    };
 
  public:
 
    ObservableConnection() = default;
 
    ObservableConnection(
      ObservableConnectionInput<Args>... inputs,
      std::shared_ptr<Observable<>> target,
      std::shared_ptr<Processable<void, Args...>> processable,
      std::unique_ptr<ExecutionPolicy> policy,
      bool muteUntilTargetSignalEmitted)
    {
      auto state = std::make_shared<State>(target, processable, std::move(policy), muteUntilTargetSignalEmitted);
      observe(state, inputs...);
    }
 
    void suspend() override
    {
      for (std::size_t i = 0; i < blocks.size(); ++i) {
        blocks[i] = SharedSignalConnectionBlock(connections[i]);
      }
    }
 
    void resume() override
    {
      for (std::size_t i = 0; i < blocks.size(); ++i) {
        blocks[i] = SharedSignalConnectionBlock();
      }
    }
 
  private:
 
    template<typename HeadType, typename... TailTypes>
    void observe(std::shared_ptr<State> state, const ObservableConnectionInput<HeadType>& head, const ObservableConnectionInput<TailTypes>&... tail)
    {
      do_observe<HeadType, sizeof...(Args) - (sizeof...(TailTypes) + 1)>(state, head);
      observe(state, tail...);
    }
 
    template<typename LastType>
    void observe(std::shared_ptr<State> state, const ObservableConnectionInput<LastType>& last)
    {
      do_observe<LastType, sizeof...(Args) - 1>(state, last);
    }
 
    template<typename Type, std::size_t Position>
    void do_observe(std::shared_ptr<State> state, const ObservableConnectionInput<Type>& input)
    {
      state->invalidations[Position] = input.shouldBeInvalidatedAfterProcessing;
      auto shouldTrigger = input.shouldTriggerProcessing;
      auto handle = guard.handle();
      connections[Position] = input.observable->signal.connect([state, handle, shouldTrigger](const Type& value) {
        auto lock = handle.lock();
        if (lock) {
          std::get<Position>(state->values) = value;
          state->validities[Position] = true;
          if (shouldTrigger && !state->isCurrentlyMuted) {
            if (std::all_of(state->validities.begin(), state->validities.end(), [](bool v) { return v; })) {
              if (state->isSynchronousExecutionPolicy) {
                detail::apply([&state](const Args&... args) {
                  state->processable->process(args...);
                  state->target->signal();
                }, state->values);
              }
              else {
                if (state->muteUntilTargetSignalEmitted) {
                  state->isCurrentlyMuted = true;
                }
                auto shouldUnmute = state->muteUntilTargetSignalEmitted;
                auto& values = state->values;
                state->policy->execute([state, values, handle, shouldUnmute] {
                  detail::apply([&state](const Args&... args) {
                    state->processable->process(args...);
                    state->target->signal();
                  }, values);
                  if (shouldUnmute) {
                    auto innerLock = handle.lock();
                    if (innerLock) {
                      state->isCurrentlyMuted = false;
                    }
                  }
                });
              }
              for (std::size_t i = 0; i < state->invalidations.size(); ++i) {
                if (state->invalidations[i]) {
                  state->validities[i] = false;
                }
              }
            }
          }
        }
      });
    }
 
    std::array<ScopedSignalConnection, sizeof...(Args)> connections;
    std::array<SharedSignalConnectionBlock, sizeof...(Args)> blocks;
 
    asio::ResourceDestructionGuard guard;
  };
 
}