ProcessorConnection.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/Executable.h>
#include <asio/ExecutionPolicy.h>
#include <asio/SynchronousExecution.h>
#include <asio/ResourceDestructionGuard.h>
 
#include <FMEThirdPartyWarningsDisable.h>
#include <memory>
#include <FMEThirdPartyWarningsRestore.h>
 
namespace asio {
 
  template<typename... Args>
  class ProcessorConnection;
 
  template<typename... Args>
  class ProcessorConnection<ParameterPack<Args...>> : public Connection
  {
  public:
 
    using Source = Observable<Args...>;
    using Target = Executable<Args...>;
 
    ProcessorConnection() = default;
 
    ProcessorConnection(std::shared_ptr<Source> source, std::shared_ptr<Target> target, std::unique_ptr<ExecutionPolicy> execution_policy, bool mute_until_target_execution_finished = false)
      : policy(std::move(execution_policy))
      , isSynchronousExecution(dynamic_cast<SynchronousExecution*>(policy.get()) != nullptr)
    {
      auto handle = guard.handle();
      if (mute_until_target_execution_finished) {
        connection = source->signal.connect_extended([this, handle, target](const SignalConnection& conn, const Args&... args) {
          auto lock = handle.lock();
          if (lock) {
            SharedSignalConnectionBlock skip_until_finished(conn);
            if (isSynchronousExecution) {
              target->execute(args...);
            }
            else {
              policy->execute([target, skip_until_finished, args...]{
                target->execute(args...);
              });
            }
          }
        });
      }
      else {
        connection = source->signal.connect([this, handle, target](const Args&... args) {
          auto lock = handle.lock();
          if (lock) {
            if (isSynchronousExecution) {
              target->execute(args...);
            }
            else {
              policy->execute([target, args...]{
                target->execute(args...);
              });
            }
          }
        });
      }
    }
 
    void suspend() override
    {
      block = SharedSignalConnectionBlock(connection);
    }
 
    void resume() override
    {
      block = SharedSignalConnectionBlock();
    }
 
  private:
 
    std::unique_ptr<ExecutionPolicy> policy;
    bool isSynchronousExecution;
    ScopedSignalConnection connection;
    SharedSignalConnectionBlock block;
 
    asio::ResourceDestructionGuard guard;
  };
 
}