detail_stream_handler_base.hpp

#pragma once
 
#include "../../global.hpp"
#include "../../async/_decl/decl_stream_handler_base.hpp"
 
namespace Cvb
{
 
  CVB_BEGIN_INLINE_NS
 
  namespace Async
  {
    namespace Internal
    {
      template <>
      struct TpedStreamHandlerTraits<ImageStream>
      {
        using type = StreamHandlerBase<ImageStream>;
      };
 
      template <>
      struct TpedStreamHandlerTraits<CompositeStream>
      {
        using type = StreamHandlerBase<CompositeStream>;
      };
 
      template <>
      struct TpedStreamHandlerTraits<PointCloudStream>
      {
        using type = StreamHandlerBase<PointCloudStream>;
      };
 
      template <>
      struct TpedStreamHandlerTraits<Stream>
      {
        using type = StreamHandlerBase<Stream>;
      };
 
    }  // namespace Internal
 
    template <class STREAMTYPE>
    inline StreamHandlerBase<STREAMTYPE>::~StreamHandlerBase()
    {
      assert(("stream must be finished prior to destruction", !IsActive()));  // NOLINT
    }
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::Run()
    {
      std::unique_lock<std::mutex> guard(mutex_);
      if (isActive_)
        throw std::runtime_error("handler is already running");
 
      lastError_ = nullptr;
      isActive_  = true;
      Setup(streamVector_);
      thread_ = std::make_unique<std::thread>(std::ref(*this));
    }
    inline void StreamHandlerBase<STREAMTYPE>::Run() {…}
 
    template <class STREAMTYPE>
    inline bool StreamHandlerBase<STREAMTYPE>::TryFinish() noexcept
    {
      if (!isActive_)
        return !lastError_;
 
      BeginFinish();
 
      if (thread_ && thread_->get_id() != std::this_thread::get_id())
      {
        thread_->join();
        EndFinish();
      }
 
      return !lastError_;
    }
    inline bool StreamHandlerBase<STREAMTYPE>::TryFinish() noexcept {…}
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::Finish()
    {
      BeginFinish();
 
      if (thread_ && thread_->joinable())
      {
        thread_->join();
        EndFinish();
      }
 
      if (lastError_)
        std::rethrow_exception(lastError_);
    }
    inline void StreamHandlerBase<STREAMTYPE>::Finish() {…}
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::operator()()
    {
      Begin();
      try
      {
        while (isActive_)
          HandleAsyncStream(streamVector_);
      }
      catch (const std::exception &error)
      {
        if (!IsActive())
          return;
        HandleError(error);
      }
      End();
    }
 
    template <>
    inline void StreamHandlerBase<Driver::Stream>::HandleAsyncStream(const StreamVectorType &streamVector)
    {
      try
      {
        std::vector<Driver::WaitResult<Driver::StreamImage>> waitResultVector;
        waitResultVector.reserve(streamVector.size());
        for (auto &stream : streamVector)
          waitResultVector.emplace_back(stream->Wait());
 
        HandleAsyncWaitResult(waitResultVector);
      }
      catch (const std::exception &error)
      {
        if (!IsActive())
          return;
        HandleError(error);
      }
    }
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::HandleAsyncStream(const StreamVectorType &streamVector)
    {
      try
      {
        std::vector<WaitResultTuple<typename Internal::DeliverableTraits<STREAMTYPE>::type>> waitResultVector;
        waitResultVector.reserve(streamVector.size());
        for (auto &stream : streamVector)
          waitResultVector.emplace_back(stream->Wait());
 
        HandleAsyncWaitResult(waitResultVector);
      }
      catch (const std::exception &error)
      {
        if (!IsActive())
          return;
        HandleError(error);
      }
    }
    inline void StreamHandlerBase<STREAMTYPE>::HandleAsyncStream(const StreamVectorType &streamVector) {…}
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::HandleAsyncWaitResult(
        const std::vector<WaitResult<typename Internal::DeliverableTraits<STREAMTYPE>::type>> &)
    {
      // intentionally left empty.
    }
    inline void StreamHandlerBase<STREAMTYPE>::HandleAsyncWaitResult( {…}
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::HandleAsyncWaitResult(
        const std::vector<WaitResultTuple<typename Internal::DeliverableTraits<STREAMTYPE>::type>> &)
    {
      // intentionally left empty.
    }
    inline void StreamHandlerBase<STREAMTYPE>::HandleAsyncWaitResult( {…}
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::Setup(const StreamVectorType streamVector)
    {
      for (auto &stream : streamVector)
        stream->Start();
    }
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::TearDown(const StreamVectorType streamVector) noexcept
    {
      for (auto &stream : streamVector)
        stream->TryAbort();
    }
    inline void StreamHandlerBase<STREAMTYPE>::TearDown(const StreamVectorType streamVector) noexcept {…}
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::HandleError(const std::exception &) noexcept
    {
      lastError_ = std::current_exception();
    }
    inline void StreamHandlerBase<STREAMTYPE>::HandleError(const std::exception &) noexcept {…}
 
    template <class STREAMTYPE>
    inline StreamHandlerBase<STREAMTYPE>::StreamHandlerBase(const StreamPtrType &stream)
        : isActive_(false)
    {
      StreamVectorType streamVector(1);
      streamVector[0] = stream;
      streamVector.swap(streamVector_);
    }
    inline StreamHandlerBase<STREAMTYPE>::StreamHandlerBase(const StreamPtrType &stream) {…}
 
    template <class STREAMTYPE>
    inline StreamHandlerBase<STREAMTYPE>::StreamHandlerBase(const StreamVectorType &streamVector)
        : isActive_(false)
    {
      StreamVectorType(streamVector).swap(streamVector_);
    }
    inline StreamHandlerBase<STREAMTYPE>::StreamHandlerBase(const StreamVectorType &streamVector) {…}
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::BeginFinish()
    {
      std::unique_lock<std::mutex> guard(mutex_);
      isActive_ = false;
      TearDown(streamVector_);
    }
 
    template <class STREAMTYPE>
    inline void StreamHandlerBase<STREAMTYPE>::EndFinish()
    {
      std::unique_lock<std::mutex> guard(mutex_);
      thread_.reset();
    }
 
  }  // namespace Async
 
  CVB_END_INLINE_NS
 
}  // namespace Cvb
 {…}
    inline void StreamHandlerBase<STREAMTYPE>::Setup(const StreamVectorType streamVector) {…}