2 files changed, 381 insertions, 46 deletions

diff --git a/src/network/protocol.cpp b/src/network/protocol.cpp
index b781693..89d6a88 100644
--- a/src/network/protocol.cpp
+++ b/src/network/protocol.cpp
@@ -15,16 +15,22 @@ using std::move;
 using std::nullopt;
 using std::runtime_error;
 using std::scoped_lock;
+using std::to_string;
+using std::unique_lock;
 using std::visit;
 
 namespace erebos {
 
+static constexpr uint8_t maxStreamNumber = 0x3F;
+
 struct NetworkProtocol::ConnectionPriv
 {
 	Connection::Id id() const;
 
+	size_t mtu() const;
 	bool send(const PartialStorage &, Header,
 			const vector<Object> &, bool secure);
+	bool send( const StreamData & chunk );
 
 	NetworkProtocol * protocol;
 	const sockaddr_in6 peerAddress;
@@ -37,7 +43,12 @@ struct NetworkProtocol::ConnectionPriv
 	ChannelState channel = monostate();
 	vector<vector<uint8_t>> secureOutQueue {};
 
+	size_t mtuLower = 1000; // TODO: MTU
+
 	vector<uint64_t> toAcknowledge {};
+
+	vector< shared_ptr< InStream >> inStreams {};
+	vector< shared_ptr< OutStream >> outStreams {};
 };
 
 
@@ -80,16 +91,35 @@ NetworkProtocol::PollResult NetworkProtocol::poll()
 		scoped_lock lock(protocolMutex);
 
 		for (const auto & c : connections) {
+			vector< StreamData > streamChunks;
+			bool sendAck = false;
 			{
 				scoped_lock clock(c->cmutex);
-				if (c->toAcknowledge.empty())
-					continue;
-
-				if (not holds_alternative<unique_ptr<Channel>>(c->channel))
-					continue;
+				sendAck = not c->toAcknowledge.empty() &&
+					holds_alternative< unique_ptr< Channel >>( c->channel );
+
+				for (auto & s : c->outStreams) {
+					unique_lock slock(s->streamMutex);
+					while (s->hasDataLocked())
+						streamChunks.push_back( s->getNextChunkLocked( c->mtu() ));
+					if( s->closed ){
+						// TODO: wait after ack
+						streamChunks.push_back( { s->id, (uint8_t) s->nextSequence, {} } );
+						slock.unlock();
+						s.reset();
+					}
+				}
+
+				while( not c->outStreams.empty() && not c->outStreams.back() )
+					c->outStreams.pop_back();
+			}
+			if (sendAck) {
+				auto pst = self->ref()->storage().deriveEphemeralStorage();
+				c->send(pst, Header {{}}, {}, true);
+			}
+			for (const auto & chunk : streamChunks) {
+				c->send( chunk );
 			}
-			auto pst = self->ref()->storage().deriveEphemeralStorage();
-			c->send(pst, Header {{}}, {}, true);
 		}
 	}
 
@@ -110,7 +140,8 @@ NetworkProtocol::PollResult NetworkProtocol::poll()
 
 		auto pst = self->ref()->storage().deriveEphemeralStorage();
 		optional<uint64_t> secure = false;
-		if (auto header = Connection::parsePacket(buffer, nullptr, pst, secure)) {
+		auto parsed = Connection::parsePacket(buffer, nullptr, pst, secure);
+		if (const auto * header = get_if< Header >( &parsed )) {
 			if (auto conn = verifyNewConnection(*header, addr))
 				return NewConnection { move(*conn) };
 
@@ -275,6 +306,8 @@ bool NetworkProtocol::verifyCookie(variant<sockaddr_in, sockaddr_in6> vaddr, con
 /* Connection                                                                 */
 /******************************************************************************/
 
+using Connection = NetworkProtocol::Connection;
+
 NetworkProtocol::Connection::Id NetworkProtocol::ConnectionPriv::id() const
 {
 	return reinterpret_cast<uintptr_t>(this);
@@ -312,6 +345,24 @@ const sockaddr_in6 & NetworkProtocol::Connection::peerAddress() const
 	return p->peerAddress;
 }
 
+size_t Connection::mtu() const
+{
+	return p->mtu();
+}
+
+size_t NetworkProtocol::ConnectionPriv::mtu() const
+{
+	if( get_if< unique_ptr< Channel >>( &channel ))
+		return mtuLower // space for:
+			- 1   // "encrypted" tag
+			- 1   // counter
+			- 1   // channel number
+			- 1   // channel sequence
+			- 16  // tag
+			;
+	return mtuLower - 128; // some space for cookie headers
+}
+
 optional<NetworkProtocol::Header> NetworkProtocol::Connection::receive(const PartialStorage & partStorage)
 {
 	vector<uint8_t> buf;
@@ -335,13 +386,14 @@ optional<NetworkProtocol::Header> NetworkProtocol::Connection::receive(const Par
 	}
 
 	optional<uint64_t> secure = false;
-	if (auto header = parsePacket(buf, channel, partStorage, secure)) {
+	auto parsed = parsePacket(buf, channel, partStorage, secure);
+	if (const auto * header = get_if< Header >( &parsed )) {
 		scoped_lock lock(p->cmutex);
 
 		if (secure) {
 			if (header->isAcknowledged())
 				p->toAcknowledge.push_back(*secure);
-			return header;
+			return *header;
 		}
 
 		if (const auto * cookieEcho = header->lookupFirst<Header::CookieEcho>()) {
@@ -353,13 +405,13 @@ optional<NetworkProtocol::Header> NetworkProtocol::Connection::receive(const Par
 			if (const auto * cookieSet = header->lookupFirst<Header::CookieSet>())
 				p->receivedCookie = cookieSet->value;
 
-			return header;
+			return *header;
 		}
 
 		if (holds_alternative<monostate>(p->channel)) {
 			if (const auto * cookieSet = header->lookupFirst<Header::CookieSet>()) {
 				p->receivedCookie = cookieSet->value;
-				return header;
+				return *header;
 			}
 		}
 
@@ -368,10 +420,36 @@ optional<NetworkProtocol::Header> NetworkProtocol::Connection::receive(const Par
 			return nullopt;
 		}
 	}
+	else if( auto * sdata = get_if< StreamData >( &parsed )){
+		scoped_lock lock(p->cmutex);
+		if (secure)
+			p->toAcknowledge.push_back(*secure);
+
+		InStream * stream = nullptr;
+		for (const auto & s : p->inStreams) {
+			if (s->id == sdata->id) {
+				stream = s.get();
+				break;
+			}
+		}
+		if (not stream) {
+			std::cerr << "unexpected stream number\n";
+			return nullopt;
+		}
+
+		stream->writeChunk( move(*sdata) );
+		if( stream->closed )
+			p->inStreams.erase(
+					std::remove_if( p->inStreams.begin(), p->inStreams.end(),
+						[&]( auto & sptr ) { return sptr.get() == stream; } ),
+					p->inStreams.end() );
+		return nullopt;
+	}
 	return nullopt;
 }
 
-optional<NetworkProtocol::Header> NetworkProtocol::Connection::parsePacket(vector<uint8_t> & buf,
+variant< monostate, NetworkProtocol::Header, NetworkProtocol::StreamData >
+NetworkProtocol::Connection::parsePacket(vector<uint8_t> & buf,
 		Channel * channel, const PartialStorage & partStorage,
 		optional<uint64_t> & secure)
 {
@@ -384,7 +462,7 @@ optional<NetworkProtocol::Header> NetworkProtocol::Connection::parsePacket(vecto
 	if ((buf[0] & 0xE0) == 0x80) {
 		if (not channel) {
 			std::cerr << "unexpected encrypted packet\n";
-			return nullopt;
+			return monostate();
 		}
 
 		if ((secure = channel->decrypt(buf.begin() + 1, buf.end(), decrypted, 0))) {
@@ -395,9 +473,17 @@ optional<NetworkProtocol::Header> NetworkProtocol::Connection::parsePacket(vecto
 				plainBegin = decrypted.begin() + 1;
 				plainEnd = decrypted.end();
 			}
+			else if (decrypted[0] <= maxStreamNumber) {
+				StreamData sdata;
+				sdata.id = decrypted[0];
+				sdata.sequence = decrypted[1];
+				sdata.data.resize( decrypted.size() - 2 );
+				std::copy(decrypted.begin() + 2, decrypted.end(), sdata.data.begin());
+				return sdata;
+			}
 			else {
-				std::cerr << "streams not implemented\n";
-				return nullopt;
+				std::cerr << "unexpected stream header\n";
+				return monostate();
 			}
 		}
 	}
@@ -414,12 +500,12 @@ optional<NetworkProtocol::Header> NetworkProtocol::Connection::parsePacket(vecto
 				pos = std::get<1>(*cdec);
 			}
 
-			return header;
+			return *header;
 		}
 	}
 
 	std::cerr << "invalid packet\n";
-	return nullopt;
+	return monostate();
 }
 
 bool NetworkProtocol::Connection::send(const PartialStorage & partStorage,
@@ -483,6 +569,37 @@ bool NetworkProtocol::ConnectionPriv::send(const PartialStorage & partStorage,
 	return true;
 }
 
+bool NetworkProtocol::Connection::send( const StreamData & chunk )
+{
+	return p->send( chunk );
+}
+
+bool NetworkProtocol::ConnectionPriv::send( const StreamData & chunk )
+{
+	vector<uint8_t> data, out;
+
+	{
+		scoped_lock clock( cmutex );
+
+		Channel * channel = nullptr;
+		if (auto uptr = get_if< unique_ptr< Channel >>( &this->channel ))
+			channel = uptr->get();
+		if (not channel)
+			return false;
+
+		data.push_back( chunk.id );
+		data.push_back( static_cast< uint8_t >( chunk.sequence ));
+		data.insert( data.end(), chunk.data.begin(), chunk.data.end() );
+
+		out.push_back( 0x80 );
+		channel->encrypt( data.begin(), data.end(), out, 1 );
+	}
+
+	protocol->sendto( out, peerAddress );
+	return true;
+}
+
+
 void NetworkProtocol::Connection::close()
 {
 	if (not p)
@@ -502,6 +619,33 @@ void NetworkProtocol::Connection::close()
 	p = nullptr;
 }
 
+shared_ptr< NetworkProtocol::InStream > NetworkProtocol::Connection::openInStream( uint8_t sid )
+{
+	scoped_lock lock( p->cmutex );
+	for (const auto & s : p->inStreams)
+		if (s->id == sid)
+			throw runtime_error("inbound stream " + to_string(sid) + " already open");
+
+	p->inStreams.emplace_back( new InStream( sid ));
+	return p->inStreams.back();
+}
+
+shared_ptr< NetworkProtocol::OutStream > NetworkProtocol::Connection::openOutStream()
+{
+	scoped_lock lock( p->cmutex );
+
+	uint8_t sid = 1;
+	if( not p->outStreams.empty() ){
+		if( p->outStreams.back()->id < maxStreamNumber )
+			sid = p->outStreams.back()->id + 1;
+		else
+			throw runtime_error("no free outbound stream");
+	}
+
+	p->outStreams.emplace_back( new OutStream( sid ));
+	return p->outStreams.back();
+}
+
 NetworkProtocol::ChannelState & NetworkProtocol::Connection::channel()
 {
 	return p->channel;
@@ -530,6 +674,127 @@ void NetworkProtocol::Connection::trySendOutQueue()
 }
 
 
+NetworkProtocol::Stream::Stream(uint8_t id_):
+	id(id_)
+{
+	readPtr = readBuffer.begin();
+}
+
+void NetworkProtocol::Stream::close()
+{
+	scoped_lock lock( streamMutex );
+	closed = true;
+}
+
+bool NetworkProtocol::Stream::hasDataLocked() const
+{
+	return not writeBuffer.empty() || readPtr < readBuffer.end();
+}
+
+size_t NetworkProtocol::Stream::writeLocked( const uint8_t * buf, size_t size )
+{
+	writeBuffer.insert( writeBuffer.end(), buf, buf + size );
+	return size;
+}
+
+size_t NetworkProtocol::Stream::readLocked( uint8_t * buf, size_t size )
+{
+	size_t res = 0;
+	if (readPtr < readBuffer.end()) {
+		res = std::min( size, static_cast< size_t >( readBuffer.end() - readPtr ));
+		std::copy_n( readPtr, res, buf );
+		readPtr += res;
+	}
+	if (res < size && not writeBuffer.empty()) {
+		std::swap( readBuffer, writeBuffer );
+		readPtr = readBuffer.begin();
+		writeBuffer.clear();
+		return res + readLocked( buf + res, size - res );
+	}
+	return res;
+}
+
+bool NetworkProtocol::InStream::isComplete() const
+{
+	scoped_lock lock( streamMutex );
+	return closed && outOfOrderChunks.empty();
+}
+
+vector< uint8_t > NetworkProtocol::InStream::readAll()
+{
+	scoped_lock lock( streamMutex );
+	if (readBuffer.empty()) {
+		vector< uint8_t > res;
+		std::swap( res, writeBuffer );
+		return res;
+	}
+
+	readBuffer.insert( readBuffer.end(), writeBuffer.begin(), writeBuffer.end() );
+	writeBuffer.clear();
+
+	vector< uint8_t > res;
+	std::swap( res, readBuffer );
+	readPtr = readBuffer.begin();
+	return res;
+}
+
+size_t NetworkProtocol::InStream::read( uint8_t * buf, size_t size )
+{
+	scoped_lock lock( streamMutex );
+	return readLocked( buf, size );
+}
+
+void NetworkProtocol::InStream::writeChunk( StreamData chunk )
+{
+	scoped_lock lock( streamMutex );
+	if( tryUseChunkLocked( chunk )) {
+		auto it = outOfOrderChunks.begin();
+		while( it != outOfOrderChunks.end() && tryUseChunkLocked( *it ))
+			it++;
+		outOfOrderChunks.erase( outOfOrderChunks.begin(), it );
+	} else {
+		auto it = outOfOrderChunks.begin();
+		while( it < outOfOrderChunks.end() &&
+				it->sequence - static_cast< uint8_t >( nextSequence )
+				< chunk.sequence - static_cast< uint8_t >( nextSequence ))
+			it++;
+		outOfOrderChunks.insert( it, move(chunk) );
+	}
+}
+
+bool NetworkProtocol::InStream::tryUseChunkLocked( const StreamData & chunk )
+{
+	if( chunk.sequence != static_cast< uint8_t >( nextSequence ))
+		return false;
+
+	if( chunk.data.empty() )
+		closed = true;
+	else
+		writeLocked( chunk.data.data(), chunk.data.size() );
+	nextSequence++;
+	return true;
+}
+
+size_t NetworkProtocol::OutStream::write( const uint8_t * buf, size_t size )
+{
+	scoped_lock lock( streamMutex );
+	return writeLocked( buf, size );
+}
+
+NetworkProtocol::StreamData NetworkProtocol::OutStream::getNextChunkLocked( size_t size )
+{
+	StreamData res;
+	res.id = id;
+	res.sequence = nextSequence++,
+
+	res.data.resize( size );
+	size = readLocked( res.data.data(), size );
+	res.data.resize( size );
+
+	return res;
+}
+
+
 /******************************************************************************/
 /* Header                                                                     */
 /******************************************************************************/
@@ -600,6 +865,9 @@ optional<NetworkProtocol::Header> NetworkProtocol::Header::load(const PartialObj
 		} else if (item.name == "SVR") {
 			if (auto ref = item.asRef())
 				items.emplace_back(ServiceRef { ref->digest() });
+		} else if (item.name == "STO") {
+			if (auto num = item.asInteger())
+				items.emplace_back( StreamOpen{ static_cast< uint8_t >( *num )});
 		}
 	}
 
@@ -652,6 +920,9 @@ PartialObject NetworkProtocol::Header::toObject(const PartialStorage & st) const
 
 		else if (const auto * ptr = get_if<ServiceRef>(&item))
 			ritems.emplace_back("SVR", st.ref(ptr->value));
+
+		else if (const auto * ptr = get_if< StreamOpen >( &item ))
+			ritems.emplace_back("STO", Record::Item::Integer( ptr->value ));
 	}
 
 	return PartialObject(PartialRecord(std::move(ritems)));
diff --git a/src/network/protocol.h b/src/network/protocol.h
index ba40744..d32b20b 100644
--- a/src/network/protocol.h
+++ b/src/network/protocol.h
@@ -35,8 +35,12 @@ public:
 	static constexpr char defaultVersion[] = "0.1";
 
 	class Connection;
+	class Stream;
+	class InStream;
+	class OutStream;
 
 	struct Header;
+	struct StreamData;
 
 	struct ReceivedAnnounce;
 	struct NewConnection;
@@ -102,25 +106,98 @@ public:
 	Id id() const;
 
 	const sockaddr_in6 & peerAddress() const;
+	size_t mtu() const;
 
 	optional<Header> receive(const PartialStorage &);
 	bool send(const PartialStorage &, NetworkProtocol::Header,
 			const vector<Object> &, bool secure);
+	bool send( const StreamData & chunk );
 
 	void close();
 
+	shared_ptr< InStream > openInStream( uint8_t sid );
+	shared_ptr< OutStream > openOutStream();
+
 	// temporary:
 	ChannelState & channel();
 	void trySendOutQueue();
 
 private:
-	static optional<Header> parsePacket(vector<uint8_t> & buf,
-			Channel * channel, const PartialStorage & st,
-			optional<uint64_t> & secure);
+	static variant< monostate, Header, StreamData >
+		parsePacket(vector<uint8_t> & buf,
+				Channel * channel, const PartialStorage & st,
+				optional<uint64_t> & secure);
 
 	unique_ptr<ConnectionPriv> p;
 };
 
+class NetworkProtocol::Stream
+{
+	friend class NetworkProtocol;
+	friend class NetworkProtocol::Connection;
+
+protected:
+	Stream(uint8_t id_);
+
+public:
+	void close();
+
+protected:
+	bool hasDataLocked() const;
+
+	size_t writeLocked( const uint8_t * buf, size_t size );
+	size_t readLocked( uint8_t * buf, size_t size );
+
+public:
+	const uint8_t id;
+
+protected:
+	bool closed { false };
+	vector< uint8_t > writeBuffer;
+	vector< uint8_t > readBuffer;
+	vector< uint8_t >::const_iterator readPtr;
+	mutable mutex streamMutex;
+};
+
+class NetworkProtocol::InStream : public NetworkProtocol::Stream
+{
+	friend class NetworkProtocol;
+	friend class NetworkProtocol::Connection;
+
+protected:
+	InStream(uint8_t id): Stream( id ) {}
+
+public:
+	bool isComplete() const;
+	vector< uint8_t > readAll();
+	size_t read( uint8_t * buf, size_t size );
+
+protected:
+	void writeChunk( StreamData chunk );
+	bool tryUseChunkLocked( const StreamData & chunk );
+
+private:
+	uint64_t nextSequence { 0 };
+	vector< StreamData > outOfOrderChunks;
+};
+
+class NetworkProtocol::OutStream : public NetworkProtocol::Stream
+{
+	friend class NetworkProtocol;
+	friend class NetworkProtocol::Connection;
+
+protected:
+	OutStream(uint8_t id): Stream( id ) {}
+
+public:
+	size_t write( const uint8_t * buf, size_t size );
+
+private:
+	StreamData getNextChunkLocked( size_t size );
+
+	uint64_t nextSequence { 0 };
+};
+
 struct NetworkProtocol::ReceivedAnnounce { sockaddr_in6 addr; Digest digest; };
 struct NetworkProtocol::NewConnection { Connection conn; };
 struct NetworkProtocol::ConnectionReadReady { Connection::Id id; };
@@ -141,6 +218,7 @@ struct NetworkProtocol::Header
 	struct ChannelAccept { Digest value; };
 	struct ServiceType { UUID value; };
 	struct ServiceRef { Digest value; };
+	struct StreamOpen { uint8_t value; };
 
 	using Item = variant<
 		Acknowledged,
@@ -156,7 +234,10 @@ struct NetworkProtocol::Header
 		ChannelRequest,
 		ChannelAccept,
 		ServiceType,
-		ServiceRef>;
+		ServiceRef,
+		StreamOpen>;
+
+	static constexpr size_t itemSize = 78; /* estimate for size of ref-containing headers */
 
 	Header(const vector<Item> & items): items(items) {}
 	static optional<Header> load(const PartialRef &);
@@ -169,6 +250,13 @@ struct NetworkProtocol::Header
 	vector<Item> items;
 };
 
+struct NetworkProtocol::StreamData
+{
+	uint8_t id;
+	uint8_t sequence;
+	vector< uint8_t > data;
+};
+
 template<class T>
 const T * NetworkProtocol::Header::lookupFirst() const
 {
@@ -186,28 +274,4 @@ inline bool operator!=(const NetworkProtocol::Header::Item & left,
 inline bool operator==(const NetworkProtocol::Cookie & left, const NetworkProtocol::Cookie & right)
 { return left.value == right.value; }
 
-class ReplyBuilder
-{
-public:
-	void header(NetworkProtocol::Header::Item &&);
-	void body(const Ref &);
-
-	const vector<NetworkProtocol::Header::Item> & header() const { return mheader; }
-	vector<Object> body() const;
-
-private:
-	vector<NetworkProtocol::Header::Item> mheader;
-	vector<Ref> mbody;
-};
-
-struct WaitingRef
-{
-	const Storage storage;
-	const PartialRef ref;
-	vector<Digest> missing;
-
-	optional<Ref> check();
-	optional<Ref> check(ReplyBuilder &);
-};
-
 }