Test message service for sending arbitrary data

4 files changed, 102 insertions, 0 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index fff6242..70d55ba 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -26,6 +26,7 @@ add_library(erebos
 if (NOT CMAKE_SYSTEM_NAME STREQUAL "Android")
 	add_executable(erebos-bin
 		main.cpp
+		test/service.cpp
 	)
 
 	set_target_properties(erebos-bin
diff --git a/src/main.cpp b/src/main.cpp
index c92d630..68ebbd5 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -8,6 +8,7 @@
 #include <erebos/sync.h>
 
 #include "storage.h"
+#include "test/service.h"
 
 #include <arpa/inet.h>
 #include <netinet/in.h>
@@ -316,6 +317,23 @@ void startServer(const vector<string> &)
 
 	config.service<SyncService>();
 
+	config.service< TestService >()
+		.onMessage([]( const Stored< Object > & msg ) {
+			auto bytes = msg->encode();
+			auto space = std::find( bytes.begin(), bytes.end(), ' ' );
+			auto type = string( bytes.begin(), space );
+			auto size = std::stoi( string( space + 1, std::find( space + 1, bytes.end(), '\n' ) ) );
+
+			ostringstream ss;
+			ss << "test-message-received"
+				<< " " << type
+				<< " " << size
+				<< " " << string( msg.ref().digest() )
+				;
+			printLine(ss.str());
+		})
+		;
+
 	server.emplace(*testHead, move(config));
 
 	server->peerList().onUpdate([](size_t idx, const Peer * peer) {
@@ -367,6 +385,16 @@ void peerAdd(const vector<string> & args)
 		throw invalid_argument("usage: peer-add <node> [<port>]");
 }
 
+void testMessageSend( const vector< string > & args)
+{
+	if (auto ref = st.ref(Digest(args.at(1)))) {
+		TestService::send(getPeer( args.at(0) ).peer, *ref );
+		printLine("test-message-send done");
+	} else {
+		printLine("test-message-send fail");
+	}
+}
+
 void sharedStateGet(const vector<string> &)
 {
 	ostringstream ss;
@@ -613,6 +641,7 @@ vector<Command> commands = {
 	{ "start-server", startServer },
 	{ "stop-server", stopServer },
 	{ "peer-add", peerAdd },
+	{ "test-message-send", testMessageSend },
 	{ "shared-state-get", sharedStateGet },
 	{ "shared-state-wait", sharedStateWait },
 	{ "watch-local-identity", watchLocalIdentity },
diff --git a/src/test/service.cpp b/src/test/service.cpp
new file mode 100644
index 0000000..32fca52
--- /dev/null
+++ b/src/test/service.cpp
@@ -0,0 +1,37 @@
+#include "service.h"
+
+#include <erebos/network.h>
+
+using namespace erebos;
+
+static const UUID myUUID("cb46b92c-9203-4694-8370-8742d8ac9dc8");
+
+TestService::TestService( Config && c, const Server & ):
+	config( move(c) )
+{
+}
+
+TestService::~TestService() = default;
+
+UUID TestService::uuid() const
+{
+	return myUUID;
+}
+
+void TestService::handle( Context & ctx )
+{
+	auto msg = Stored< Object >::load( ctx.ref() );
+	for (const auto & w : config.watchers)
+		w( msg );
+}
+
+void TestService::send( const Peer & peer, const Ref & msg )
+{
+	peer.send( myUUID, msg );
+}
+
+TestService::Config & TestService::Config::onMessage( MessageWatcher w )
+{
+	watchers.push_back(w);
+	return *this;
+}
diff --git a/src/test/service.h b/src/test/service.h
new file mode 100644
index 0000000..c693ce4
--- /dev/null
+++ b/src/test/service.h
@@ -0,0 +1,35 @@
+#pragma once
+
+#include <erebos/service.h>
+
+namespace erebos
+{
+
+class TestService : public Service
+{
+public:
+	using MessageWatcher = std::function<void( const Stored< Object > & )>;
+
+	class Config
+	{
+	public:
+		Config & onMessage( MessageWatcher );
+
+	private:
+		friend class TestService;
+		vector< MessageWatcher > watchers;
+	};
+
+	TestService( Config &&, const Server & );
+	virtual ~TestService();
+
+	UUID uuid() const override;
+	void handle( Context & ) override;
+
+	static void send( const Peer &, const Ref & );
+
+private:
+	const Config config;
+};
+
+}