forstio
/
forstio
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

renaming things

This commit is contained in:
Claudius Holeksa 2022-12-05 17:17:49 +01:00
parent fc12627f54
commit 8f4614b264
20 changed files with 498 additions and 501 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
driver/io-unix.cpp
View File

@ -4,16 +4,16 @@
namespace saw {
namespace unix {
IFdOwner::IFdOwner(UnixEventPort &event_port, int file_descriptor, int fd_flags,
uint32_t event_mask)
: event_port{event_port}, file_descriptor{file_descriptor},
i_fd_owner::i_fd_owner(UnixEventPort &event_port, int file_descriptor,
int fd_flags, uint32_t event_mask)
: m_event_port{event_port}, file_descriptor{file_descriptor},
fd_flags{fd_flags}, event_mask{event_mask} {
event_port.subscribe(*this, file_descriptor, event_mask);
m_event_port.subscribe(*this, file_descriptor, event_mask);
}
IFdOwner::~IFdOwner() {
i_fd_owner::~i_fd_owner() {
if (file_descriptor >= 0) {
event_port.unsubscribe(file_descriptor);
m_event_port.unsubscribe(file_descriptor);
::close(file_descriptor);
}
}
@ -28,33 +28,33 @@ ssize_t unixWrite(int fd, const void *buffer, size_t length) {
UnixIoStream::UnixIoStream(UnixEventPort &event_port, int file_descriptor,
int fd_flags, uint32_t event_mask)
: IFdOwner{event_port, file_descriptor, fd_flags, event_mask | EPOLLRDHUP} {
}
: i_fd_owner{event_port, file_descriptor, fd_flags,
event_mask | EPOLLRDHUP} {}
ErrorOr<size_t> UnixIoStream::read(void *buffer, size_t length) {
error_or<size_t> UnixIoStream::read(void *buffer, size_t length) {
ssize_t read_bytes = unixRead(fd(), buffer, length);
if (read_bytes > 0) {
return static_cast<size_t>(read_bytes);
} else if (read_bytes == 0) {
return criticalError("Disconnected", Error::Code::Disconnected);
return critical_error("Disconnected", error::code::disconnected);
}
return recoverableError("Currently busy");
return recoverable_error("Currently busy");
}
Conveyor<void> UnixIoStream::readReady() {
auto caf = newConveyorAndFeeder<void>();
conveyor<void> UnixIoStream::readReady() {
auto caf = new_conveyor_and_feeder<void>();
read_ready = std::move(caf.feeder);
return std::move(caf.conveyor);
}
Conveyor<void> UnixIoStream::onReadDisconnected() {
auto caf = newConveyorAndFeeder<void>();
conveyor<void> UnixIoStream::onReadDisconnected() {
auto caf = new_conveyor_and_feeder<void>();
on_read_disconnect = std::move(caf.feeder);
return std::move(caf.conveyor);
}
ErrorOr<size_t> UnixIoStream::write(const void *buffer, size_t length) {
error_or<size_t> UnixIoStream::write(const void *buffer, size_t length) {
ssize_t write_bytes = unixWrite(fd(), buffer, length);
if (write_bytes > 0) {
return static_cast<size_t>(write_bytes);
@ -63,14 +63,14 @@ ErrorOr<size_t> UnixIoStream::write(const void *buffer, size_t length) {
int error = errno;
if (error == EAGAIN || error == EWOULDBLOCK) {
return recoverableError("Currently busy");
return recoverable_error("Currently busy");
}
return criticalError("Disconnected", Error::Code::Disconnected);
return critical_error("Disconnected", error::code::disconnected);
}
Conveyor<void> UnixIoStream::writeReady() {
auto caf = newConveyorAndFeeder<void>();
conveyor<void> UnixIoStream::writeReady() {
auto caf = new_conveyor_and_feeder<void>();
write_ready = std::move(caf.feeder);
return std::move(caf.conveyor);
}
@ -97,10 +97,10 @@ void UnixIoStream::notify(uint32_t mask) {
UnixServer::UnixServer(UnixEventPort &event_port, int file_descriptor,
int fd_flags)
: IFdOwner{event_port, file_descriptor, fd_flags, EPOLLIN} {}
: i_fd_owner{event_port, file_descriptor, fd_flags, EPOLLIN} {}
Conveyor<Own<IoStream>> UnixServer::accept() {
auto caf = newConveyorAndFeeder<Own<IoStream>>();
conveyor<own<IoStream>> UnixServer::accept() {
auto caf = new_conveyor_and_feeder<own<IoStream>>();
accept_feeder = std::move(caf.feeder);
return std::move(caf.conveyor);
}
@ -117,7 +117,7 @@ void UnixServer::notify(uint32_t mask) {
if (accept_fd < 0) {
return;
}
auto fd_stream = heap<UnixIoStream>(event_port, accept_fd, 0,
auto fd_stream = heap<UnixIoStream>(m_event_port, accept_fd, 0,
EPOLLIN | EPOLLOUT);
accept_feeder->feed(std::move(fd_stream));
}
@ -126,7 +126,7 @@ void UnixServer::notify(uint32_t mask) {
UnixDatagram::UnixDatagram(UnixEventPort &event_port, int file_descriptor,
int fd_flags)
: IFdOwner{event_port, file_descriptor, fd_flags, EPOLLIN | EPOLLOUT} {}
: i_fd_owner{event_port, file_descriptor, fd_flags, EPOLLIN | EPOLLOUT} {}
namespace {
ssize_t unix_read_msg(int fd, void *buffer, size_t length) {
@ -144,22 +144,22 @@ ssize_t unix_write_msg(int fd, const void *buffer, size_t length,
}
} // namespace
ErrorOr<size_t> UnixDatagram::read(void *buffer, size_t length) {
error_or<size_t> UnixDatagram::read(void *buffer, size_t length) {
ssize_t read_bytes = unix_read_msg(fd(), buffer, length);
if (read_bytes > 0) {
return static_cast<size_t>(read_bytes);
}
return recoverableError("Currently busy");
return recoverable_error("Currently busy");
}
Conveyor<void> UnixDatagram::readReady() {
auto caf = newConveyorAndFeeder<void>();
conveyor<void> UnixDatagram::readReady() {
auto caf = new_conveyor_and_feeder<void>();
read_ready = std::move(caf.feeder);
return std::move(caf.conveyor);
}
ErrorOr<size_t> UnixDatagram::write(const void *buffer, size_t length,
NetworkAddress &dest) {
error_or<size_t> UnixDatagram::write(const void *buffer, size_t length,
NetworkAddress &dest) {
UnixNetworkAddress &unix_dest = static_cast<UnixNetworkAddress &>(dest);
SocketAddress &sock_addr = unix_dest.unixAddress();
socklen_t sock_addr_length = sock_addr.getRawLength();
@ -168,11 +168,11 @@ ErrorOr<size_t> UnixDatagram::write(const void *buffer, size_t length,
if (write_bytes > 0) {
return static_cast<size_t>(write_bytes);
}
return recoverableError("Currently busy");
return recoverable_error("Currently busy");
}
Conveyor<void> UnixDatagram::writeReady() {
auto caf = newConveyorAndFeeder<void>();
conveyor<void> UnixDatagram::writeReady() {
auto caf = new_conveyor_and_feeder<void>();
write_ready = std::move(caf.feeder);
return std::move(caf.conveyor);
}
@ -239,7 +239,7 @@ UnixNetworkAddress &translate_to_unix_address_ref(
} // namespace
Own<Server> UnixNetwork::listen(NetworkAddress &addr) {
own<Server> UnixNetwork::listen(NetworkAddress &addr) {
auto unix_addr_storage =
translate_network_address_to_unix_network_address(addr);
UnixNetworkAddress &address =
@ -273,7 +273,7 @@ Own<Server> UnixNetwork::listen(NetworkAddress &addr) {
return heap<UnixServer>(event_port, fd, 0);
}
Conveyor<Own<IoStream>> UnixNetwork::connect(NetworkAddress &addr) {
conveyor<own<IoStream>> UnixNetwork::connect(NetworkAddress &addr) {
auto unix_addr_storage =
translate_network_address_to_unix_network_address(addr);
UnixNetworkAddress &address =
@ -281,15 +281,15 @@ Conveyor<Own<IoStream>> UnixNetwork::connect(NetworkAddress &addr) {
assert(address.unixAddressSize() > 0);
if (address.unixAddressSize() == 0) {
return Conveyor<Own<IoStream>>{criticalError("No address found")};
return conveyor<own<IoStream>>{critical_error("No address found")};
}
int fd = address.unixAddress(0).socket(SOCK_STREAM);
if (fd < 0) {
return Conveyor<Own<IoStream>>{criticalError("Couldn't open socket")};
return conveyor<own<IoStream>>{critical_error("Couldn't open socket")};
}
Own<UnixIoStream> io_stream =
own<UnixIoStream> io_stream =
heap<UnixIoStream>(event_port, fd, 0, EPOLLIN | EPOLLOUT);
bool success = false;
@ -308,7 +308,7 @@ Conveyor<Own<IoStream>> UnixNetwork::connect(NetworkAddress &addr) {
/// @todo Add limit node when implemented
if (error == EINPROGRESS) {
/*
Conveyor<void> write_ready = io_stream->writeReady();
conveyor<void> write_ready = io_stream->writeReady();
return write_ready.then(
[ios{std::move(io_stream)}]() mutable {
ios->write_ready = nullptr;
@ -319,8 +319,8 @@ Conveyor<Own<IoStream>> UnixNetwork::connect(NetworkAddress &addr) {
break;
} else if (error != EINTR) {
/// @todo Push error message from
return Conveyor<Own<IoStream>>{
criticalError("Couldn't connect")};
return conveyor<own<IoStream>>{
critical_error("Couldn't connect")};
}
} else {
success = true;
@ -329,13 +329,13 @@ Conveyor<Own<IoStream>> UnixNetwork::connect(NetworkAddress &addr) {
}
if (!success) {
return criticalError("Couldn't connect");
return critical_error("Couldn't connect");
}
return Conveyor<Own<IoStream>>{std::move(io_stream)};
return conveyor<own<IoStream>>{std::move(io_stream)};
}
Own<Datagram> UnixNetwork::datagram(NetworkAddress &addr) {
own<Datagram> UnixNetwork::datagram(NetworkAddress &addr) {
auto unix_addr_storage =
translate_network_address_to_unix_network_address(addr);
UnixNetworkAddress &address =
@ -376,7 +376,7 @@ size_t UnixNetworkAddress::unixAddressSize() const { return addresses.size(); }
UnixNetwork::UnixNetwork(UnixEventPort &event) : event_port{event} {}
Conveyor<Own<NetworkAddress>>
conveyor<own<NetworkAddress>>
UnixNetwork::resolveAddress(const std::string &path, uint16_t port_hint) {
std::string_view addr_view{path};
{
@ -389,40 +389,40 @@ UnixNetwork::resolveAddress(const std::string &path, uint16_t port_hint) {
std::vector<SocketAddress> addresses =
SocketAddress::resolve(addr_view, port_hint);
return Conveyor<Own<NetworkAddress>>{
return conveyor<own<NetworkAddress>>{
heap<UnixNetworkAddress>(path, port_hint, std::move(addresses))};
}
UnixIoProvider::UnixIoProvider(UnixEventPort &port_ref, Own<EventPort> port)
: event_port{port_ref}, event_loop{std::move(port)}, unix_network{
port_ref} {}
UnixIoProvider::UnixIoProvider(UnixEventPort &port_ref, own<event_port> port)
: m_event_port{port_ref}, m_event_loop{std::move(port)}, unix_network{
port_ref} {}
Own<InputStream> UnixIoProvider::wrapInputFd(int fd) {
return heap<UnixIoStream>(event_port, fd, 0, EPOLLIN);
own<InputStream> UnixIoProvider::wrapInputFd(int fd) {
return heap<UnixIoStream>(m_event_port, fd, 0, EPOLLIN);
}
Network &UnixIoProvider::network() {
return static_cast<Network &>(unix_network);
}
EventLoop &UnixIoProvider::eventLoop() { return event_loop; }
event_loop &UnixIoProvider::eventLoop() { return m_event_loop; }
} // namespace unix
ErrorOr<AsyncIoContext> setupAsyncIo() {
error_or<AsyncIoContext> setupAsyncIo() {
using namespace unix;
try {
Own<UnixEventPort> prt = heap<UnixEventPort>();
own<UnixEventPort> prt = heap<UnixEventPort>();
UnixEventPort &prt_ref = *prt;
Own<UnixIoProvider> io_provider =
own<UnixIoProvider> io_provider =
heap<UnixIoProvider>(prt_ref, std::move(prt));
EventLoop &loop_ref = io_provider->eventLoop();
event_loop &loop_ref = io_provider->eventLoop();
return {{std::move(io_provider), loop_ref, prt_ref}};
} catch (std::bad_alloc &) {
return criticalError("Out of memory");
return critical_error("Out of memory");
}
}
} // namespace saw

97
driver/io-unix.h
View File

@ -33,9 +33,9 @@ namespace unix {
constexpr int MAX_EPOLL_EVENTS = 256;
class UnixEventPort;
class IFdOwner {
class i_fd_owner {
protected:
UnixEventPort &event_port;
UnixEventPort &m_event_port;
private:
int file_descriptor;
@ -43,24 +43,25 @@ private:
uint32_t event_mask;
public:
IFdOwner(UnixEventPort &event_port, int file_descriptor, int fd_flags,
uint32_t event_mask);
i_fd_owner(UnixEventPort &event_port, int file_descriptor, int fd_flags,
uint32_t event_mask);
virtual ~IFdOwner();
virtual ~i_fd_owner();
virtual void notify(uint32_t mask) = 0;
int fd() const { return file_descriptor; }
};
class UnixEventPort final : public EventPort {
class UnixEventPort final : public event_port {
private:
int epoll_fd;
int signal_fd;
sigset_t signal_fd_set;
std::unordered_multimap<Signal, Own<ConveyorFeeder<void>>> signal_conveyors;
std::unordered_multimap<Signal, own<conveyor_feeder<void>>>
signal_conveyors;
int pipefds[2];
@ -137,8 +138,8 @@ private:
}
}
} else {
IFdOwner *owner =
reinterpret_cast<IFdOwner *>(events[i].data.ptr);
i_fd_owner *owner =
reinterpret_cast<i_fd_owner *>(events[i].data.ptr);
if (owner) {
owner->notify(events[i].events);
}
@ -187,8 +188,8 @@ public:
::close(pipefds[1]);
}
Conveyor<void> onSignal(Signal signal) override {
auto caf = newConveyorAndFeeder<void>();
conveyor<void> on_signal(Signal signal) override {
auto caf = new_conveyor_and_feeder<void>();
signal_conveyors.insert(std::make_pair(signal, std::move(caf.feeder)));
@ -200,8 +201,8 @@ public:
::sigprocmask(SIG_BLOCK, &signal_fd_set, nullptr);
::signalfd(signal_fd, &signal_fd_set, SFD_NONBLOCK | SFD_CLOEXEC);
auto node = Conveyor<void>::fromConveyor(std::move(caf.conveyor));
return Conveyor<void>::toConveyor(std::move(node));
auto node = conveyor<void>::from_conveyor(std::move(caf.conveyor));
return conveyor<void>::to_conveyor(std::move(node));
}
void poll() override { pollImpl(0); }
@ -234,7 +235,7 @@ public:
::send(pipefds[1], &i, sizeof(i), MSG_DONTWAIT);
}
void subscribe(IFdOwner &owner, int fd, uint32_t event_mask) {
void subscribe(i_fd_owner &owner, int fd, uint32_t event_mask) {
if (epoll_fd < 0 || fd < 0) {
return;
}
@ -263,67 +264,67 @@ public:
ssize_t unixRead(int fd, void *buffer, size_t length);
ssize_t unixWrite(int fd, const void *buffer, size_t length);
class UnixIoStream final : public IoStream, public IFdOwner {
class UnixIoStream final : public IoStream, public i_fd_owner {
private:
Own<ConveyorFeeder<void>> read_ready = nullptr;
Own<ConveyorFeeder<void>> on_read_disconnect = nullptr;
Own<ConveyorFeeder<void>> write_ready = nullptr;
own<conveyor_feeder<void>> read_ready = nullptr;
own<conveyor_feeder<void>> on_read_disconnect = nullptr;
own<conveyor_feeder<void>> write_ready = nullptr;
public:
UnixIoStream(UnixEventPort &event_port, int file_descriptor, int fd_flags,
uint32_t event_mask);
ErrorOr<size_t> read(void *buffer, size_t length) override;
error_or<size_t> read(void *buffer, size_t length) override;
Conveyor<void> readReady() override;
conveyor<void> readReady() override;
Conveyor<void> onReadDisconnected() override;
conveyor<void> onReadDisconnected() override;
ErrorOr<size_t> write(const void *buffer, size_t length) override;
error_or<size_t> write(const void *buffer, size_t length) override;
Conveyor<void> writeReady() override;
conveyor<void> writeReady() override;
/*
void read(void *buffer, size_t min_length, size_t max_length) override;
Conveyor<size_t> readDone() override;
Conveyor<void> readReady() override;
conveyor<size_t> readDone() override;
conveyor<void> readReady() override;
Conveyor<void> onReadDisconnected() override;
conveyor<void> onReadDisconnected() override;
void write(const void *buffer, size_t length) override;
Conveyor<size_t> writeDone() override;
Conveyor<void> writeReady() override;
conveyor<size_t> writeDone() override;
conveyor<void> writeReady() override;
*/
void notify(uint32_t mask) override;
};
class UnixServer final : public Server, public IFdOwner {
class UnixServer final : public Server, public i_fd_owner {
private:
Own<ConveyorFeeder<Own<IoStream>>> accept_feeder = nullptr;
own<conveyor_feeder<own<IoStream>>> accept_feeder = nullptr;
public:
UnixServer(UnixEventPort &event_port, int file_descriptor, int fd_flags);
Conveyor<Own<IoStream>> accept() override;
conveyor<own<IoStream>> accept() override;
void notify(uint32_t mask) override;
};
class UnixDatagram final : public Datagram, public IFdOwner {
class UnixDatagram final : public Datagram, public i_fd_owner {
private:
Own<ConveyorFeeder<void>> read_ready = nullptr;
Own<ConveyorFeeder<void>> write_ready = nullptr;
own<conveyor_feeder<void>> read_ready = nullptr;
own<conveyor_feeder<void>> write_ready = nullptr;
public:
UnixDatagram(UnixEventPort &event_port, int file_descriptor, int fd_flags);
ErrorOr<size_t> read(void *buffer, size_t length) override;
Conveyor<void> readReady() override;
error_or<size_t> read(void *buffer, size_t length) override;
conveyor<void> readReady() override;
ErrorOr<size_t> write(const void *buffer, size_t length,
NetworkAddress &dest) override;
Conveyor<void> writeReady() override;
error_or<size_t> write(const void *buffer, size_t length,
NetworkAddress &dest) override;
conveyor<void> writeReady() override;
void notify(uint32_t mask) override;
};
@ -435,31 +436,31 @@ private:
public:
UnixNetwork(UnixEventPort &event_port);
Conveyor<Own<NetworkAddress>>
conveyor<own<NetworkAddress>>
resolveAddress(const std::string &address, uint16_t port_hint = 0) override;
Own<Server> listen(NetworkAddress &addr) override;
own<Server> listen(NetworkAddress &addr) override;
Conveyor<Own<IoStream>> connect(NetworkAddress &addr) override;
conveyor<own<IoStream>> connect(NetworkAddress &addr) override;
Own<Datagram> datagram(NetworkAddress &addr) override;
own<Datagram> datagram(NetworkAddress &addr) override;
};
class UnixIoProvider final : public IoProvider {
private:
UnixEventPort &event_port;
EventLoop event_loop;
UnixEventPort &m_event_port;
event_loop m_event_loop;
UnixNetwork unix_network;
public:
UnixIoProvider(UnixEventPort &port_ref, Own<EventPort> port);
UnixIoProvider(UnixEventPort &port_ref, own<event_port> port);
Network &network() override;
Own<InputStream> wrapInputFd(int fd) override;
own<InputStream> wrapInputFd(int fd) override;
EventLoop &eventLoop();
event_loop &eventLoop();
};
} // namespace unix
} // namespace saw

23
source/forstio/async.h
View File

@ -151,10 +151,10 @@ template <typename T> T remove_error_or_type(T *);
template <typename T> T remove_error_or_type(error_or<T> *);
template <typename T>
using remove_error_or = decltype(removeerror_orType((T *)nullptr));
using remove_error_or = decltype(remove_error_or_type((T *)nullptr));
template <typename T>
using chained_conveyors = decltype(chainedconveyorType((T *)nullptr));
using chained_conveyors = decltype(chained_conveyor_type((T *)nullptr));
template <typename Func, typename T>
using conveyor_result =
@ -733,20 +733,19 @@ public:
err_or_val.as<unfix_void<remove_error_or<T>>>();
if (child_mixin.child) {
child_mixin.child->get_result(dep_eov);
if (dep_eov.isValue()) {
// try {
if (dep_eov.is_value()) {
try {
eov = fix_void_caller<T, DepT>::apply(
func, std::move(dep_eov.value()));
/*} catch (const std::bad_alloc &) {
eov = fix_void_caller<T, DepT>::apply(
func, std::move(dep_eov.value()));
} catch (const std::bad_alloc &) {
eov = critical_error("Out of memory");
} catch (const std::exception &) {
eov = critical_error(
"Exception in chain occured. Return error_or<T> if you "
"want to handle errors which are recoverable");
}
*/
} else if (dep_eov.iserror()) {
} else if (dep_eov.is_error()) {
eov = error_func(std::move(dep_eov.error()));
} else {
eov = critical_error("No value set in dependency");
@ -979,7 +978,7 @@ public:
template <typename T> class merge_conveyor_node_data {
public:
std::vector<own<typename merge_conveyor_node<T>::Appendage>> appendages;
std::vector<own<typename merge_conveyor_node<T>::appendage>> appendages;
merge_conveyor_node<T> *merger = nullptr;
@ -1000,7 +999,7 @@ template <typename... Args>
class JoinconveyorNode final : public JoinconveyorNodeBase {
private:
template<typename T>
class Appendage : public conveyorEventStorage {
class appendage : public conveyorEventStorage {
private:
maybe<T> data = std::nullopt;
@ -1012,7 +1011,7 @@ private:
void getResult(error_or_value& eov) override;
};
std::tuple<Appendage<Args>...> appendages;
std::tuple<appendage<Args>...> appendages;
public:
};

48
source/forstio/async.tmpl.h
View File

@ -124,18 +124,18 @@ conveyor_sink conveyor<void>::sink(ErrorFunc &&error_func) {
return conveyor_sink{std::move(sink_node)};
}
void detachconveyor(conveyor<void> &&conveyor);
void detach_conveyor(conveyor<void> &&conveyor);
template <typename T>
template <typename ErrorFunc>
void conveyor<T>::detach(ErrorFunc &&func) {
detachconveyor(std::move(then([](T &&) {}, std::move(func))));
detach_conveyor(std::move(then([](T &&) {}, std::move(func))));
}
template <>
template <typename ErrorFunc>
void conveyor<void>::detach(ErrorFunc &&func) {
detachconveyor(std::move(then([]() {}, std::move(func))));
detach_conveyor(std::move(then([]() {}, std::move(func))));
}
template <typename T>
@ -175,8 +175,8 @@ template <typename T> conveyor_and_feeder<T> new_conveyor_and_feeder() {
own<adapt_conveyor_node<fix_void<T>>> node =
heap<adapt_conveyor_node<fix_void<T>>>();
feeder->setFeedee(node.get());
node->setFeeder(feeder.get());
feeder->set_feedee(node.get());
node->set_feeder(feeder.get());
return conveyor_and_feeder<T>{std::move(feeder),
conveyor<T>::to_conveyor(std::move(node))};
@ -187,7 +187,7 @@ template <typename T> void queue_buffer_conveyor_node<T>::fire() {
if (child_mixin.child) {
if (!storage.empty()) {
if (storage.front().is_error()) {
if (storage.front().error().isCritical()) {
if (storage.front().error().is_critical()) {
child_mixin.child = nullptr;
}
}
@ -237,7 +237,7 @@ template <typename T> void queue_buffer_conveyor_node<T>::child_has_fired() {
child_mixin.child->get_result(eov);
if (eov.is_error()) {
if (eov.error().isCritical()) {
if (eov.error().is_critical()) {
}
}
@ -328,7 +328,7 @@ error_or<own<conveyor_node>>
merge_conveyor_node<T>::swap_child(own<conveyor_node> &&swapee_) noexcept {
(void)swapee_;
return critical_error(
"MergeconveyorNode<T>::Appendage should block calls to this class");
"MergeconveyorNode<T>::appendage should block calls to this class");
}
template <typename T>
@ -344,22 +344,22 @@ void merge_conveyor_node<T>::get_result(error_or_value &eov) noexcept {
for (size_t i = next_appendage; i < appendages.size(); ++i) {
if (appendages[i]->queued() > 0) {
err_or_val = std::move(appendages[i]->error_or_value.value());
appendages[i]->error_or_value = std::nullopt;
err_or_val = std::move(appendages[i]->error_or_value_data.value());
appendages[i]->error_or_value_data = std::nullopt;
next_appendage = i + 1;
return;
}
}
for (size_t i = 0; i < next_appendage; ++i) {
if (appendages[i]->queued() > 0) {
err_or_val = std::move(appendages[i]->error_or_value.value());
appendages[i]->error_or_value = std::nullopt;
err_or_val = std::move(appendages[i]->error_or_value_data.value());
appendages[i]->error_or_value_data = std::nullopt;
next_appendage = i + 1;
return;
}
}
err_or_val = critical_error("No value in Merge Appendages");
err_or_val = critical_error("No value in Merge appendages");
}
template <typename T> void merge_conveyor_node<T>::fire() {
@ -425,7 +425,7 @@ void merge_conveyor_node<T>::appendage::get_result(error_or_value &eov) {
SAW_ASSERT(queued() > 0) {
err_or_val =
critical_error("No element queued in Merge Appendage Node");
critical_error("No element queued in Merge appendage Node");
return;
}
@ -461,7 +461,7 @@ void merge_conveyor_node<T>::appendage::get_appendage_result(
SAW_ASSERT(queued() > 0) {
err_or_val =
critical_error("No element queued in Merge Appendage Node");
critical_error("No element queued in Merge appendage Node");
return;
}
@ -503,11 +503,11 @@ template <typename T>
void merge_conveyor_node_data<T>::attach(conveyor<T> conv) {
auto nas = conveyor<T>::from_conveyor(std::move(conv));
SAW_ASSERT(nas) { return; }
conveyor_storage *storage = nas->nextStorage();
conveyor_storage *storage = nas->next_storage();
SAW_ASSERT(storage) { return; }
auto merge_node_appendage =
heap<typename merge_conveyor_node<T>::Appendage>(std::move(nas),
heap<typename merge_conveyor_node<T>::appendage>(std::move(nas),
*merger);
auto merge_node_appendage_ptr = merge_node_appendage.get();
@ -515,10 +515,10 @@ void merge_conveyor_node_data<T>::attach(conveyor<T> conv) {
SAW_ASSERT(merger) { return; }
conveyor_storage *mrg_storage = merger->nextStorage();
conveyor_storage *mrg_storage = merger->next_storage();
SAW_ASSERT(mrg_storage) { return; }
merge_node_appendage->setParent(mrg_storage);
merge_node_appendage->set_parent(mrg_storage);
appendages.push_back(std::move(merge_node_appendage));
}
@ -531,7 +531,7 @@ void merge_conveyor_node_data<T>::governing_node_destroyed() {
template <typename T> adapt_conveyor_feeder<T>::~adapt_conveyor_feeder() {
if (feedee) {
feedee->setFeeder(nullptr);
feedee->set_feeder(nullptr);
feedee = nullptr;
}
}
@ -573,7 +573,7 @@ error adapt_conveyor_feeder<T>::swap(conveyor<T> &&conv) noexcept {
auto node = conveyor<T>::from_conveyor(std::move(conv));
feedee->swapChild(std::move(node));
feedee->swap_child(std::move(node));
return no_error();
}
@ -583,7 +583,7 @@ adapt_conveyor_node<T>::adapt_conveyor_node() : conveyor_event_storage{} {}
template <typename T> adapt_conveyor_node<T>::~adapt_conveyor_node() {
if (feeder) {
feeder->setFeedee(nullptr);
feeder->set_feedee(nullptr);
feeder = nullptr;
}
}
@ -676,7 +676,7 @@ template <typename T> void adapt_conveyor_node<T>::fire() {
template <typename T> one_time_conveyor_feeder<T>::~one_time_conveyor_feeder() {
if (feedee) {
feedee->setFeeder(nullptr);
feedee->set_feeder(nullptr);
feedee = nullptr;
}
}
@ -715,7 +715,7 @@ template <typename T> size_t one_time_conveyor_feeder<T>::space() const {
template <typename T> one_time_conveyor_node<T>::~one_time_conveyor_node() {
if (feeder) {
feeder->setFeedee(nullptr);
feeder->set_feedee(nullptr);
feeder = nullptr;
}
}

42
source/forstio/buffer.cpp
View File

@ -7,19 +7,19 @@
#include <sstream>
namespace saw {
Error Buffer::push(const uint8_t &value) {
error Buffer::push(const uint8_t &value) {
size_t write_remain = writeCompositeLength();
if (write_remain > 0) {
write() = value;
writeAdvance(1);
} else {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
return noError();
return no_error();
}
Error Buffer::push(const uint8_t &buffer, size_t size) {
Error error = writeRequireLength(size);
error Buffer::push(const uint8_t &buffer, size_t size) {
error error = writeRequireLength(size);
if (error.failed()) {
return error;
}
@ -31,20 +31,20 @@ Error Buffer::push(const uint8_t &buffer, size_t size) {
size -= segment;
buffer_ptr += segment;
}
return noError();
return no_error();
}
Error Buffer::pop(uint8_t &value) {
error Buffer::pop(uint8_t &value) {
if (readCompositeLength() > 0) {
value = read();
readAdvance(1);
} else {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
return noError();
return no_error();
}
Error Buffer::pop(uint8_t &buffer, size_t size) {
error Buffer::pop(uint8_t &buffer, size_t size) {
if (readCompositeLength() >= size) {
uint8_t *buffer_ptr = &buffer;
while (size > 0) {
@ -55,9 +55,9 @@ Error Buffer::pop(uint8_t &buffer, size_t size) {
buffer_ptr += segment;
}
} else {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
return noError();
return no_error();
}
std::string Buffer::toString() const {
@ -183,7 +183,7 @@ const uint8_t &BufferView::write(size_t i) const {
return buffer.write(pos);
}
Error BufferView::writeRequireLength(size_t bytes) {
error BufferView::writeRequireLength(size_t bytes) {
return buffer.writeRequireLength(bytes + write_offset);
}
@ -327,7 +327,7 @@ const uint8_t &RingBuffer::write(size_t i) const {
return buffer[pos];
}
/*
Error RingBuffer::increaseSize(size_t size){
error RingBuffer::increaseSize(size_t size){
size_t old_size = buffer.size();
size_t new_size = old_size + size;
buffer.resize(new_size);
@ -341,15 +341,15 @@ const uint8_t &RingBuffer::write(size_t i) const {
}
}
return noError();
return no_error();
}
*/
Error RingBuffer::writeRequireLength(size_t bytes) {
error RingBuffer::writeRequireLength(size_t bytes) {
size_t write_remain = writeCompositeLength();
if (bytes > write_remain) {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
return noError();
return no_error();
}
ArrayBuffer::ArrayBuffer(size_t size) : read_position{0}, write_position{0} {
@ -421,12 +421,12 @@ const uint8_t &ArrayBuffer::write(size_t i) const {
assert(i < writeCompositeLength());
return buffer[i + write_position];
}
Error ArrayBuffer::writeRequireLength(size_t bytes) {
error ArrayBuffer::writeRequireLength(size_t bytes) {
size_t write_remain = writeCompositeLength();
if (bytes > write_remain) {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
return noError();
return no_error();
}
} // namespace saw

18
source/forstio/buffer.h
View File

@ -40,12 +40,12 @@ public:
* There is nothing you can do if read hasn't been filled, but at
* least write can be increased if it is demanded.
*/
virtual Error writeRequireLength(size_t bytes) = 0;
virtual error writeRequireLength(size_t bytes) = 0;
Error push(const uint8_t &value);
Error push(const uint8_t &buffer, size_t size);
Error pop(uint8_t &value);
Error pop(uint8_t &buffer, size_t size);
error push(const uint8_t &value);
error push(const uint8_t &buffer, size_t size);
error pop(uint8_t &value);
error pop(uint8_t &buffer, size_t size);
std::string toString() const;
std::string toHex() const;
@ -80,7 +80,7 @@ public:
uint8_t &write(size_t i = 0) override;
const uint8_t &write(size_t i = 0) const override;
Error writeRequireLength(size_t bytes) override;
error writeRequireLength(size_t bytes) override;
size_t readOffset() const;
size_t writeOffset() const;
@ -126,7 +126,7 @@ public:
uint8_t &write(size_t i = 0) override;
const uint8_t &write(size_t i = 0) const override;
Error writeRequireLength(size_t bytes) override;
error writeRequireLength(size_t bytes) override;
};
/*
@ -158,7 +158,7 @@ public:
uint8_t &write(size_t i = 0) override;
const uint8_t &write(size_t i = 0) const override;
Error writeRequireLength(size_t bytes) override;
error writeRequireLength(size_t bytes) override;
};
class ChainArrayBuffer : public Buffer {
@ -187,6 +187,6 @@ public:
uint8_t &write(size_t i = 0) override;
const uint8_t &write(size_t i = 0) const override;
Error writeRequireLength(size_t bytes) override;
error writeRequireLength(size_t bytes) override;
};
} // namespace saw

10
source/forstio/error.cpp
View File

@ -54,18 +54,20 @@ error error::copy_error() const {
return error;
}
error::code error::code() const { return static_cast<error::code>(error_); }
error::code error::error_code() const {
return static_cast<error::code>(error_);
}
error makeerror(const std::string_view &generic, error::code code) {
error make_error(const std::string_view &generic, error::code code) {
return error{generic, code};
}
error critical_error(const std::string_view &generic, error::code c) {
return makeerror(generic, c);
return make_error(generic, c);
}
error recoverable_error(const std::string_view &generic, error::code c) {
return makeerror(generic, c);
return make_error(generic, c);
}
error no_error() { return error{}; }

8
source/forstio/error.h
View File

@ -53,16 +53,12 @@ error make_error(const std::string_view &generic, error::code c);
template <typename Formatter>
error make_error(const Formatter &formatter, error::code code,
const std::string_view &generic) {
/*
try {
std::string error_msg = formatter();
return error{std::move(error_msg), code};
} catch (std::bad_alloc &) {
return error{generic, code};
}
*/
std::string error_msg = formatter();
return error{std::move(error_msg), code};
}
error critical_error(const std::string_view &generic,
@ -72,7 +68,7 @@ template <typename Formatter>
error critical_error(const Formatter &formatter,
const std::string_view &generic,
error::code c = error::code::generic_critical) {
return makeerror(formatter, c, generic);
return make_error(formatter, c, generic);
}
error recoverable_error(const std::string_view &generic,
@ -82,7 +78,7 @@ template <typename Formatter>
error recoverable_error(const Formatter &formatter,
const std::string_view &generic,
error::code c = error::code::generic_recoverable) {
return makeerror(formatter, c, generic);
return make_error(formatter, c, generic);
}
error no_error();

14
source/forstio/io.cpp
View File

@ -4,7 +4,7 @@
namespace saw {
AsyncIoStream::AsyncIoStream(Own<IoStream> str)
AsyncIoStream::AsyncIoStream(own<IoStream> str)
: stream{std::move(str)}, read_ready{stream->readReady()
.then([this]() {
read_stepper.readStep(*stream);
@ -31,14 +31,14 @@ void AsyncIoStream::read(void *buffer, size_t min_length, size_t max_length) {
read_stepper.readStep(*stream);
}
Conveyor<size_t> AsyncIoStream::readDone() {
auto caf = newConveyorAndFeeder<size_t>();
conveyor<size_t> AsyncIoStream::readDone() {
auto caf = new_conveyor_and_feeder<size_t>();
read_stepper.read_done = std::move(caf.feeder);
return std::move(caf.conveyor);
}
Conveyor<void> AsyncIoStream::onReadDisconnected() {
auto caf = newConveyorAndFeeder<void>();
conveyor<void> AsyncIoStream::onReadDisconnected() {
auto caf = new_conveyor_and_feeder<void>();
read_stepper.on_read_disconnect = std::move(caf.feeder);
return std::move(caf.conveyor);
}
@ -53,8 +53,8 @@ void AsyncIoStream::write(const void *buffer, size_t length) {
write_stepper.writeStep(*stream);
}
Conveyor<size_t> AsyncIoStream::writeDone() {
auto caf = newConveyorAndFeeder<size_t>();
conveyor<size_t> AsyncIoStream::writeDone() {
auto caf = new_conveyor_and_feeder<size_t>();
write_stepper.write_done = std::move(caf.feeder);
return std::move(caf.conveyor);
}

64
source/forstio/io.h
View File

@ -15,11 +15,11 @@ class InputStream {
public:
virtual ~InputStream() = default;
virtual ErrorOr<size_t> read(void *buffer, size_t length) = 0;
virtual error_or<size_t> read(void *buffer, size_t length) = 0;
virtual Conveyor<void> readReady() = 0;
virtual conveyor<void> readReady() = 0;
virtual Conveyor<void> onReadDisconnected() = 0;
virtual conveyor<void> onReadDisconnected() = 0;
};
/*
@ -29,9 +29,9 @@ class OutputStream {
public:
virtual ~OutputStream() = default;
virtual ErrorOr<size_t> write(const void *buffer, size_t length) = 0;
virtual error_or<size_t> write(const void *buffer, size_t length) = 0;
virtual Conveyor<void> writeReady() = 0;
virtual conveyor<void> writeReady() = 0;
};
/*
@ -48,8 +48,8 @@ public:
virtual void read(void *buffer, size_t min_length, size_t max_length) = 0;
virtual Conveyor<size_t> readDone() = 0;
virtual Conveyor<void> onReadDisconnected() = 0;
virtual conveyor<size_t> readDone() = 0;
virtual conveyor<void> onReadDisconnected() = 0;
};
class AsyncOutputStream {
@ -58,42 +58,42 @@ public:
virtual void write(const void *buffer, size_t length) = 0;
virtual Conveyor<size_t> writeDone() = 0;
virtual conveyor<size_t> writeDone() = 0;
};
class AsyncIoStream final : public AsyncInputStream, public AsyncOutputStream {
private:
Own<IoStream> stream;
own<IoStream> stream;
ConveyorSink read_ready;
ConveyorSink write_ready;
ConveyorSink read_disconnected;
conveyor_sink read_ready;
conveyor_sink write_ready;
conveyor_sink read_disconnected;
ReadTaskAndStepHelper read_stepper;
WriteTaskAndStepHelper write_stepper;
public:
AsyncIoStream(Own<IoStream> str);
AsyncIoStream(own<IoStream> str);
SAW_FORBID_COPY(AsyncIoStream);
SAW_FORBID_MOVE(AsyncIoStream);
void read(void *buffer, size_t length, size_t max_length) override;
Conveyor<size_t> readDone() override;
conveyor<size_t> readDone() override;
Conveyor<void> onReadDisconnected() override;
conveyor<void> onReadDisconnected() override;
void write(const void *buffer, size_t length) override;
Conveyor<size_t> writeDone() override;
conveyor<size_t> writeDone() override;
};
class Server {
public:
virtual ~Server() = default;
virtual Conveyor<Own<IoStream>> accept() = 0;
virtual conveyor<own<IoStream>> accept() = 0;
};
class NetworkAddress;
@ -106,12 +106,12 @@ class Datagram {
public:
virtual ~Datagram() = default;
virtual ErrorOr<size_t> read(void *buffer, size_t length) = 0;
virtual Conveyor<void> readReady() = 0;
virtual error_or<size_t> read(void *buffer, size_t length) = 0;
virtual conveyor<void> readReady() = 0;
virtual ErrorOr<size_t> write(const void *buffer, size_t length,
NetworkAddress &dest) = 0;
virtual Conveyor<void> writeReady() = 0;
virtual error_or<size_t> write(const void *buffer, size_t length,
NetworkAddress &dest) = 0;
virtual conveyor<void> writeReady() = 0;
};
class OsNetworkAddress;
@ -158,7 +158,7 @@ public:
/**
* Resolve the provided string and uint16 to the preferred storage method
*/
virtual Conveyor<Own<NetworkAddress>>
virtual conveyor<own<NetworkAddress>>
resolveAddress(const std::string &addr, uint16_t port_hint = 0) = 0;
/**
@ -166,39 +166,39 @@ public:
* Since no dns request is made here, no async conveyors have to be used.
*/
/// @todo implement
// virtual Own<NetworkAddress> parseAddress(const std::string& addr,
// virtual own<NetworkAddress> parseAddress(const std::string& addr,
// uint16_t port_hint = 0) = 0;
/**
* Set up a listener on this address
*/
virtual Own<Server> listen(NetworkAddress &bind_addr) = 0;
virtual own<Server> listen(NetworkAddress &bind_addr) = 0;
/**
* Connect to a remote address
*/
virtual Conveyor<Own<IoStream>> connect(NetworkAddress &address) = 0;
virtual conveyor<own<IoStream>> connect(NetworkAddress &address) = 0;
/**
* Bind a datagram socket at this address.
*/
virtual Own<Datagram> datagram(NetworkAddress &address) = 0;
virtual own<Datagram> datagram(NetworkAddress &address) = 0;
};
class IoProvider {
public:
virtual ~IoProvider() = default;
virtual Own<InputStream> wrapInputFd(int fd) = 0;
virtual own<InputStream> wrapInputFd(int fd) = 0;
virtual Network &network() = 0;
};
struct AsyncIoContext {
Own<IoProvider> io;
EventLoop &event_loop;
EventPort &event_port;
own<IoProvider> io;
event_loop &event_loop;
event_port &event_port;
};
ErrorOr<AsyncIoContext> setupAsyncIo();
error_or<AsyncIoContext> setupAsyncIo();
} // namespace saw

10
source/forstio/io_auth.h
View File

@ -19,15 +19,15 @@ public:
// This is the easiest way to implement Authenticated streams.
// This is a simple pair of the stream and the peer.
Own<IoStream> stream;
Maybe<Own<Peer>> peer;
own<IoStream> stream;
maybe<own<Peer>> peer;
};
class AuthenticatedServer {
public:
virtual ~AuthenticatedServer() = default;
virtual Conveyor<AuthenticatedIoStream> accept() = 0;
virtual conveyor<AuthenticatedIoStream> accept() = 0;
};
/**
@ -41,13 +41,13 @@ public:
* Connects to the provided address.
* Returns as soon as it is authenticated or fails
*/
virtual Conveyor<AuthenticatedIoStream>
virtual conveyor<AuthenticatedIoStream>
connect(NetworkAddress &address) = 0;
/**
* Creates a server listening for connections
*/
virtual Own<AuthenticatedServer> listen() = 0;
virtual own<AuthenticatedServer> listen() = 0;
};
} // namespace saw

30
source/forstio/io_helpers.cpp
View File

@ -9,18 +9,18 @@ void ReadTaskAndStepHelper::readStep(InputStream &reader) {
while (read_task.has_value()) {
ReadIoTask &task = *read_task;
ErrorOr<size_t> n_err = reader.read(task.buffer, task.max_length);
if (n_err.isError()) {
const Error &error = n_err.error();
if (error.isCritical()) {
error_or<size_t> n_err = reader.read(task.buffer, task.max_length);
if (n_err.is_error()) {
const error &error = n_err.error();
if (error.is_critical()) {
if (read_done) {
read_done->fail(error.copyError());
read_done->fail(error.copy_error());
}
read_task = std::nullopt;
}
break;
} else if (n_err.isValue()) {
} else if (n_err.is_value()) {
size_t n = n_err.value();
if (static_cast<size_t>(n) >= task.min_length &&
static_cast<size_t>(n) <= task.max_length) {
@ -37,7 +37,7 @@ void ReadTaskAndStepHelper::readStep(InputStream &reader) {
} else {
if (read_done) {
read_done->fail(criticalError("Read failed"));
read_done->fail(critical_error("Read failed"));
}
read_task = std::nullopt;
}
@ -48,9 +48,9 @@ void WriteTaskAndStepHelper::writeStep(OutputStream &writer) {
while (write_task.has_value()) {
WriteIoTask &task = *write_task;
ErrorOr<size_t> n_err = writer.write(task.buffer, task.length);
error_or<size_t> n_err = writer.write(task.buffer, task.length);
if (n_err.isValue()) {
if (n_err.is_value()) {
size_t n = n_err.value();
assert(n <= task.length);
@ -64,22 +64,22 @@ void WriteTaskAndStepHelper::writeStep(OutputStream &writer) {
task.length -= n;
task.already_written += n;
}
} else if (n_err.isError()) {
const Error &error = n_err.error();
if (error.isCritical()) {
} else if (n_err.is_error()) {
const error &error = n_err.error();
if (error.is_critical()) {
if (write_done) {
write_done->fail(error.copyError());
write_done->fail(error.copy_error());
}
write_task = std::nullopt;
}
break;
} else {
if (write_done) {
write_done->fail(criticalError("Write failed"));
write_done->fail(critical_error("Write failed"));
}
write_task = std::nullopt;
}
}
}
} // namespace saw
} // namespace saw

6
source/forstio/io_helpers.h
View File

@ -27,9 +27,9 @@ public:
size_t already_read = 0;
};
std::optional<ReadIoTask> read_task;
Own<ConveyorFeeder<size_t>> read_done = nullptr;
own<conveyor_feeder<size_t>> read_done = nullptr;
Own<ConveyorFeeder<void>> on_read_disconnect = nullptr;
own<conveyor_feeder<void>> on_read_disconnect = nullptr;
public:
void readStep(InputStream &reader);
@ -45,7 +45,7 @@ public:
size_t already_written = 0;
};
std::optional<WriteIoTask> write_task;
Own<ConveyorFeeder<size_t>> write_done = nullptr;
own<conveyor_feeder<size_t>> write_done = nullptr;
public:
void writeStep(OutputStream &writer);

10
source/forstio/message.h
View File

@ -513,10 +513,10 @@ public:
template <class Schema, class Container = MessageContainer<Schema>>
class HeapMessageRoot {
private:
Own<Message<Schema, Container>> root;
own<Message<Schema, Container>> root;
public:
HeapMessageRoot(Own<Message<Schema, Container>> r) : root{std::move(r)} {}
HeapMessageRoot(own<Message<Schema, Container>> r) : root{std::move(r)} {}
typename Message<Schema, Container>::Builder build() {
assert(root);
@ -535,10 +535,10 @@ public:
using Schema = schema::Array<T>;
private:
Own<Message<Schema, Container>> root;
own<Message<Schema, Container>> root;
public:
HeapMessageRoot(Own<Message<Schema, Container>> r) : root{std::move(r)} {}
HeapMessageRoot(own<Message<Schema, Container>> r) : root{std::move(r)} {}
typename Message<Schema, Container>::Builder build(size_t size) {
assert(root);
@ -556,7 +556,7 @@ public:
*/
template <class Schema, class Container = MessageContainer<Schema>>
inline HeapMessageRoot<Schema, Container> heapMessageRoot() {
Own<Message<Schema, Container>> root = heap<Message<Schema, Container>>();
own<Message<Schema, Container>> root = heap<Message<Schema, Container>>();
return HeapMessageRoot<Schema, Container>{std::move(root)};
}
} // namespace saw

217
source/forstio/proto_kel.h
View File

@ -38,11 +38,11 @@ public:
const Version version() const { return Version{0, 0, 0}; }
template <class Schema, class Container = MessageContainer<Schema>>
Error encode(typename Message<Schema, Container>::Reader reader,
error encode(typename Message<Schema, Container>::Reader reader,
Buffer &buffer);
template <class Schema, class Container = MessageContainer<Schema>>
Error decode(typename Message<Schema, Container>::Builder builder,
error decode(typename Message<Schema, Container>::Builder builder,
Buffer &buffer, const Limits &limits = Limits{});
};
@ -50,12 +50,12 @@ template <class T> struct ProtoKelEncodeImpl;
template <class T, size_t N, class Container>
struct ProtoKelEncodeImpl<Message<schema::Primitive<T, N>, Container>> {
static Error
static error
encode(typename Message<schema::Primitive<T, N>, Container>::Reader data,
Buffer &buffer) {
Error error = StreamValue<typename PrimitiveTypeHelper<
error err = StreamValue<typename PrimitiveTypeHelper<
schema::Primitive<T, N>>::Type>::encode(data.get(), buffer);
return error;
return err;
}
static size_t
@ -67,26 +67,26 @@ struct ProtoKelEncodeImpl<Message<schema::Primitive<T, N>, Container>> {
template <class Container>
struct ProtoKelEncodeImpl<Message<schema::String, Container>> {
static Error
static error
encode(typename Message<schema::String, Container>::Reader data,
Buffer &buffer) {
std::string_view view = data.get();
size_t size = view.size();
Error error = buffer.writeRequireLength(sizeof(size) + size);
if (error.failed()) {
return error;
error err = buffer.writeRequireLength(sizeof(size) + size);
if (err.failed()) {
return err;
}
error = StreamValue<size_t>::encode(size, buffer);
if (error.failed()) {
return error;
err = StreamValue<size_t>::encode(size, buffer);
if (err.failed()) {
return err;
}
for (size_t i = 0; i < view.size(); ++i) {
buffer.write(i) = view[i];
}
buffer.writeAdvance(view.size());
return noError();
return no_error();
}
static size_t
@ -98,27 +98,27 @@ struct ProtoKelEncodeImpl<Message<schema::String, Container>> {
template <class... T, class Container>
struct ProtoKelEncodeImpl<Message<schema::Tuple<T...>, Container>> {
template <size_t i = 0>
static typename std::enable_if<i == sizeof...(T), Error>::type
static typename std::enable_if<i == sizeof...(T), error>::type
encodeMembers(typename Message<schema::Tuple<T...>, Container>::Reader,
Buffer &) {
return noError();
return no_error();
}
template <size_t i = 0>
static typename std::enable_if<(i < sizeof...(T)), Error>::type
static typename std::enable_if<(i < sizeof...(T)), error>::type
encodeMembers(typename Message<schema::Tuple<T...>, Container>::Reader data,
Buffer &buffer) {
Error error =
error err =
ProtoKelEncodeImpl<typename Container::template ElementType<i>>::
encode(data.template get<i>(), buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
return encodeMembers<i + 1>(data, buffer);
}
static Error
static error
encode(typename Message<schema::Tuple<T...>, Container>::Reader data,
Buffer &buffer) {
return encodeMembers<0>(data, buffer);
@ -149,29 +149,29 @@ template <typename... V, StringLiteral... K, class Container>
struct ProtoKelEncodeImpl<
Message<schema::Struct<schema::NamedMember<V, K>...>, Container>> {
template <size_t i = 0>
static typename std::enable_if<i == sizeof...(V), Error>::type
static typename std::enable_if<i == sizeof...(V), error>::type
encodeMembers(typename Message<schema::Struct<schema::NamedMember<V, K>...>,
Container>::Reader,
Buffer &) {
return noError();
return no_error();
}
template <size_t i = 0>
static typename std::enable_if <
i<sizeof...(V), Error>::type encodeMembers(
i<sizeof...(V), error>::type encodeMembers(
typename Message<schema::Struct<schema::NamedMember<V, K>...>,
Container>::Reader data,
Buffer &buffer) {
Error error =
error err =
ProtoKelEncodeImpl<typename Container::template ElementType<i>>::
encode(data.template get<i>(), buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
return encodeMembers<i + 1>(data, buffer);
}
static Error
static error
encode(typename Message<schema::Struct<schema::NamedMember<V, K>...>,
Container>::Reader data,
Buffer &buffer) {
@ -207,24 +207,23 @@ struct ProtoKelEncodeImpl<
Message<schema::Union<schema::NamedMember<V, K>...>, Container>> {
template <size_t i = 0>
static typename std::enable_if<i == sizeof...(V), Error>::type
static typename std::enable_if<i == sizeof...(V), error>::type
encodeMembers(typename Message<schema::Union<schema::NamedMember<V, K>...>,
Container>::Reader,
Buffer &) {
return noError();
return no_error();
}
template <size_t i = 0>
static typename std::enable_if <
i<sizeof...(V), Error>::type encodeMembers(
i<sizeof...(V), error>::type encodeMembers(
typename Message<schema::Union<schema::NamedMember<V, K>...>,
Container>::Reader reader,
Buffer &buffer) {
if (reader.index() == i) {
Error error =
StreamValue<ProtoKelCodec::UnionIdT>::encode(i, buffer);
if (error.failed()) {
return error;
error err = StreamValue<ProtoKelCodec::UnionIdT>::encode(i, buffer);
if (err.failed()) {
return err;
}
return ProtoKelEncodeImpl<typename Container::template ElementType<
i>>::encode(reader.template get<i>(), buffer);
@ -232,7 +231,7 @@ struct ProtoKelEncodeImpl<
return encodeMembers<i + 1>(reader, buffer);
}
static Error
static error
encode(typename Message<schema::Union<schema::NamedMember<V, K>...>,
Container>::Reader reader,
Buffer &buffer) {
@ -270,27 +269,27 @@ struct ProtoKelEncodeImpl<
template <class T, class Container>
struct ProtoKelEncodeImpl<Message<schema::Array<T>, Container>> {
static Error
static error
encode(typename Message<schema::Array<T>, Container>::Reader data,
Buffer &buffer) {
ProtoKelCodec::ArrayLengthT array_length = data.size();
{
Error error = StreamValue<ProtoKelCodec::ArrayLengthT>::encode(
error err = StreamValue<ProtoKelCodec::ArrayLengthT>::encode(
array_length, buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
for (size_t i = 0; i < array_length; ++i) {
Error error =
error err =
ProtoKelEncodeImpl<typename Container::ElementType>::encode(
data.get(i), buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
return noError();
return no_error();
}
/*
@ -316,76 +315,76 @@ template <typename T> struct ProtoKelDecodeImpl;
template <class T, size_t N, class Container>
struct ProtoKelDecodeImpl<Message<schema::Primitive<T, N>, Container>> {
static Error
static error
decode(typename Message<schema::Primitive<T, N>, Container>::Builder data,
Buffer &buffer) {
typename PrimitiveTypeHelper<schema::Primitive<T, N>>::Type val = 0;
Error error = StreamValue<typename PrimitiveTypeHelper<
error err = StreamValue<typename PrimitiveTypeHelper<
schema::Primitive<T, N>>::Type>::decode(val, buffer);
data.set(val);
return error;
return err;
}
};
template <class Container>
struct ProtoKelDecodeImpl<Message<schema::String, Container>> {
static Error
static error
decode(typename Message<schema::String, Container>::Builder data,
Buffer &buffer) {
size_t size = 0;
if (sizeof(size) > buffer.readCompositeLength()) {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
Error error = StreamValue<size_t>::decode(size, buffer);
if (error.failed()) {
return error;
error err = StreamValue<size_t>::decode(size, buffer);
if (err.failed()) {
return err;
}
if (size > buffer.readCompositeLength()) {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
std::string value;
value.resize(size);
if (size > buffer.readCompositeLength()) {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
for (size_t i = 0; i < value.size(); ++i) {
value[i] = buffer.read(i);
}
buffer.readAdvance(value.size());
data.set(std::move(value));
return noError();
return no_error();
}
};
template <class... T, class Container>
struct ProtoKelDecodeImpl<Message<schema::Tuple<T...>, Container>> {
template <size_t i = 0>
static typename std::enable_if<i == sizeof...(T), Error>::type
static typename std::enable_if<i == sizeof...(T), error>::type
decodeMembers(typename Message<schema::Tuple<T...>, Container>::Builder,
Buffer &) {
return noError();
return no_error();
}
template <size_t i = 0>
static typename std::enable_if <
i<sizeof...(T), Error>::type decodeMembers(
i<sizeof...(T), error>::type decodeMembers(
typename Message<schema::Tuple<T...>, Container>::Builder builder,
Buffer &buffer) {
Error error =
error err =
ProtoKelDecodeImpl<typename Container::template ElementType<i>>::
decode(builder.template init<i>(), buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
return decodeMembers<i + 1>(builder, buffer);
}
static Error
static error
decode(typename Message<schema::Tuple<T...>, Container>::Builder builder,
Buffer &buffer) {
return decodeMembers<0>(builder, buffer);
@ -397,30 +396,30 @@ struct ProtoKelDecodeImpl<
Message<schema::Struct<schema::NamedMember<V, K>...>, Container>> {
template <size_t i = 0>
static typename std::enable_if<i == sizeof...(V), Error>::type
static typename std::enable_if<i == sizeof...(V), error>::type
decodeMembers(typename Message<schema::Struct<schema::NamedMember<V, K>...>,
Container>::Builder,
Buffer &) {
return noError();
return no_error();
}
template <size_t i = 0>
static typename std::enable_if <
i<sizeof...(V), Error>::type decodeMembers(
i<sizeof...(V), error>::type decodeMembers(
typename Message<schema::Struct<schema::NamedMember<V, K>...>,
Container>::Builder builder,
Buffer &buffer) {
Error error =
error err =
ProtoKelDecodeImpl<typename Container::template ElementType<i>>::
decode(builder.template init<i>(), buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
return decodeMembers<i + 1>(builder, buffer);
}
static Error
static error
decode(typename Message<schema::Struct<schema::NamedMember<V, K>...>,
Container>::Builder builder,
Buffer &buffer) {
@ -432,42 +431,42 @@ template <class... V, StringLiteral... K, class Container>
struct ProtoKelDecodeImpl<
Message<schema::Union<schema::NamedMember<V, K>...>, Container>> {
template <size_t i = 0>
static typename std::enable_if<i == sizeof...(V), Error>::type
static typename std::enable_if<i == sizeof...(V), error>::type
decodeMembers(typename Message<schema::Union<schema::NamedMember<V, K>...>,
Container>::Builder,
Buffer &, ProtoKelCodec::UnionIdT) {
return noError();
return no_error();
}
template <size_t i = 0>
static typename std::enable_if <
i<sizeof...(V), Error>::type decodeMembers(
i<sizeof...(V), error>::type decodeMembers(
typename Message<schema::Union<schema::NamedMember<V, K>...>,
Container>::Builder builder,
Buffer &buffer, ProtoKelCodec::UnionIdT id) {
if (id == i) {
Error error =
error err =
ProtoKelDecodeImpl<typename Container::template ElementType<
i>>::decode(builder.template init<i>(), buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
return decodeMembers<i + 1>(builder, buffer, id);
}
static Error
static error
decode(typename Message<schema::Union<schema::NamedMember<V, K>...>,
Container>::Builder builder,
Buffer &buffer) {
ProtoKelCodec::UnionIdT id = 0;
Error error = StreamValue<ProtoKelCodec::UnionIdT>::decode(id, buffer);
if (error.failed()) {
return error;
error err = StreamValue<ProtoKelCodec::UnionIdT>::decode(id, buffer);
if (err.failed()) {
return err;
}
if (id >= sizeof...(V)) {
return criticalError("Union doesn't have this many id's");
return critical_error("Union doesn't have this many id's");
}
return decodeMembers<0>(builder, buffer, id);
@ -476,34 +475,34 @@ struct ProtoKelDecodeImpl<
template <class T, class Container>
struct ProtoKelDecodeImpl<Message<schema::Array<T>, Container>> {
static Error
static error
decode(typename Message<schema::Array<T>, Container>::Builder data,
Buffer &buffer) {
ProtoKelCodec::ArrayLengthT array_length = 0;
{
Error error = StreamValue<ProtoKelCodec::ArrayLengthT>::decode(
error err = StreamValue<ProtoKelCodec::ArrayLengthT>::decode(
array_length, buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
data.resize(array_length);
for (size_t i = 0; i < array_length; ++i) {
Error error =
error err =
ProtoKelDecodeImpl<typename Container::ElementType>::decode(
data.init(i), buffer);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
return noError();
return no_error();
}
};
template <class Schema, class Container>
Error ProtoKelCodec::encode(typename Message<Schema, Container>::Reader reader,
error ProtoKelCodec::encode(typename Message<Schema, Container>::Reader reader,
Buffer &buffer) {
BufferView view{buffer};
@ -512,48 +511,48 @@ Error ProtoKelCodec::encode(typename Message<Schema, Container>::Reader reader,
// Check the size of the packet for the first
// message length description
Error error = view.writeRequireLength(packet_length +
sizeof(ProtoKelCodec::PacketLengthT));
if (error.failed()) {
return error;
error err = view.writeRequireLength(packet_length +
sizeof(ProtoKelCodec::PacketLengthT));
if (err.failed()) {
return err;
}
{
Error error = StreamValue<ProtoKelCodec::PacketLengthT>::encode(
error err = StreamValue<ProtoKelCodec::PacketLengthT>::encode(
packet_length, view);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
{
Error error = ProtoKelEncodeImpl<Message<Schema, Container>>::encode(
error err = ProtoKelEncodeImpl<Message<Schema, Container>>::encode(
reader, view);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
buffer.writeAdvance(view.writeOffset());
return noError();
return no_error();
}
template <class Schema, class Container>
Error ProtoKelCodec::decode(
error ProtoKelCodec::decode(
typename Message<Schema, Container>::Builder builder, Buffer &buffer,
const Limits &limits) {
BufferView view{buffer};
ProtoKelCodec::PacketLengthT packet_length = 0;
{
Error error = StreamValue<ProtoKelCodec::PacketLengthT>::decode(
error err = StreamValue<ProtoKelCodec::PacketLengthT>::decode(
packet_length, view);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
if (packet_length > limits.packet_size) {
return criticalError(
return critical_error(
[packet_length]() {
return std::string{"Packet size too big: "} +
std::to_string(packet_length);
@ -562,21 +561,21 @@ Error ProtoKelCodec::decode(
}
{
Error error = ProtoKelDecodeImpl<Message<Schema, Container>>::decode(
error err = ProtoKelDecodeImpl<Message<Schema, Container>>::decode(
builder, view);
if (error.failed()) {
return error;
if (err.failed()) {
return err;
}
}
{
if (ProtoKelEncodeImpl<Message<Schema, Container>>::size(
builder.asReader()) != packet_length) {
return criticalError("Bad packet format");
return critical_error("Bad packet format");
}
}
buffer.readAdvance(view.readOffset());
return noError();
return no_error();
}
} // namespace saw

40
source/forstio/stream_endian.h
View File

@ -19,12 +19,12 @@ template <typename T, size_t size = sizeof(T)> class ShiftStreamValue;
template <typename T> class ShiftStreamValue<T, 1> {
public:
inline static Error decode(T &val, Buffer &buffer) {
inline static error decode(T &val, Buffer &buffer) {
uint8_t &raw = reinterpret_cast<uint8_t &>(val);
return buffer.pop(raw, sizeof(T));
}
inline static Error encode(const T &val, Buffer &buffer) {
inline static error encode(const T &val, Buffer &buffer) {
const uint8_t &raw = reinterpret_cast<const uint8_t &>(val);
return buffer.push(raw, sizeof(T));
}
@ -34,9 +34,9 @@ public:
template <typename T> class ShiftStreamValue<T, 2> {
public:
inline static Error decode(T &val, Buffer &buffer) {
inline static error decode(T &val, Buffer &buffer) {
if (buffer.readCompositeLength() < sizeof(T)) {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
uint16_t raw = 0;
@ -47,11 +47,11 @@ public:
memcpy(&val, &raw, sizeof(T));
buffer.readAdvance(sizeof(T));
return noError();
return no_error();
}
inline static Error encode(const T &val, Buffer &buffer) {
Error error = buffer.writeRequireLength(sizeof(T));
inline static error encode(const T &val, Buffer &buffer) {
error error = buffer.writeRequireLength(sizeof(T));
if (error.failed()) {
return error;
}
@ -64,7 +64,7 @@ public:
}
buffer.writeAdvance(sizeof(T));
return noError();
return no_error();
}
inline static size_t size() { return sizeof(T); }
@ -72,9 +72,9 @@ public:
template <typename T> class ShiftStreamValue<T, 4> {
public:
inline static Error decode(T &val, Buffer &buffer) {
inline static error decode(T &val, Buffer &buffer) {
if (buffer.readCompositeLength() < sizeof(T)) {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
uint32_t raw = 0;
@ -85,11 +85,11 @@ public:
memcpy(&val, &raw, sizeof(T));
buffer.readAdvance(sizeof(T));
return noError();
return no_error();
}
inline static Error encode(const T &val, Buffer &buffer) {
Error error = buffer.writeRequireLength(sizeof(T));
inline static error encode(const T &val, Buffer &buffer) {
error error = buffer.writeRequireLength(sizeof(T));
if (error.failed()) {
return error;
}
@ -102,7 +102,7 @@ public:
}
buffer.writeAdvance(sizeof(T));
return noError();
return no_error();
}
inline static size_t size() { return sizeof(T); }
@ -110,9 +110,9 @@ public:
template <typename T> class ShiftStreamValue<T, 8> {
public:
inline static Error decode(T &val, Buffer &buffer) {
inline static error decode(T &val, Buffer &buffer) {
if (buffer.readCompositeLength() < sizeof(T)) {
return recoverableError("Buffer too small");
return recoverable_error("Buffer too small");
}
uint64_t raw = 0;
@ -124,11 +124,11 @@ public:
memcpy(&val, &raw, sizeof(T));
buffer.readAdvance(sizeof(T));
return noError();
return no_error();
}
inline static Error encode(const T &val, Buffer &buffer) {
Error error = buffer.writeRequireLength(sizeof(T));
inline static error encode(const T &val, Buffer &buffer) {
error error = buffer.writeRequireLength(sizeof(T));
if (error.failed()) {
return error;
}
@ -141,7 +141,7 @@ public:
}
buffer.writeAdvance(sizeof(T));
return noError();
return no_error();
}
inline static size_t size() { return sizeof(T); }

72
source/forstio/tls/tls.cpp
View File

@ -40,58 +40,58 @@ Tls::Impl &Tls::getImpl() { return *impl; }
class TlsIoStream final : public IoStream {
private:
Own<IoStream> internal;
own<IoStream> internal;
gnutls_session_t session_handle;
public:
TlsIoStream(Own<IoStream> internal_) : internal{std::move(internal_)} {}
TlsIoStream(own<IoStream> internal_) : internal{std::move(internal_)} {}
~TlsIoStream() { gnutls_bye(session_handle, GNUTLS_SHUT_RDWR); }
ErrorOr<size_t> read(void *buffer, size_t length) override {
error_or<size_t> read(void *buffer, size_t length) override {
ssize_t size = gnutls_record_recv(session_handle, buffer, length);
if (size < 0) {
if(gnutls_error_is_fatal(size) == 0){
return recoverableError([size](){return std::string{"Read recoverable Error "}+std::string{gnutls_strerror(size)};}, "Error read r");
return recoverable_error([size](){return std::string{"Read recoverable error "}+std::string{gnutls_strerror(size)};}, "error read r");
}else{
return criticalError([size](){return std::string{"Read critical Error "}+std::string{gnutls_strerror(size)};}, "Error read c");
return critical_error([size](){return std::string{"Read critical error "}+std::string{gnutls_strerror(size)};}, "error read c");
}
}else if(size == 0){
return criticalError("Disconnected");
return critical_error("Disconnected");
}
return static_cast<size_t>(length);
}
Conveyor<void> readReady() override { return internal->readReady(); }
conveyor<void> readReady() override { return internal->readReady(); }
Conveyor<void> onReadDisconnected() override {
conveyor<void> onReadDisconnected() override {
return internal->onReadDisconnected();
}
ErrorOr<size_t> write(const void *buffer, size_t length) override {
error_or<size_t> write(const void *buffer, size_t length) override {
ssize_t size = gnutls_record_send(session_handle, buffer, length);
if(size < 0){
if(gnutls_error_is_fatal(size) == 0){
return recoverableError([size](){return std::string{"Write recoverable Error "}+std::string{gnutls_strerror(size)} + " " + std::to_string(size);}, "Error write r");
return recoverable_error([size](){return std::string{"Write recoverable error "}+std::string{gnutls_strerror(size)} + " " + std::to_string(size);}, "error write r");
}else{
return criticalError([size](){return std::string{"Write critical Error "}+std::string{gnutls_strerror(size)} + " " + std::to_string(size);}, "Error write c");
return critical_error([size](){return std::string{"Write critical error "}+std::string{gnutls_strerror(size)} + " " + std::to_string(size);}, "error write c");
}
}
return static_cast<size_t>(size);
}
Conveyor<void> writeReady() override { return internal->writeReady(); }
conveyor<void> writeReady() override { return internal->writeReady(); }
gnutls_session_t &session() { return session_handle; }
};
TlsServer::TlsServer(Own<Server> srv) : internal{std::move(srv)} {}
TlsServer::TlsServer(own<Server> srv) : internal{std::move(srv)} {}
Conveyor<Own<IoStream>> TlsServer::accept() {
SAW_ASSERT(internal) { return Conveyor<Own<IoStream>>{FixVoid<Own<IoStream>>{nullptr}}; }
return internal->accept().then([](Own<IoStream> stream) -> Own<IoStream> {
conveyor<own<IoStream>> TlsServer::accept() {
SAW_ASSERT(internal) { return conveyor<own<IoStream>>{fix_void<own<IoStream>>{nullptr}}; }
return internal->accept().then([](own<IoStream> stream) -> own<IoStream> {
/// @todo handshake
@ -105,14 +105,14 @@ namespace {
*/
struct TlsClientStreamHelper {
public:
Own<ConveyorFeeder<Own<IoStream>>> feeder;
ConveyorSink connection_sink;
ConveyorSink stream_reader;
ConveyorSink stream_writer;
own<conveyor_feeder<own<IoStream>>> feeder;
conveyor_sink connection_sink;
conveyor_sink stream_reader;
conveyor_sink stream_writer;
Own<TlsIoStream> stream = nullptr;
own<TlsIoStream> stream = nullptr;
public:
TlsClientStreamHelper(Own<ConveyorFeeder<Own<IoStream>>> f):
TlsClientStreamHelper(own<conveyor_feeder<own<IoStream>>> f):
feeder{std::move(f)}
{}
@ -145,7 +145,7 @@ public:
} while ( (ret == GNUTLS_E_AGAIN || ret == GNUTLS_E_INTERRUPTED) && gnutls_error_is_fatal(ret) == 0);
if(gnutls_error_is_fatal(ret)){
feeder->fail(criticalError("Couldn't create Tls connection"));
feeder->fail(critical_error("Couldn't create Tls connection"));
stream = nullptr;
}else if(ret == GNUTLS_E_SUCCESS){
feeder->feed(std::move(stream));
@ -155,19 +155,19 @@ public:
};
}
Own<Server> TlsNetwork::listen(NetworkAddress& address) {
own<Server> TlsNetwork::listen(NetworkAddress& address) {
return heap<TlsServer>(internal.listen(address));
}
Conveyor<Own<IoStream>> TlsNetwork::connect(NetworkAddress& address) {
conveyor<own<IoStream>> TlsNetwork::connect(NetworkAddress& address) {
// Helper setups
auto caf = newConveyorAndFeeder<Own<IoStream>>();
Own<TlsClientStreamHelper> helper = heap<TlsClientStreamHelper>(std::move(caf.feeder));
auto caf = new_conveyor_and_feeder<own<IoStream>>();
own<TlsClientStreamHelper> helper = heap<TlsClientStreamHelper>(std::move(caf.feeder));
TlsClientStreamHelper* hlp_ptr = helper.get();
// Conveyor entangled structure
// conveyor entangled structure
auto prim_conv = internal.connect(address).then([this, hlp_ptr, addr = address.address()](
Own<IoStream> stream) -> ErrorOr<void> {
own<IoStream> stream) -> error_or<void> {
IoStream* inner_stream = stream.get();
auto tls_stream = heap<TlsIoStream>(std::move(stream));
@ -201,7 +201,7 @@ Conveyor<Own<IoStream>> TlsNetwork::connect(NetworkAddress& address) {
return caf.conveyor.attach(std::move(helper));
}
Own<Datagram> TlsNetwork::datagram(NetworkAddress& address){
own<Datagram> TlsNetwork::datagram(NetworkAddress& address){
///@unimplemented
return nullptr;
}
@ -213,8 +213,8 @@ static ssize_t forst_tls_push_func(gnutls_transport_ptr_t p, const void *data,
return -1;
}
ErrorOr<size_t> length = stream->write(data, size);
if (length.isError() || !length.isValue()) {
error_or<size_t> length = stream->write(data, size);
if (length.is_error() || !length.is_value()) {
return -1;
}
@ -227,8 +227,8 @@ static ssize_t forst_tls_pull_func(gnutls_transport_ptr_t p, void *data, size_t
return -1;
}
ErrorOr<size_t> length = stream->read(data, size);
if (length.isError() || !length.isValue()) {
error_or<size_t> length = stream->read(data, size);
if (length.is_error() || !length.is_value()) {
return -1;
}
@ -237,14 +237,14 @@ static ssize_t forst_tls_pull_func(gnutls_transport_ptr_t p, void *data, size_t
TlsNetwork::TlsNetwork(Tls& tls_, Network &network) : tls{tls_},internal{network} {}
Conveyor<Own<NetworkAddress>> TlsNetwork::resolveAddress(const std::string &addr,
conveyor<own<NetworkAddress>> TlsNetwork::resolveAddress(const std::string &addr,
uint16_t port) {
/// @todo tls server name needed. Check validity. Won't matter later on, because gnutls should fail anyway. But
/// it's better to find the error source sooner rather than later
return internal.resolveAddress(addr, port);
}
std::optional<Own<TlsNetwork>> setupTlsNetwork(Network &network) {
std::optional<own<TlsNetwork>> setupTlsNetwork(Network &network) {
return std::nullopt;
}
} // namespace saw

18
source/forstio/tls/tls.h
View File

@ -11,12 +11,12 @@ class Tls;
class TlsServer final : public Server {
private:
Own<Server> internal;
own<Server> internal;
public:
TlsServer(Own<Server> srv);
TlsServer(own<Server> srv);
Conveyor<Own<IoStream>> accept() override;
conveyor<own<IoStream>> accept() override;
};
class TlsNetwork final : public Network {
@ -26,13 +26,13 @@ private:
public:
TlsNetwork(Tls& tls_, Network &network_);
Conveyor<Own<NetworkAddress>> resolveAddress(const std::string &addr, uint16_t port = 0) override;
conveyor<own<NetworkAddress>> resolveAddress(const std::string &addr, uint16_t port = 0) override;
Own<Server> listen(NetworkAddress& address) override;
own<Server> listen(NetworkAddress& address) override;
Conveyor<Own<IoStream>> connect(NetworkAddress& address) override;
conveyor<own<IoStream>> connect(NetworkAddress& address) override;
Own<Datagram> datagram(NetworkAddress& address) override;
own<Datagram> datagram(NetworkAddress& address) override;
};
/**
@ -42,7 +42,7 @@ public:
class Tls {
private:
class Impl;
Own<Impl> impl;
own<Impl> impl;
public:
Tls();
~Tls();
@ -65,6 +65,6 @@ private:
Options options;
};
std::optional<Own<TlsNetwork>> setupTlsNetwork(Network &network);
std::optional<own<TlsNetwork>> setupTlsNetwork(Network &network);
} // namespace saw

136
test/async.cpp
View File

@ -6,38 +6,38 @@ namespace {
SAW_TEST("Async Immediate"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
Conveyor<size_t> number{5};
conveyor<size_t> number{5};
Conveyor<bool> is_number = number.then([](size_t val){
conveyor<bool> is_number = number.then([](size_t val){
return val == 5;
});
wait_scope.poll();
ErrorOr<bool> error_or_number = is_number.take();
error_or<bool> error_or_number = is_number.take();
SAW_EXPECT(!error_or_number.isError(), error_or_number.error().message());
SAW_EXPECT(error_or_number.isValue(), "Return is not a value");
SAW_EXPECT(!error_or_number.is_error(), error_or_number.error().message());
SAW_EXPECT(error_or_number.is_value(), "Return is not a value");
SAW_EXPECT(error_or_number.value(), "Value is not 5");
}
SAW_TEST("Async Adapt"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
auto feeder_conveyor = newConveyorAndFeeder<size_t>();
auto feeder_conveyor = new_conveyor_and_feeder<size_t>();
feeder_conveyor.feeder->feed(5);
ErrorOr<size_t> foo = feeder_conveyor.conveyor.take();
error_or<size_t> foo = feeder_conveyor.conveyor.take();
SAW_EXPECT(!foo.isError(), foo.error().message());
SAW_EXPECT(foo.isValue(), "Return is not a value");
SAW_EXPECT(!foo.is_error(), foo.error().message());
SAW_EXPECT(foo.is_value(), "Return is not a value");
SAW_EXPECT(foo.value() == 5, "Values not 5, but " + std::to_string(foo.value()));
}
@ -45,72 +45,72 @@ SAW_TEST("Async Adapt"){
SAW_TEST("Async Adapt Multiple"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
auto feeder_conveyor = newConveyorAndFeeder<size_t>();
auto feeder_conveyor = new_conveyor_and_feeder<size_t>();
feeder_conveyor.feeder->feed(5);
ErrorOr<size_t> foo = feeder_conveyor.conveyor.take();
error_or<size_t> foo = feeder_conveyor.conveyor.take();
SAW_EXPECT(!foo.isError(), foo.error().message());
SAW_EXPECT(foo.isValue(), "Return is not a value");
SAW_EXPECT(!foo.is_error(), foo.error().message());
SAW_EXPECT(foo.is_value(), "Return is not a value");
SAW_EXPECT(foo.value() == 5, "Values not 5, but " + std::to_string(foo.value()));
feeder_conveyor.feeder->feed(10);
ErrorOr<size_t> bar = feeder_conveyor.conveyor.take();
error_or<size_t> bar = feeder_conveyor.conveyor.take();
SAW_EXPECT(!foo.isError(), bar.error().message());
SAW_EXPECT(bar.isValue(), "Return is not a value");
SAW_EXPECT(!foo.is_error(), bar.error().message());
SAW_EXPECT(bar.is_value(), "Return is not a value");
SAW_EXPECT(bar.value() == 10, "Values not 10, but " + std::to_string(bar.value()));
feeder_conveyor.feeder->feed(2);
feeder_conveyor.feeder->feed(4234);
ErrorOr<size_t> a = feeder_conveyor.conveyor.take();
ErrorOr<size_t> b = feeder_conveyor.conveyor.take();
error_or<size_t> a = feeder_conveyor.conveyor.take();
error_or<size_t> b = feeder_conveyor.conveyor.take();
SAW_EXPECT(!foo.isError(), a.error().message());
SAW_EXPECT(a.isValue(), "Return is not a value");
SAW_EXPECT(!foo.is_error(), a.error().message());
SAW_EXPECT(a.is_value(), "Return is not a value");
SAW_EXPECT(a.value() == 2, "Values not 2, but " + std::to_string(a.value()));
SAW_EXPECT(!foo.isError(), b.error().message());
SAW_EXPECT(b.isValue(), "Return is not a value");
SAW_EXPECT(!foo.is_error(), b.error().message());
SAW_EXPECT(b.is_value(), "Return is not a value");
SAW_EXPECT(b.value() == 4234, "Values not 4234, but " + std::to_string(b.value()));
}
SAW_TEST("Async Conversion"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
auto feeder_conveyor = newConveyorAndFeeder<size_t>();
auto feeder_conveyor = new_conveyor_and_feeder<size_t>();
Conveyor<std::string> string_conveyor = feeder_conveyor.conveyor.then([](size_t foo){
conveyor<std::string> string_conveyor = feeder_conveyor.conveyor.then([](size_t foo){
return std::to_string(foo);
});
feeder_conveyor.feeder->feed(10);
ErrorOr<std::string> foo = string_conveyor.take();
error_or<std::string> foo = string_conveyor.take();
SAW_EXPECT(!foo.isError(), foo.error().message());
SAW_EXPECT(foo.isValue(), "Return is not a value");
SAW_EXPECT(!foo.is_error(), foo.error().message());
SAW_EXPECT(foo.is_value(), "Return is not a value");
SAW_EXPECT(foo.value() == std::to_string(10), "Values is not 10, but " + foo.value());
}
SAW_TEST("Async Conversion Multistep"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
auto feeder_conveyor = newConveyorAndFeeder<size_t>();
auto feeder_conveyor = new_conveyor_and_feeder<size_t>();
Conveyor<bool> conveyor = feeder_conveyor.conveyor.then([](size_t foo){
conveyor<bool> conveyor = feeder_conveyor.conveyor.then([](size_t foo){
return std::to_string(foo);
}).then([](const std::string& value){
return value != "10";
@ -120,29 +120,29 @@ SAW_TEST("Async Conversion Multistep"){
feeder_conveyor.feeder->feed(10);
ErrorOr<bool> foo = conveyor.take();
error_or<bool> foo = conveyor.take();
SAW_EXPECT(!foo.isError(), foo.error().message());
SAW_EXPECT(foo.isValue(), "Return is not a value");
SAW_EXPECT(!foo.is_error(), foo.error().message());
SAW_EXPECT(foo.is_value(), "Return is not a value");
SAW_EXPECT(foo.value(), "Values is not true");
}
SAW_TEST("Async Scheduling"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
auto feeder_conveyor = newConveyorAndFeeder<size_t>();
auto feeder_conveyor = new_conveyor_and_feeder<size_t>();
/*
* Attach node test data
*/
Own<size_t> counter = heap<size_t>();
own<size_t> counter = heap<size_t>();
size_t* ctr_ptr = counter.get();
*ctr_ptr = 0;
Conveyor<std::string> string_conveyor = feeder_conveyor.conveyor
conveyor<std::string> string_conveyor = feeder_conveyor.conveyor
.then([ctr_ptr](size_t foo){
return std::to_string(foo + ++(*ctr_ptr));
})
@ -163,34 +163,34 @@ SAW_TEST("Async Scheduling"){
wait_scope.poll();
ErrorOr<std::string> foo_10 = string_conveyor.take();
error_or<std::string> foo_10 = string_conveyor.take();
SAW_EXPECT(!foo_10.isError(), foo_10.error().message());
SAW_EXPECT(foo_10.isValue(), "Return is not a value");
SAW_EXPECT(!foo_10.is_error(), foo_10.error().message());
SAW_EXPECT(foo_10.is_value(), "Return is not a value");
SAW_EXPECT(foo_10.value() == (std::string{"pre"} + std::to_string(11) + std::string{"post"}), "Values is not pre11post, but " + foo_10.value());
ErrorOr<std::string> foo_20 = string_conveyor.take();
error_or<std::string> foo_20 = string_conveyor.take();
SAW_EXPECT(!foo_20.isError(), foo_20.error().message());
SAW_EXPECT(foo_20.isValue(), "Return is not a value");
SAW_EXPECT(!foo_20.is_error(), foo_20.error().message());
SAW_EXPECT(foo_20.is_value(), "Return is not a value");
SAW_EXPECT(foo_20.value() == (std::string{"pre"} + std::to_string(22) + std::string{"post"}), "Values is not pre22post, but " + foo_20.value());
ErrorOr<std::string> foo_30 = string_conveyor.take();
error_or<std::string> foo_30 = string_conveyor.take();
SAW_EXPECT(!foo_30.isError(), foo_30.error().message());
SAW_EXPECT(foo_30.isValue(), "Return is not a value");
SAW_EXPECT(!foo_30.is_error(), foo_30.error().message());
SAW_EXPECT(foo_30.is_value(), "Return is not a value");
SAW_EXPECT(foo_30.value() == (std::string{"pre"} + std::to_string(33) + std::string{"post"}), "Values is not pre33post, but " + foo_30.value());
}
SAW_TEST("Async Detach"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
int num = 0;
Conveyor<int>{10}.then([&num](int bar){
conveyor<int>{10}.then([&num](int bar){
num = bar;
}).detach();
@ -202,14 +202,14 @@ SAW_TEST("Async Detach"){
SAW_TEST("Async Merge"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
auto cam = Conveyor<int>{10}.merge();
auto cam = conveyor<int>{10}.merge();
cam.second.attach(Conveyor<int>{11});
cam.second.attach(conveyor<int>{11});
cam.second.attach(Conveyor<int>{14});
cam.second.attach(conveyor<int>{14});
size_t elements_passed = 0;
bool wrong_value = false;
@ -231,12 +231,12 @@ SAW_TEST("Async Merge"){
SAW_TEST("Async Connect"){
using namespace saw;
EventLoop event_loop;
WaitScope wait_scope{event_loop};
event_loop event_loop;
wait_scope wait_scope{event_loop};
auto caf1 = newConveyorAndFeeder<int>();
auto caf1 = new_conveyor_and_feeder<int>();
auto caf2 = newConveyorAndFeeder<float>();
auto caf2 = new_conveyor_and_feeder<float>();
bool val_passed = false;
auto conveyor1 = caf1.conveyor.then([&val_passed](int val) -> float{

20
test/proto_kel.cpp
View File

@ -35,7 +35,7 @@ SAW_TEST("Primitive Encoding"){
RingBuffer temp_buffer;
ProtoKelCodec codec;
Error error = codec.encode<TestSize>(root.read(), temp_buffer);
error error = codec.encode<TestSize>(root.read(), temp_buffer);
SAW_EXPECT(!error.failed(), error.message());
SAW_EXPECT(temp_buffer.readCompositeLength() == (sizeof(value)+sizeof(ProtoKelCodec::PacketLengthT)), "Bad Size: " + std::to_string(temp_buffer.readCompositeLength()));
@ -57,7 +57,7 @@ SAW_TEST("List Encoding"){
RingBuffer buffer;
ProtoKelCodec codec;
Error error = codec.encode<TestTuple>(root.read(), buffer);
error error = codec.encode<TestTuple>(root.read(), buffer);
SAW_EXPECT(!error.failed(), error.message());
SAW_EXPECT(buffer.readCompositeLength() == 14, "Bad Size: " + std::to_string(buffer.readCompositeLength()));
@ -83,7 +83,7 @@ SAW_TEST("Struct Encoding"){
RingBuffer buffer;
ProtoKelCodec codec;
Error error = codec.encode<TestStruct>(builder.asReader(), buffer);
error error = codec.encode<TestStruct>(builder.asReader(), buffer);
SAW_EXPECT(!error.failed(), error.message());
SAW_EXPECT(buffer.readCompositeLength() == 40, "Bad Size: " + std::to_string(buffer.readCompositeLength()));
@ -103,7 +103,7 @@ SAW_TEST("Union Encoding"){
RingBuffer buffer;
ProtoKelCodec codec;
Error error = codec.encode<TestUnion>(builder.asReader(), buffer);
error error = codec.encode<TestUnion>(builder.asReader(), buffer);
SAW_EXPECT(!error.failed(), error.message());
SAW_EXPECT(buffer.readCompositeLength() == 16, "Bad Size: " + std::to_string(buffer.readCompositeLength()));
@ -120,7 +120,7 @@ SAW_TEST("Union Encoding"){
RingBuffer buffer;
ProtoKelCodec codec;
Error error = codec.encode<TestUnion>(builder.asReader(), buffer);
error error = codec.encode<TestUnion>(builder.asReader(), buffer);
SAW_EXPECT(!error.failed(), error.message());
SAW_EXPECT(buffer.readCompositeLength() == 23, "Bad Size: " + std::to_string(buffer.readCompositeLength()));
@ -141,7 +141,7 @@ SAW_TEST("Tuple Decoding"){
auto root = heapMessageRoot<TestTuple>();
auto builder = root.build();
Error error = codec.decode<TestTuple>(builder, buffer);
error error = codec.decode<TestTuple>(builder, buffer);
SAW_EXPECT(!error.failed(), error.message());
auto reader = builder.asReader();
@ -164,7 +164,7 @@ SAW_TEST("Struct Decoding"){
auto root = heapMessageRoot<TestStruct>();
auto builder = root.build();
Error error = codec.decode<TestStruct>(builder, buffer);
error error = codec.decode<TestStruct>(builder, buffer);
auto reader = builder.asReader();
auto foo_string = reader.get<"test_string">();
@ -188,7 +188,7 @@ SAW_TEST("Union Decoding"){
auto builder = root.build();
auto reader = builder.asReader();
Error error = codec.decode<TestUnion>(builder, buffer);
error error = codec.decode<TestUnion>(builder, buffer);
SAW_EXPECT(!error.failed(), error.message());
SAW_EXPECT(reader.hasAlternative<"test_string">(), "Wrong union value");
@ -220,8 +220,8 @@ SAW_TEST("Array Encoding"){
RingBuffer buffer;
Error error = codec.encode<TestArrayStruct>(root.read(), buffer);
error error = codec.encode<TestArrayStruct>(root.read(), buffer);
SAW_EXPECT(!error.failed(), "Error occured");
SAW_EXPECT(!error.failed(), "error occured");
}
}