From d2520819150d9a794baf7505d7d02dadeacd5266 Mon Sep 17 00:00:00 2001 From: "Claudius \"keldu\" Holeksa" Date: Thu, 7 Mar 2024 10:22:17 +0100 Subject: io_codec: Moved to new dir structure --- modules/io_codec/c++/SConscript | 38 +++++++++++ modules/io_codec/c++/io_peer.hpp | 104 ++++++++++++++++++++++++++++++ modules/io_codec/c++/io_peer.tmpl.hpp | 117 ++++++++++++++++++++++++++++++++++ modules/io_codec/c++/rpc.hpp | 31 +++++++++ 4 files changed, 290 insertions(+) create mode 100644 modules/io_codec/c++/SConscript create mode 100644 modules/io_codec/c++/io_peer.hpp create mode 100644 modules/io_codec/c++/io_peer.tmpl.hpp create mode 100644 modules/io_codec/c++/rpc.hpp (limited to 'modules/io_codec/c++') diff --git a/modules/io_codec/c++/SConscript b/modules/io_codec/c++/SConscript new file mode 100644 index 0000000..2a277cb --- /dev/null +++ b/modules/io_codec/c++/SConscript @@ -0,0 +1,38 @@ +#!/bin/false + +import os +import os.path +import glob + + +Import('env') + +dir_path = Dir('.').abspath + +# Environment for base library +io_env = env.Clone(); + +io_env.sources = sorted(glob.glob(dir_path + "/*.cpp")) +io_env.headers = sorted(glob.glob(dir_path + "/*.hpp")) + +env.sources += io_env.sources; +env.headers += io_env.headers; + +## Shared lib +objects_shared = [] +io_env.add_source_files(objects_shared, io_env.sources, shared=True); +io_env.library_shared = io_env.SharedLibrary('#build/forstio-io_codec', [objects_shared]); + +## Static lib +objects_static = [] +io_env.add_source_files(objects_static, io_env.sources, shared=False); +io_env.library_static = io_env.StaticLibrary('#build/forstio-io_codec', [objects_static]); + +# Set Alias +env.Alias('library_io_codec', [io_env.library_shared, io_env.library_static]); + +env.targets += ['library_io_codec']; + +# Install +env.Install('$prefix/lib/', [io_env.library_shared, io_env.library_static]); +env.Install('$prefix/include/forstio/io_codec/', [io_env.headers]); diff --git a/modules/io_codec/c++/io_peer.hpp b/modules/io_codec/c++/io_peer.hpp new file mode 100644 index 0000000..9ba623f --- /dev/null +++ b/modules/io_codec/c++/io_peer.hpp @@ -0,0 +1,104 @@ +#pragma once + +#include +#include +#include +#include + +namespace saw { + +template , + typename OutContainer = message_container, + typename BufferT = ring_buffer> +class streaming_io_peer { +public: + /** + * + */ + streaming_io_peer( + own>> feed, + own stream, Codec codec, BufferT in, BufferT out); + /** + * + */ + streaming_io_peer( + own>> feed, + own stream); + + /** + * Deleted copy and move constructors + */ + SAW_FORBID_COPY(streaming_io_peer); + SAW_FORBID_MOVE(streaming_io_peer); + + /** + * Send a message to the remote peer + */ + error send(heap_message_root builder); + + /** + * A phantom conveyor feeder. Meant for interfacing with other components + */ + conveyor_feeder> &feeder(); + + conveyor on_read_disconnected(); + +private: + /// @unimplemented + class peer_conveyor_feeder final + : public conveyor_feeder> { + public: + peer_conveyor_feeder( + streaming_io_peer &peer_) + : peer_{peer_} {} + + void feed(heap_message_root &&data) override { + (void)data; + } + + void fail(error &&error) override { (void)error; } + + size_t space() const override { return 0; } + + size_t queued() const override { return 0; } + + private: + streaming_io_peer &peer_; + }; + +private: + own>> + incoming_feeder_ = nullptr; + + own io_stream_; + + Codec codec_; + + BufferT in_buffer_; + BufferT out_buffer_; + + conveyor_sink sink_read_; + conveyor_sink sink_write_; + + peer_conveyor_feeder conveyor_feeder_; +}; + +/** + * Setup new streaming io peer with the provided network protocols. + * This is a convenience wrapper intended for a faster setup of this class + */ +template , + typename OutContainer = message_container, + typename BufferT = ring_buffer> +std::pair>, + conveyor>> +new_streaming_io_peer(own stream); + +} // namespace saw + +#include "io_peer.tmpl.hpp" diff --git a/modules/io_codec/c++/io_peer.tmpl.hpp b/modules/io_codec/c++/io_peer.tmpl.hpp new file mode 100644 index 0000000..880a58a --- /dev/null +++ b/modules/io_codec/c++/io_peer.tmpl.hpp @@ -0,0 +1,117 @@ +namespace saw { + +template +streaming_io_peer:: + streaming_io_peer( + own>> feed, + own str) + : streaming_io_peer{std::move(feed), std::move(str), {}, {}, {}} {} + +template +streaming_io_peer:: + streaming_io_peer( + own>> feed, + own stream, Codec codec, BufferT in, BufferT out) + : incoming_feeder_{std::move(feed)}, + io_stream_{std::move(stream)}, codec_{std::move(codec)}, + in_buffer_{std::move(in)}, out_buffer_{std::move(out)}, + sink_read_{ + io_stream_->read_done() + .then([this](size_t bytes) -> error_or { + in_buffer_.write_advance(bytes); + + if (in_buffer_.write_segment_length() == 0) { + return critical_error("Message too long"); + } + + io_stream_->read(&in_buffer_.write(), 1, + in_buffer_.write_segment_length()); + + while (true) { + auto root = heap_message_root(); + auto builder = root.build(); + + error err = codec_.template decode( + builder, in_buffer_); + if (err.is_critical()) { + return err; + } + + if (!err.failed()) { + incoming_feeder_->feed(std::move(root)); + } else { + break; + } + } + + return void_t{}; + }) + .sink([this](error err) { + incoming_feeder_->fail(err.copy_error()); + + return err; + })}, + sink_write_{io_stream_->write_done() + .then([this](size_t bytes) -> error_or { + out_buffer_.read_advance(bytes); + if (out_buffer_.readCompositeLength() > 0) { + io_stream_->write( + &out_buffer_.read(), + out_buffer_.read_segment_length()); + } + + return void_t{}; + }) + .sink()} { + io_stream_->read(&in_buffer_.write(), 1, in_buffer_.write_segment_length()); +} + +template +error streaming_io_peer::send(heap_message_root + msg) { + bool restart_write = out_buffer_.read_segment_length() == 0; + + error err = + codec_.template encode(msg.read(), out_buffer_); + if (err.failed()) { + return err; + } + + if (restart_write) { + io_stream_->write(&out_buffer_.read(), + out_buffer_.read_segment_length()); + } + + return no_error(); +} + +template +conveyor +streaming_io_peer::on_read_disconnected() { + return io_stream_->on_read_disconnected(); +} + +template +std::pair>, + conveyor>> +newstreaming_io_peer(own stream) { + auto caf = + new_conveyor_and_feeder>(); + + return {heap>( + std::move(caf.feeder), std::move(stream)), + std::move(caf.conveyor)}; +} + +} // namespace saw diff --git a/modules/io_codec/c++/rpc.hpp b/modules/io_codec/c++/rpc.hpp new file mode 100644 index 0000000..947ce54 --- /dev/null +++ b/modules/io_codec/c++/rpc.hpp @@ -0,0 +1,31 @@ +#pragma once + +#include + +namespace saw { +namespace rmt { +struct Network {}; +} + +template<> +class remote { +private: + std::string addr_str_; +public: + remote(std::string addr_str); + + template + conveyor> connect(); +}; + +template