message_input.h

Go to the documentation of this file.

// Copyright (c) Space-ng, inc. All rights reserved.
#pragma once
#include "sol3/core/any_message_input.h"
#include "sol3/core/message_spec.h"
#include "sol3/core/message_view.h"
 
#include <flatbuffers/flatbuffers.h>
#include <flatbuffers/verifier.h>
#include <spdlog/spdlog.h>
 
#include <chrono>
 
namespace sol3::core {
 
template <typename TMessageSpec>
class MessageInput : public AnyMessageInput {
  static_assert(
      is_message_spec<TMessageSpec>::value,
      "TMessageSpec must be sol3::core::MessageSpec<...>");
 
 public:
  using TableType = typename TMessageSpec::TableType;
  using NativeTableType = typename TableType::NativeTableType;
 
  explicit MessageInput(IBufferExchange& exchange, uint32_t component_id)
      : AnyMessageInput(
            exchange,
            TMessageSpec::makeMessagePort(component_id),
            MessageSchema(
                NativeTableType::GetSchemaData(),
                MessageSchema::EvolutionPolicy::FORWARDS_COMPATIBLE)) {}
 
  explicit MessageInput(
      IBufferExchange& exchange,
      IBufferExchange::Endpoint endpoint,
      uint32_t component_id)
      : AnyMessageInput(
            exchange,
            std::move(endpoint),
            TMessageSpec::makeMessagePort(component_id),
            MessageSchema(
                NativeTableType::GetSchemaData(),
                MessageSchema::EvolutionPolicy::FORWARDS_COMPATIBLE)) {}
 
  MessageInput(MessageInput&& rhs) = delete;
  MessageInput& operator=(MessageInput&& rhs) = delete;
  MessageInput(MessageInput const&) = delete;
  MessageInput& operator=(MessageInput const& rhs) = delete;
  ~MessageInput() = default;
 
  MessageView<TMessageSpec> read(
      uint64_t seq, std::chrono::steady_clock::duration time_out = {}) {
    return AnyMessageInput::read(seq, time_out);
  }
 
  MessageView<TMessageSpec> readHead(
      std::chrono::steady_clock::duration time_out = {}) {
    return AnyMessageInput::readHead(time_out);
  }
};
 
template <typename TMessageSpec>
struct MessageViewSuccess {
  MessageView<TMessageSpec> view;
  bool success = false;
};
 
template <typename TMessageSpec, typename TPredicate>
MessageViewSuccess<TMessageSpec> waitUntil(
    MessageInput<TMessageSpec>& input,
    std::chrono::nanoseconds timeout,
    TPredicate predicate,
    std::chrono::nanoseconds poll_interval = std::chrono::milliseconds(10)) {
  auto start_time = std::chrono::steady_clock::now();
  MessageViewSuccess<TMessageSpec> result;
  result.view = input.readHead();
  // first predicate test, we want to test regardless if this is a cached read.
  if (result.view.hasValue()) {
    result.success = predicate(result.view);
  }
  if (result.success) {
    return result;
  }
  // poll head, and only evaluate predicate if the result is cached.
  do {
    std::this_thread::sleep_for(poll_interval);
    result.view = input.readHead();
    if (result.view.hasValue() && !result.view.cached()) {
      result.success = predicate(result.view);
    }
  } while (!result.success &&
           (std::chrono::steady_clock::now() - start_time) < timeout);
 
  return result;
}
 
}  // namespace sol3::core