Merge branch 'dev'
commit
ff5144d7a5
|
@ -61,7 +61,7 @@ public:
|
||||||
virtual Conveyor<size_t> writeDone() = 0;
|
virtual Conveyor<size_t> writeDone() = 0;
|
||||||
};
|
};
|
||||||
|
|
||||||
class AsyncIoStream : public AsyncInputStream, public AsyncOutputStream {
|
class AsyncIoStream final : public AsyncInputStream, public AsyncOutputStream {
|
||||||
private:
|
private:
|
||||||
Own<IoStream> stream;
|
Own<IoStream> stream;
|
||||||
|
|
||||||
|
@ -75,6 +75,9 @@ private:
|
||||||
public:
|
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;
|
void read(void *buffer, size_t length, size_t max_length) override;
|
||||||
|
|
||||||
Conveyor<size_t> readDone() override;
|
Conveyor<size_t> readDone() override;
|
||||||
|
|
|
@ -0,0 +1,42 @@
|
||||||
|
#pragma once
|
||||||
|
|
||||||
|
#include "async.h"
|
||||||
|
#include "message.h"
|
||||||
|
#include "io.h"
|
||||||
|
|
||||||
|
namespace saw {
|
||||||
|
|
||||||
|
template <typename Codec, typename Incoming, typename Outgoing, typename InContainer = MessageContainer<Incoming>, typename OutContainer = MessageContainer<Outgoing>, typename BufferT = RingBuffer>
|
||||||
|
class StreamingIoPeer {
|
||||||
|
private:
|
||||||
|
Own<ConveyorFeeder<HeapMessageRoot<Incoming, InContainer>>> incoming_feeder = nullptr;
|
||||||
|
|
||||||
|
Own<AsyncIoStream> io_stream;
|
||||||
|
|
||||||
|
Codec codec;
|
||||||
|
|
||||||
|
BufferT in_buffer;
|
||||||
|
BufferT out_buffer;
|
||||||
|
|
||||||
|
SinkConveyor sink_read;
|
||||||
|
SinkConveyor sink_write;
|
||||||
|
|
||||||
|
public:
|
||||||
|
StreamingIoPeer(Own<ConveyorFeeder<HeapMessageRoot<Incoming, InContainer>>> feed, Own<AsyncIoStream> stream, Codec codec, BufferT in, BufferT out);
|
||||||
|
StreamingIoPeer(Own<ConveyorFeeder<HeapMessageRoot<Incoming, InContainer>>> feed, Own<AsyncIoStream> stream);
|
||||||
|
|
||||||
|
void send(HeapMessageRoot<Outgoing, OutContainer> builder);
|
||||||
|
|
||||||
|
Conveyor<void> onReadDisconnected();
|
||||||
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Setup new streaming io peer with the provided network protocols.
|
||||||
|
* This is a convenience wrapper intended for a faster setup of
|
||||||
|
*/
|
||||||
|
template <typename Codec, typename Incoming, typename Outgoing, typename InContainer = MessageContainer<Incoming>, typename OutContainer = MessageContainer<Outgoing>, typename BufferT = RingBuffer>
|
||||||
|
std::pair<StreamingIoPeer<Codec, Incoming, Outgoing, InContainer, OutContainer, BufferT>, Conveyor<HeapMessageRoot<Incoming, InContainer>>> newStreamingIoPeer(Own<AsyncIoStream> stream);
|
||||||
|
|
||||||
|
} // namespace saw
|
||||||
|
|
||||||
|
#include "io_peer.tmpl.h"
|
|
@ -0,0 +1,96 @@
|
||||||
|
namespace saw {
|
||||||
|
|
||||||
|
template <typename Codec, typename Incoming, typename Outgoing, typename InContainer = MessageContainer<Incoming>, typename OutContainer = MessageContainer<Outgoing>, typename BufferT = RingBuffer>
|
||||||
|
StreamingIoPeer<Codec, Incoming, Outgoing, InContainer, OutContainer, BufferT>::StreamingIoPeer(
|
||||||
|
Own<ConveyorFeeder<HeapMessageRoot<Incoming, InContainer>>> feed,
|
||||||
|
Own<AsyncIoStream> str
|
||||||
|
):
|
||||||
|
StreamingIoPeer{std::move(feed), std::move(str), {}, {}, {}}
|
||||||
|
{
|
||||||
|
}
|
||||||
|
|
||||||
|
template <typename Codec, typename Incoming, typename Outgoing, typename InContainer = MessageContainer<Incoming>, typename OutContainer = MessageContainer<Outgoing>, typename BufferT = RingBuffer>
|
||||||
|
StreamingIoPeer<Codec, Incoming, Outgoing, InContainer, OutContainer, BufferT>::StreamingIoPeer(
|
||||||
|
Own<ConveyorFeeder<HeapMessageRoot<Incoming, InContainer>>> feed,
|
||||||
|
Own<AsyncIoStream> str, Codec codec, BufferT in, BufferT out):
|
||||||
|
incoming_feeder{std::move(feed)},
|
||||||
|
io_stream{std::move(str)},
|
||||||
|
codec{std::move(codec)},
|
||||||
|
in_buffer{std::move(in)},
|
||||||
|
out_buffer{std::move(out)},
|
||||||
|
sink_read{
|
||||||
|
io_stream->readDone().then([this](size_t bytes) -> ErrorOr<void> {
|
||||||
|
in_buffer.writeAdvance(bytes);
|
||||||
|
|
||||||
|
if(in_buffer.writeSegmentLength() == 0){
|
||||||
|
return criticalError("Message too long");
|
||||||
|
}
|
||||||
|
|
||||||
|
io_stream->read(&in_buffer.write(), 1, in_buffer.writeSegmentLength());
|
||||||
|
|
||||||
|
while(true){
|
||||||
|
auto root = heapMessageRoot<Incoming, InContainer>();
|
||||||
|
auto builder = root.build();
|
||||||
|
|
||||||
|
Error error = codec.template decode<Incoming, InContainer>(builder, in_buffer);
|
||||||
|
if(error.isCritical()){
|
||||||
|
return error;
|
||||||
|
}
|
||||||
|
|
||||||
|
if(!error.failed()){
|
||||||
|
incoming_feeder->handle(std::move(root));
|
||||||
|
}else{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return Void{};
|
||||||
|
}).sink([this](Error error){
|
||||||
|
incoming_feeder->fail(error);
|
||||||
|
|
||||||
|
return error;
|
||||||
|
})
|
||||||
|
},
|
||||||
|
sink_write{
|
||||||
|
io_stream->writeDone().then([this](size_bytes) -> ErrorOr<void> {
|
||||||
|
out_buffer.readAdvance(bytes);
|
||||||
|
if(out_buffer.readCompositeLength() > 0){
|
||||||
|
io_stream->write(&out_buffer.read(), out_buffer.readSegmengtLength());
|
||||||
|
}
|
||||||
|
|
||||||
|
return Void{};
|
||||||
|
}).sink();
|
||||||
|
}
|
||||||
|
{
|
||||||
|
io_stream->read(&in_buffer.write(), 1, in_buffer.writeSegmentLength());
|
||||||
|
}
|
||||||
|
|
||||||
|
template <typename Codec, typename Incoming, typename Outgoing, typename InContainer = MessageContainer<Incoming>, typename OutContainer = MessageContainer<Outgoing>, typename BufferT = RingBuffer>
|
||||||
|
void StreamingIoPeer<Codec, Incoming, Outgoing, InContainer, OutContainer, BufferT>::send(HeapMessageRoot<Outgoing, OutContainer> msg){
|
||||||
|
bool restart_write = out_buffer.readSegmentLength() == 0;
|
||||||
|
|
||||||
|
Error error = codec.template encode<Outgoing, OutContainer>(msg.read(), out_buffer);
|
||||||
|
if(error.failed()){
|
||||||
|
return error;
|
||||||
|
}
|
||||||
|
|
||||||
|
if(restart_write){
|
||||||
|
io_stream->write(&out_buffer.read(), out_buffer.readSegmentLength());
|
||||||
|
}
|
||||||
|
|
||||||
|
return noError();
|
||||||
|
}
|
||||||
|
|
||||||
|
template <typename Codec, typename Incoming, typename Outgoing, typename InContainer = MessageContainer<Incoming>, typename OutContainer = MessageContainer<Outgoing>, typename BufferT = RingBuffer>
|
||||||
|
Conveyor<void> StreamingIoPeer<Codec, Incoming, Outgoing, InContainer, OutContainer, BufferT>::onReadDisconnected(){
|
||||||
|
return io_stream->onReadDisconnected();
|
||||||
|
}
|
||||||
|
|
||||||
|
template <typename Codec, typename Incoming, typename Outgoing, typename InContainer = MessageContainer<Incoming>, typename OutContainer = MessageContainer<Outgoing>, typename BufferT = RingBuffer>
|
||||||
|
std::pair<StreamingIoPeer<Codec, Incoming, Outgoing, InContainer, OutContainer, BufferT>, Conveyor<HeapMessageRoot<Incoming, InContainer>>> newStreamingIoPeer(Own<AsyncIoStream> stream){
|
||||||
|
auto caf = newConveyorAndFeeder<HeapMessageRoot<Incoming, InContainer>>();
|
||||||
|
|
||||||
|
return {{std::move(caf.feeder), std::move(stream)}, std::move(caf.conveyor)};
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
|
@ -1,27 +0,0 @@
|
||||||
#pragma once
|
|
||||||
|
|
||||||
#include "async.h"
|
|
||||||
#include "message.h"
|
|
||||||
#include "io.h"
|
|
||||||
|
|
||||||
namespace saw {
|
|
||||||
|
|
||||||
template <typename Codec, typename Incoming, typename Outgoing, class InContainer = MessageContainer<Incoming>, class OutContainer = MessageContainer<Outgoing>>
|
|
||||||
class StreamingIoPeer {
|
|
||||||
private:
|
|
||||||
Codec codec;
|
|
||||||
|
|
||||||
Own<AsyncIoStream> io_stream;
|
|
||||||
|
|
||||||
Own<ConveyorFeeder<HeapMessageRoot<Incoming, InContainer>>> incoming_feeder = nullptr;
|
|
||||||
public:
|
|
||||||
StreamingIoPeer(Own<AsyncIoStream> stream);
|
|
||||||
|
|
||||||
void send(HeapMessageRoot<Outgoing, OutContainer> builder);
|
|
||||||
|
|
||||||
Conveyor<HeapMessageRootIncoming> startReadPump();
|
|
||||||
};
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
} // namespace saw
|
|
Loading…
Reference in New Issue