diff options
Diffstat (limited to 'modules')
| -rw-r--r-- | modules/io_codec/SConstruct | 10 | ||||
| -rw-r--r-- | modules/io_codec/examples/SConscript | 32 | ||||
| -rw-r--r-- | modules/io_codec/examples/echo.hpp | 4 | ||||
| -rw-r--r-- | modules/io_codec/examples/peer_echo_client.cpp | 68 | ||||
| -rw-r--r-- | modules/io_codec/examples/peer_echo_server.cpp | 162 |
5 files changed, 275 insertions, 1 deletions
diff --git a/modules/io_codec/SConstruct b/modules/io_codec/SConstruct index f4b8164..7d0fb6f 100644 --- a/modules/io_codec/SConstruct +++ b/modules/io_codec/SConstruct @@ -37,11 +37,18 @@ env_vars = Variables( args=ARGUMENTS ) +env_vars.Add( + BoolVariable('build_examples', + help='Build examples', + default=False + ) +); + env_vars.Add('prefix', help='Installation target location of build results and headers', default='/usr/local/', validator=isAbsolutePath -) +); env=Environment(ENV=os.environ, variables=env_vars, CPPPATH=[], CPPDEFINES=['SAW_UNIX'], @@ -64,6 +71,7 @@ env.targets = []; Export('env') SConscript('c++/SConscript') +SConscript('examples/SConscript') env.Alias('cdb', env.cdb); env.Alias('all', [env.targets]); diff --git a/modules/io_codec/examples/SConscript b/modules/io_codec/examples/SConscript new file mode 100644 index 0000000..5c35209 --- /dev/null +++ b/modules/io_codec/examples/SConscript @@ -0,0 +1,32 @@ +#!/bin/false + +import os +import os.path +import glob + + +Import('env') + +dir_path = Dir('.').abspath + +# Environment for base library +examples_env = env.Clone(); + +examples_env.sources = sorted(glob.glob(dir_path + "/*.cpp")) +examples_env.headers = sorted(glob.glob(dir_path + "/*.hpp")) + +env.sources += examples_env.sources; +env.headers += examples_env.headers; + +objects_static = [] +examples_env.echo_client = examples_env.Program('#bin/peer_echo_client', ['peer_echo_client.cpp', env.library_static]); +#examples_env.echo_server = examples_env.Program('#bin/peer_echo_server', ['peer_echo_server.cpp', env.library_static]); + +# Set Alias +env.examples = [examples_env.echo_client, examples_env.echo_server]; +env.Alias('examples', env.examples); + +if env["build_examples"]: + env.targets += ['examples']; + env.Install('$prefix/bin/', env.examples); +#endif diff --git a/modules/io_codec/examples/echo.hpp b/modules/io_codec/examples/echo.hpp new file mode 100644 index 0000000..7a1c6ab --- /dev/null +++ b/modules/io_codec/examples/echo.hpp @@ -0,0 +1,4 @@ +#pragma once + +namespace saw { +} diff --git a/modules/io_codec/examples/peer_echo_client.cpp b/modules/io_codec/examples/peer_echo_client.cpp new file mode 100644 index 0000000..54f82a9 --- /dev/null +++ b/modules/io_codec/examples/peer_echo_client.cpp @@ -0,0 +1,68 @@ +#include "../c++/io_peer.hpp" + +#include <array> +#include <iostream> + +#include "echo.hpp" + +int main(){ + /** + * Create EventLoop + * Setup EventPort to the outside world + * And setup the io comms to the outside. + */ + auto eo_aio = saw::setup_async_io(); + if(eo_aio.is_error()){ + auto& err = eo_aio.get_error(); + std::cerr<<err.get_message()<<std::endl; + return err.get_id(); + } + auto& aio = eo_aio.get_value(); + /** + * Make the event loop the current event loop on this thread + */ + saw::wait_scope wait_scope{aio.event_loop}; + + bool keep_running = true; + aio.event_port.on_signal(saw::Signal::Terminate).then([&keep_running](){ + keep_running = false; + }).detach(); + + saw::own<saw::network_address> net_addr = nullptr; + saw::own<saw::async_io_stream> async_rmt = nullptr; + + std::array<uint8_t, 32> read_data; + uint64_t read_bytes = 0; + auto& network = aio.io->get_network(); + network.resolve_address("127.0.0.1", 4322).then([&](auto addr){ + net_addr = std::move(addr); + network.connect(*net_addr).then([&](auto rmt_srv){ + async_rmt = saw::heap<saw::async_io_stream>(std::move(rmt_srv)); + async_rmt->write("foo", 3); + + async_rmt->read(&read_data[0], 3, read_data.size()-1); + + async_rmt->read_done().then([&](size_t b){ + for(uint64_t i = 0; i < b; ++i){ + std::cout<<static_cast<char>(read_data[i]); + } + std::cout<<std::endl; + + keep_running = false; + }).detach(); + + async_rmt->on_read_disconnected().then([&](){ + keep_running = false; + }).detach(); + }).detach(); + }).detach(); + + wait_scope.poll(); + while(keep_running){ + wait_scope.wait(); + } + + std::cout<<"Shutting down echo client"<<std::endl; + + return 0; +} diff --git a/modules/io_codec/examples/peer_echo_server.cpp b/modules/io_codec/examples/peer_echo_server.cpp new file mode 100644 index 0000000..50863db --- /dev/null +++ b/modules/io_codec/examples/peer_echo_server.cpp @@ -0,0 +1,162 @@ +#include "../c++/io.hpp" + +#include <iostream> + +#include "echo.hpp" + +saw::error_or<void> handle_echo_write(saw::io_stream& rmt_clt, message& state, uint64_t tbw){ + auto eov = rmt_clt.write(&state.data[state.already_written], tbw); + if(eov.is_error()){ + return std::move(eov.get_error()); + } + + auto val = eov.get_value(); + state.already_written += val; + if(state.already_written > state.already_read){ + exit(-1); + } + + return saw::void_t{}; +} + +void handle_echo_message(saw::io_stream& rmt_clt, bool& keep_running, message& state){ + rmt_clt.read_ready().then([&](){ + for(;;){ + uint64_t tbr = state.data.size() < state.already_read ? 0: state.data.size() - state.already_read; + if(tbr == 0){ + exit(-1); + } + auto eov = rmt_clt.read(&state.data[state.already_read], tbr); + + if(eov.is_error()){ + auto& err = eov.get_error(); + if(err.is_critical()){ + std::cerr<<err.get_category()<<std::endl; + exit(err.get_id()); + }else{ + break; + } + } + auto read_bytes = eov.get_value(); + if(read_bytes == 0u){ + exit(-1); + } + + bool trigger_write = (state.already_read == state.already_written); + + state.already_read += read_bytes; + if(state.already_read > state.data.size()){ + state.already_read = state.data.size(); + } + if(trigger_write){ + auto eov = handle_echo_write(rmt_clt, state, state.already_read - state.already_written); + if(eov.is_error()){ + auto& err = eov.get_error(); + if(err.is_critical()){ + std::cerr<<err.get_category()<<std::endl; + exit(err.get_id()); + }else { + break; + } + } + } + } + }).detach(); + + rmt_clt.write_ready().then([&](){ + for(;;){ + if(state.already_read < state.already_written){ + exit(-1); + } + uint64_t tbw = state.already_read - state.already_written; + if(tbw == 0){ + break; + } + + auto eov = handle_echo_write(rmt_clt, state, tbw); + if(eov.is_error()){ + auto& err = eov.get_error(); + if(err.is_critical()){ + std::cerr<<err.get_category()<<std::endl; + exit(err.get_id()); + }else { + break; + } + } + } + }).detach(); + + rmt_clt.on_read_disconnected().then([&](){ + keep_running = false; + }).detach(); +} + +int main(){ + auto eo_aio = saw::setup_async_io(); + if(eo_aio.is_error()){ + auto& err = eo_aio.get_error(); + std::cerr<<err.get_message()<<std::endl; + return err.get_id(); + } + auto& aio = eo_aio.get_value(); + /** + * Make the event loop the current event loop on this thread + */ + saw::wait_scope wait_scope{aio.event_loop}; + + bool keep_running = true; + aio.event_port.on_signal(saw::Signal::Terminate).then([&keep_running](){ + keep_running = false; + }).detach(); + + auto& network = aio.io->get_network(); + saw::own<saw::network_address> addr = nullptr; + saw::own<saw::server> srv = nullptr; + saw::own<saw::io_stream> remote_client = nullptr; + + message msg_state; + + std::cout<<"Starting to resolve address"<<std::endl; + /** + * Try to resolve address. If resolved + */ + network.resolve_address("127.0.0.1", 4322).then([&](auto net_addr){ + std::cout<<"Resolved address"<<std::endl; + addr = std::move(net_addr); + srv = network.listen(*addr); + if(srv){ + srv->accept().then([&](auto client) -> saw::error_or<void>{ + + if(!remote_client){ + std::cout<<"Accepted client"<<std::endl; + remote_client = std::move(client); + + if(remote_client){ + std::cout<<"Spinning up handler"<<std::endl; + handle_echo_message(*remote_client, keep_running, msg_state); + }else{ + keep_running = false; + } + } + return saw::make_error<saw::err::critical>(); + }).detach(); + }else{ + keep_running = false; + } + }).detach([&](auto err) { + std::cout<<"Failed to resolve address: "<<err.get_category()<<":"<<err.get_message()<<std::endl; + keep_running = false; + return err; + }); + + std::cout<<"Entering waiting loop"<<std::endl; + + wait_scope.poll(); + while(keep_running){ + wait_scope.wait(std::chrono::seconds{5}); + } + + std::cout<<"\n\nShutting down echo server"<<std::endl; + + return 0; +} |