summaryrefslogtreecommitdiff
path: root/modules/io_codec
diff options
context:
space:
mode:
authorClaudius 'keldu' Holeksa <mail@keldu.de>2024-07-24 11:09:55 +0200
committerClaudius 'keldu' Holeksa <mail@keldu.de>2024-07-24 11:09:55 +0200
commitea306799624d0390074f6afa5d38644cce076c9f (patch)
tree81563442aa083b8293f382017cf8d78984e527eb /modules/io_codec
parent977ac8bce989285eaabc76c4ed8571ce5fd6793a (diff)
wip
Diffstat (limited to 'modules/io_codec')
-rw-r--r--modules/io_codec/c++/io_peer.hpp22
-rw-r--r--modules/io_codec/c++/io_peer.tmpl.hpp36
-rw-r--r--modules/io_codec/examples/peer_echo_client.cpp4
3 files changed, 33 insertions, 29 deletions
diff --git a/modules/io_codec/c++/io_peer.hpp b/modules/io_codec/c++/io_peer.hpp
index f576cbe..613f60e 100644
--- a/modules/io_codec/c++/io_peer.hpp
+++ b/modules/io_codec/c++/io_peer.hpp
@@ -11,19 +11,21 @@ template <typename Incoming, typename Outgoing,
typename TransportEncoding, typename ContentEncoding,
typename BufferT = ring_buffer>
class streaming_io_peer {
+private:
+ static_assert(not std::is_same_v<ContentEncoding, encode::Native>, "The native encoding by definition is not fit for transport.");
public:
/**
* Constructor with the option to provide a custom codec, in and out buffer
*/
streaming_io_peer(
- own<conveyor_feeder<data<Incoming, Encoding>>> feed,
- own<async_io_stream> stream, codec<Incoming, Encoding> in_codec, codec<Outgoing, Encoding> out_codec, BufferT in, BufferT out);
+ own<conveyor_feeder<data<Incoming, ContentEncoding>>> feed,
+ own<async_io_stream> stream, codec<Incoming, TransportEncoding> in_codec, codec<Outgoing, TransportEncoding> out_codec, BufferT in, BufferT out);
/**
* Constructor with mostly default assignements
*/
streaming_io_peer(
- own<conveyor_feeder<data<Incoming, Encoding>>> feed,
+ own<conveyor_feeder<data<Incoming, ContentEncoding>>> feed,
own<async_io_stream> stream);
/**
@@ -35,25 +37,25 @@ public:
/**
* Send a message to the remote peer
*/
- error_or<void> send(data<Outgoing> builder);
+ error_or<void> send(data<Outgoing, ContentEncoding> builder);
/**
* A phantom conveyor feeder. Meant for interfacing with other components
*/
- conveyor_feeder<data<Outgoing, Encoding>> &feeder();
+ conveyor_feeder<data<Outgoing, ContentEncoding>> &feeder();
conveyor<void> on_read_disconnected();
private:
/// @unimplemented
class peer_conveyor_feeder final
- : public conveyor_feeder<data<Outgoing, Encoding>> {
+ : public conveyor_feeder<data<Outgoing, ContentEncoding>> {
public:
peer_conveyor_feeder(
streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding, BufferT> &peer_)
: peer_{peer_} {}
- void feed(data<Outgoing, Encoding> &&data_) override {
+ void feed(data<Outgoing, ContentEncoding> &&data_) override {
(void)data_;
}
@@ -63,14 +65,14 @@ private:
size_t queued() const override { return 0; }
- error_or<void> swap(conveyor<data<Outgoing, Encoding>> &&conveyor) noexcept override { return make_error<err::not_implemented>();}
+ error_or<void> swap(conveyor<data<Outgoing, ContentEncoding>> &&) noexcept override { return make_error<err::not_implemented>();}
private:
streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
BufferT> &peer_;
};
private:
- own<conveyor_feeder<data<Incoming, Encoding>>>
+ own<conveyor_feeder<data<Incoming, ContentEncoding>>>
incoming_feeder_ = nullptr;
own<async_io_stream> io_stream_;
@@ -95,7 +97,7 @@ template <typename Incoming, typename Outgoing,
typename TransportEncoding, typename ContentEncoding,
typename BufferT = ring_buffer>
std::pair<own<streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding, BufferT>>,
- conveyor<data<Incoming, Encoding>>>
+ conveyor<data<Incoming, ContentEncoding>>>
new_streaming_io_peer(own<async_io_stream> stream);
} // namespace saw
diff --git a/modules/io_codec/c++/io_peer.tmpl.hpp b/modules/io_codec/c++/io_peer.tmpl.hpp
index d9dfe04..b7ccb49 100644
--- a/modules/io_codec/c++/io_peer.tmpl.hpp
+++ b/modules/io_codec/c++/io_peer.tmpl.hpp
@@ -1,20 +1,20 @@
namespace saw {
-template <typename Incoming, typename Outgoing, typename TransportEncoding, typename ContentEncoding
+template <typename Incoming, typename Outgoing, typename TransportEncoding, typename ContentEncoding,
typename BufferT>
streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
BufferT>::
streaming_io_peer(
- own<conveyor_feeder<data<Incoming, Encoding>>> feed,
+ own<conveyor_feeder<data<Incoming, ContentEncoding>>> feed,
own<async_io_stream> str)
: streaming_io_peer{std::move(feed), std::move(str), {}, {}, {}, {}} {}
-template <typename Incoming, typename Outgoing, typename TransportEncoding, ContentEncoding, typename BufferT>
+template <typename Incoming, typename Outgoing, typename TransportEncoding, typename ContentEncoding, typename BufferT>
streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
BufferT>::
streaming_io_peer(
- own<conveyor_feeder<data<Incoming, Encoding>>> feed,
- own<async_io_stream> stream, codec<Incoming, Encoding> in_codec, codec<Outgoing, Encoding> out_codec, BufferT in, BufferT out)
+ own<conveyor_feeder<data<Incoming, ContentEncoding>>> feed,
+ own<async_io_stream> stream, codec<Incoming, TransportEncoding> in_codec, codec<Outgoing, TransportEncoding> out_codec, BufferT in, BufferT out)
: incoming_feeder_{std::move(feed)},
io_stream_{std::move(stream)},
in_codec_{std::move(in_codec)},
@@ -34,7 +34,7 @@ streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
while (true) {
buffer_view in_view{in_buffer_};
- auto in_data = data<Incoming, Encoding>{in_view};
+ auto in_data = data<Incoming, TransportEncoding>{in_view};
incoming_feeder_->feed(std::move(in_data));
}
@@ -61,14 +61,14 @@ streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
io_stream_->read(&in_buffer_.write(), 1, in_buffer_.write_segment_length());
}
-template <typename Incoming, typename Outgoing,
- typename Encoding, typename BufferT>
-error_or<void> streaming_io_peer<Incoming, Outgoing, Encoding,
- BufferT>::send(data<Outgoing>
+template <typename Incoming, typename Outgoing, typename TransportEncoding,
+ typename ContentEncoding, typename BufferT>
+error_or<void> streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
+ BufferT>::send(data<Outgoing, ContentEncoding>
msg) {
bool restart_write = out_buffer_.read_segment_length() == 0;
- data<Outgoing, Encoding> enc;
+ data<Outgoing, TransportEncoding> enc;
auto eov =
out_codec_.encode(msg, enc);//msg.read(), out_buffer_);
@@ -84,22 +84,22 @@ error_or<void> streaming_io_peer<Incoming, Outgoing, Encoding,
return void_t{};
}
-template <typename Incoming, typename Outgoing, typename Encoding,
+template <typename Incoming, typename Outgoing, typename TransportEncoding, typename ContentEncoding,
typename BufferT>
conveyor<void>
-streaming_io_peer<Incoming, Outgoing, Encoding,
+streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
BufferT>::on_read_disconnected() {
return io_stream_->on_read_disconnected();
}
-template <typename Incoming, typename Outgoing, typename Encoding, typename BufferT>
-std::pair<own<streaming_io_peer<Incoming, Outgoing, Encoding, BufferT>>,
- conveyor<data<Incoming,Encoding>>>
+template <typename Incoming, typename Outgoing, typename TransportEncoding, typename ContentEncoding, typename BufferT>
+std::pair<own<streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding, BufferT>>,
+ conveyor<data<Incoming,ContentEncoding>>>
new_streaming_io_peer(own<async_io_stream> stream) {
auto caf =
- new_conveyor_and_feeder<data<Incoming, Encoding>>();
+ new_conveyor_and_feeder<data<Incoming, ContentEncoding>>();
- return {heap<streaming_io_peer<Incoming, Outgoing, Encoding,BufferT>>(
+ return {heap<streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,BufferT>>(
std::move(caf.feeder), std::move(stream)),
std::move(caf.conveyor)};
}
diff --git a/modules/io_codec/examples/peer_echo_client.cpp b/modules/io_codec/examples/peer_echo_client.cpp
index f1836e4..b09a51a 100644
--- a/modules/io_codec/examples/peer_echo_client.cpp
+++ b/modules/io_codec/examples/peer_echo_client.cpp
@@ -4,6 +4,8 @@
#include <iostream>
+#include <forstio/codec/transport.hpp>
+
int main(){
using namespace saw;
@@ -34,7 +36,7 @@ int main(){
network.connect(*addr).then([](saw::own<saw::io_stream> client){
auto echo_stream = saw::heap<saw::async_io_stream>(std::move(client));
- auto echo_peer_stream_p = saw::new_streaming_io_peer<sch::Echo, sch::Echo, encode::KelSimple, ring_buffer>(std::move(echo_stream));
+ auto echo_peer_stream_p = saw::new_streaming_io_peer<sch::Echo, sch::Echo, transport::FixedLength<8u>, encode::KelSimple, ring_buffer>(std::move(echo_stream));
echo_peer_stream_p.first->on_read_disconnected().attach(std::move(echo_peer_stream_p.first)).detach();
}).detach();