Method RawSubscribe

src/interface/aimrt_module_cpp_interface/context/context.h:353–409

template <class T, class F>
bool OpSub::RawSubscribe(
    aimrt::channel::SubscriberRef subscriber,
    aimrt::executor::ExecutorRef exe,
    F&& callback) {
  auto type_support = aimrt::GetMessageTypeSupport<T>();

  auto cb_ptr = std::make_shared<typename Context::ChannelCallback<T>>(std::forward<F>(callback));
  auto ctx = details::GetCurrentContext();

  if (!exe) {
    return subscriber.Subscribe(
        type_support,
        [cb_ptr, ctx_ptr = ctx->shared_from_this()](
            const aimrt_channel_context_base_t* chn_ctx_ptr,
            const void* msg_ptr,
            aimrt_function_base_t* release_callback_base) mutable {
          channel::SubscriberReleaseCallback release_callback(release_callback_base);
          ctx_ptr->LetMe();

          if (!ctx_ptr->Running() || ctx_ptr->GetSubState() != context::ChannelState::kOn) [[unlikely]] {
            ctx_ptr->log().Trace("Subscriber is not ready.");
            release_callback();
            return;
          }
          (*cb_ptr)(
              aimrt::channel::ContextRef(chn_ctx_ptr),
              std::shared_ptr<const T>(
                  static_cast<const T*>(msg_ptr),
                  [release_callback{std::move(release_callback)}](const T*) { release_callback(); }));
        });
  }

  return subscriber.Subscribe(
      type_support,
      [ctx_ptr = ctx->shared_from_this(), exeref = std::move(exe), cb_ptr = std::move(cb_ptr)](
          const aimrt_channel_context_base_t* chn_ctx_ptr,
          const void* msg_ptr,
          aimrt_function_base_t* release_callback_base) mutable {
        if (!ctx_ptr->Running() || ctx_ptr->GetSubState() != context::ChannelState::kOn) [[unlikely]] {
          ctx_ptr->log().Trace("Subscriber is not ready.");
          return;
        }

        channel::SubscriberReleaseCallback release_callback(release_callback_base);
        auto msg = std::shared_ptr<const T>(
            static_cast<const T*>(msg_ptr),
            [release_callback{std::move(release_callback)}](const T*) { release_callback(); });
        exeref.Execute([cb_ptr, ctx_ptr, chn_ctx_ptr, msg = std::move(msg)]() mutable {
          ctx_ptr->LetMe();
          if (!ctx_ptr->Running() || ctx_ptr->GetSubState() != context::ChannelState::kOn) [[unlikely]] {
            ctx_ptr->log().Trace("Subscriber is not ready.");
            return;
          }
          (*cb_ptr)(aimrt::channel::ContextRef(chn_ctx_ptr), std::move(msg));
        });
      });
}