demo_component.h

Go to the documentation of this file.

// Copyright (c) Space-ng, inc. All rights reserved.
#pragma once
 
#include "sol3/core/message_input.h"
#include "sol3/core/message_output.h"
#include "sol3/core/message_spec.h"
#include "sol3/core/msg/buffer.h"
#include "sol3/core/msg/text.h"
#include "sol3/core/shmem_buffer.h"
 
#include <algorithm>
#include <cstdint>
#include <cstring>
#include <string>
 
namespace sol3::core {
 
struct DemoStepResult {
  bool text_written = false;
  bool buffer_written = false;
  uint64_t text_seq = 0;
  uint64_t buffer_seq = 0;
};
 
// Simple demo component that echoes text and increments buffer metadata.
class DemoComponent {
 public:
  using TextInputSpec = MessageSpec<msg::Text, msg::InterfaceType_INPUT, 0>;
  using TextOutputSpec = MessageSpec<msg::Text, msg::InterfaceType_OUTPUT, 0>;
  using BufferInputSpec = MessageSpec<msg::Buffer, msg::InterfaceType_INPUT, 1>;
  using BufferOutputSpec =
      MessageSpec<msg::Buffer, msg::InterfaceType_OUTPUT, 1>;
 
  DemoComponent(
      IBufferExchange& exchange,
      uint32_t component_id,
      size_t slot_count = 2,
      size_t slot_size = 1024,
      uint32_t width = 8,
      uint32_t height = 8)
      : exchange_(exchange),
        text_input_(exchange, component_id),
        buffer_input_(exchange, component_id),
        text_output_(exchange, component_id, slot_count, slot_size),
        buffer_output_(exchange, component_id, slot_count, slot_size),
        width_(width),
        height_(height) {
    for (size_t i = 0; i < slot_count; ++i) {
      msg::BufferInfo info;
      info.mutate_id(buffer_output_.port().encoded());
      info.mutate_idx(static_cast<uint32_t>(i + 1));
      info.mutate_format(msg::BufferFormat_CPU_GENERIC);
      info.mutate_width(width_);
      info.mutate_height(height_);
      info.mutate_stride_w(width_);
      info.mutate_stride_h(height_);
      info.mutate_size(static_cast<uint64_t>(width_) * height_);
      info.mutate_buffer_use(msg::BufferUse_NONE);
      buffer_pool_.push_back(makeShmemBuffer(info));
      exchange.registerBuffer(*buffer_pool_.back());
    }
    last_buffer_info_ = buffer_pool_.front()->info();
  }
 
  MessagePort textInputPort() const { return text_input_.port(); }
  MessagePort bufferInputPort() const { return buffer_input_.port(); }
  MessagePort textOutputPort() const { return text_output_.port(); }
  MessagePort bufferOutputPort() const { return buffer_output_.port(); }
 
  DemoStepResult step() {
    DemoStepResult result;
    ++step_count_;
 
    auto text_view = text_input_.readHead();
    if (text_view.hasValue()) {
      auto const& payload = text_view.payload();
      if (payload.data()) {
        last_text_ = payload.data()->str();
      }
    }
 
    auto buffer_view = buffer_input_.readHead();
    if (buffer_view.hasValue()) {
      auto const& payload = buffer_view.payload();
      if (payload.info()) {
        last_buffer_info_ = *payload.info();
      }
    }
 
    // Always write outputs; use the most recent inputs when available.
    text_output_.payload().data =
        "echo: " + last_text_ + " step=" + std::to_string(step_count_);
    result.text_seq = text_output_.write();
    result.text_written = result.text_seq > 0;
 
    auto& buffer = buffer_pool_[step_count_ % buffer_pool_.size()];
    auto* data = static_cast<uint8_t*>(buffer->mutableData());
    if (last_buffer_info_.id() != 0 && last_buffer_info_.size() != 0) {
      auto incoming =
          exchange_.get(last_buffer_info_.id(), last_buffer_info_.idx());
      if (incoming && data != nullptr) {
        auto copy_size = std::min(buffer->size(), incoming->size());
        std::memcpy(data, incoming->data(), copy_size);
      }
    } else if (data != nullptr) {
      std::memset(data, static_cast<int>(step_count_ & 0xFF), buffer->size());
    }
    buffer_output_.payload().info = buffer->info();
    result.buffer_seq = buffer_output_.write();
    result.buffer_written = result.buffer_seq > 0;
 
    return result;
  }
 
 private:
  IBufferExchange& exchange_;
  MessageInput<TextInputSpec> text_input_;
  MessageInput<BufferInputSpec> buffer_input_;
  MessageOutput<TextOutputSpec> text_output_;
  MessageOutput<BufferOutputSpec> buffer_output_;
  uint32_t width_ = 0;
  uint32_t height_ = 0;
  uint64_t step_count_ = 0;
  std::string last_text_ = "";
  msg::BufferInfo last_buffer_info_{};
  std::vector<std::unique_ptr<IBufferMutable>> buffer_pool_;
};
 
}  // namespace sol3::core