Functioning and tested RPC bind and call

net/include/ftl/net/dispatcher.hpp

@@ -33,9 +33,11 @@ namespace internal {
 }
 
 namespace net {
+class Socket;
 
 class Dispatcher {
 	public:
+	Dispatcher(Socket *s) : sock_(s) {}
 	
 	void dispatch(const std::string &msg);
 	
@@ -115,8 +117,12 @@ class Dispatcher {
 
     //! \brief This is the type of notification messages.
     using notification_t = std::tuple<int8_t, std::string, msgpack::object>;
+    
+    using response_t =
+        std::tuple<uint32_t, uint32_t, msgpack::object, msgpack::object>;
 	
 	private:
+	ftl::net::Socket *sock_;
 	std::unordered_map<std::string, adaptor_type> funcs_;
 	
 	static void enforce_arg_count(std::string const &func, std::size_t found,

net/include/ftl/net/protocol.hpp

@@ -2,6 +2,7 @@
 #define _FTL_NET_PROTOCOL_HPP_
 
 #define FTL_PROTOCOL_RPC		0x0100
+#define FTL_PROTOCOL_RPCRETURN	0x0101
 #define FTL_PROTOCOL_P2P		0x1000
 
 #endif // _FTL_NET_PROTOCOL_HPP_

net/include/ftl/net/socket.hpp

@@ -21,6 +21,11 @@
 namespace ftl {
 namespace net {
 
+struct Header {
+	uint32_t size;
+	uint32_t service;
+};
+
 class Socket {
 	public:
 	Socket(const char *uri);
@@ -49,13 +54,13 @@ class Socket {
 		     typename ftl::internal::func_kind_info<F>::args_kind());
 	}
 	
-	template <typename... ARGS>
-	msgpack::object_handle call(const std::string &name, ARGS... args) {
+	template <typename T, typename... ARGS>
+	T call(const std::string &name, ARGS... args) {
 		bool hasreturned = false;
-		msgpack::object_handle result;
-		async_call(name, [result,hasreturned](msgpack::object_handle r) {
+		T result;
+		async_call(name, [&result,&hasreturned](msgpack::object &r) {
 			hasreturned = true;
-			result = r;
+			result = r.as<T>();
 		}, std::forward<ARGS>(args)...);
 		
 		// Loop the network
@@ -71,7 +76,7 @@ class Socket {
 	template <typename... ARGS>
 	void async_call(
 			const std::string &name,
-			std::function<void(msgpack::object_handle)> cb,
+			std::function<void(msgpack::object&)> cb,
 			ARGS... args) {
 		auto args_obj = std::make_tuple(args...);
 		auto rpcid = rpcid__++;
@@ -107,7 +112,7 @@ class Socket {
 	char *m_buffer;
 	sockdatahandler_t m_handler;
 	bool m_valid;
-	std::map<int, std::function<void(msgpack::object_handle)>> callbacks_;
+	std::map<int, std::function<void(msgpack::object&)>> callbacks_;
 	ftl::net::Dispatcher disp_;
 	
 	static int rpcid__;

net/src/dispatcher.cpp

 #include <ftl/net/dispatcher.hpp>
+#include <ftl/net/socket.hpp>
+#include <iostream>
+
+/*static std::string hexStr(const std::string &s)
+{
+	const char *data = s.data();
+	int len = s.size();
+    std::stringstream ss;
+    ss << std::hex;
+    for(int i=0;i<len;++i)
+        ss << std::setw(2) << std::setfill('0') << (int)data[i];
+    return ss.str();
+}*/
 
 void ftl::net::Dispatcher::dispatch(const std::string &msg) {
+	//std::cout << "Received dispatch : " << hexStr(msg) << std::endl;
     auto unpacked = msgpack::unpack(msg.data(), msg.size());
     dispatch(unpacked.get());
 }
@@ -8,10 +22,11 @@ void ftl::net::Dispatcher::dispatch(const std::string &msg) {
 void ftl::net::Dispatcher::dispatch(const msgpack::object &msg) {
     switch (msg.via.array.size) {
     case 3:
-        dispatch_notification(msg);
+        dispatch_notification(msg); break;
     case 4:
-        dispatch_call(msg);
+        dispatch_call(msg); break;
     default:
+    	std::cout << "Unrecognised msgpack : " << msg.via.array.size << std::endl;
         return;
     }
 }
@@ -24,7 +39,7 @@ void ftl::net::Dispatcher::dispatch_call(const msgpack::object &msg) {
     // auto &&type = std::get<0>(the_call);
     // assert(type == 0);
 
-   // auto &&id = std::get<1>(the_call);
+    auto &&id = std::get<1>(the_call);
     auto &&name = std::get<2>(the_call);
     auto &&args = std::get<3>(the_call);
 
@@ -32,8 +47,12 @@ void ftl::net::Dispatcher::dispatch_call(const msgpack::object &msg) {
 
     if (it_func != end(funcs_)) {
         try {
-            auto result = (it_func->second)(args);
-            // TODO SEND RESULTS
+            auto result = (it_func->second)(args)->get();
+			auto res_obj = std::make_tuple(1,id,msgpack::object(),result);
+			std::stringstream buf;
+			msgpack::pack(buf, res_obj);
+			
+			sock_->send(FTL_PROTOCOL_RPCRETURN, buf.str());
         } catch (...) {
 			throw;
 		}
@@ -50,6 +69,8 @@ void ftl::net::Dispatcher::dispatch_notification(msgpack::object const &msg) {
 
     auto &&name = std::get<1>(the_call);
     auto &&args = std::get<2>(the_call);
+    
+    std::cout << "RPC NOTIFY" << name << std::endl;
 
     auto it_func = funcs_.find(name);
 

net/src/socket.cpp

@@ -25,6 +25,8 @@ using namespace ftl;
 using ftl::net::Socket;
 using namespace std;
 
+int Socket::rpcid__ = 0;
+
 static int tcpConnect(URI &uri) {
 	int rc;
 	sockaddr_in destAddr;
@@ -105,13 +107,13 @@ static int wsConnect(URI &uri) {
 	return 1;
 }
 
-Socket::Socket(int s) : m_sock(s), m_pos(0) {
+Socket::Socket(int s) : m_sock(s), m_pos(0), disp_(this) {
 	// TODO Get the remote address.
 	m_valid = true;
 	m_buffer = new char[BUFFER_SIZE];
 }
 
-Socket::Socket(const char *pUri) : m_uri(pUri), m_pos(0) {
+Socket::Socket(const char *pUri) : m_uri(pUri), m_pos(0), disp_(this) {
 	// Allocate buffer
 	m_buffer = new char[BUFFER_SIZE];
 	
@@ -169,6 +171,7 @@ bool Socket::data() {
 	while (m_pos < len+4) {
 		if (len > MAX_MESSAGE) {
 			close();
+			std::cout << "Length is too big" << std::endl;
 			return false; // Prevent DoS
 		}
 
@@ -185,8 +188,10 @@ bool Socket::data() {
 			}
 		} else if (rc == EWOULDBLOCK || rc == 0) {
 			// Data not yet available
+			std::cout << "No data to read" << std::endl;
 			return false;
 		} else {
+			std::cout << "Socket error" << std::endl;
 			// Close socket due to error
 			close();
 			return false;
@@ -194,18 +199,55 @@ bool Socket::data() {
 	}
 
 	// All data available
-	if (m_handler) {
+	//if (m_handler) {
 		uint32_t service = ((uint32_t*)m_buffer)[1];
 		auto d = std::string(m_buffer+8, len-4);
 		//std::cerr << "DATA : " << service << " -> " << d << std::endl;
-		m_handler(service, d); 
-	}
+		
+		if (service == FTL_PROTOCOL_RPC) {
+			dispatch(d);
+		} else if (service == FTL_PROTOCOL_RPCRETURN) {
+			auto unpacked = msgpack::unpack(d.data(), d.size());
+			Dispatcher::response_t the_result;
+			unpacked.get().convert(the_result);
+
+			// TODO: proper validation of protocol (and responding to it)
+			// auto &&type = std::get<0>(the_call);
+			// assert(type == 0);
+
+			// auto &&id = std::get<1>(the_call);
+			auto &&id = std::get<1>(the_result);
+			//auto &&err = std::get<2>(the_result);
+			auto &&res = std::get<3>(the_result);
+
+			if (callbacks_.count(id) > 0) callbacks_[id](res);
+			else std::cout << "NO CALLBACK FOUND FOR RPC RESULT" << std::endl;
+		} else {
+			if (m_handler) m_handler(service, d);
+		} 
+	//}
 
 	m_pos = 0;
 
 	return true;
 }
 
+int Socket::send(uint32_t service, const std::string &data) {
+	ftl::net::Header h;
+	h.size = data.size()+4;
+	h.service = service;
+	
+	iovec vec[2];
+	vec[0].iov_base = &h;
+	vec[0].iov_len = sizeof(h);
+	vec[1].iov_base = const_cast<char*>(data.data());
+	vec[1].iov_len = data.size();
+	
+	::writev(m_sock, &vec[0], 2);
+	
+	return 0;
+}
+
 Socket::~Socket() {
 	close();
 	

net/test/CMakeLists.txt

-add_executable(tests EXCLUDE_FROM_ALL
+include(CTest)
+enable_testing()
+
+add_executable(rpc_test EXCLUDE_FROM_ALL
+	./tests.cpp
+	./rpc.cpp
+	../src/dispatcher.cpp
+	../src/socket.cpp
+)
+target_include_directories(rpc_test PUBLIC ${PROJECT_SOURCE_DIR}/include)
+target_link_libraries(rpc_test uriparser)
+
+add_executable(socket_test EXCLUDE_FROM_ALL
 	./tests.cpp
 	./net_raw.cpp
 	../src/net.cpp
@@ -7,9 +19,11 @@ add_executable(tests EXCLUDE_FROM_ALL
 	./ice.cpp
 	../src/ice.cpp
 	./uri.cpp
-	./rpc.cpp
 	../src/dispatcher.cpp
 )
-target_include_directories(tests PUBLIC ${PROJECT_SOURCE_DIR}/include)
-target_link_libraries(tests uriparser)
+target_include_directories(socket_test PUBLIC ${PROJECT_SOURCE_DIR}/include)
+target_link_libraries(socket_test uriparser)
+
+add_custom_target(tests)
+add_dependencies(tests rpc_test socket_test)
 

net/test/rpc.cpp

 #include "catch.hpp"
 #include <ftl/net/socket.hpp>
 #include <iostream>
+#include <memory>
+#include <map>
+
+/*struct FakeHeader {
+	uint32_t size;
+	uint32_t service;
+};*/
+
+static std::map<int, std::string> fakedata;
+
+/*static std::string hexStr(const std::string &s)
+{
+	const char *data = s.data();
+	int len = s.size();
+    std::stringstream ss;
+    ss << std::hex;
+    for(int i=0;i<len;++i)
+        ss << std::setw(2) << std::setfill('0') << (int)data[i];
+    return ss.str();
+}*/
+
+void fake_send(int sd, uint32_t  service, const std::string &data) {
+	//std::cout << "HEX SEND: " << hexStr(data) << std::endl;
+	char buf[8+data.size()];
+	ftl::net::Header *h = (ftl::net::Header*)&buf;
+	h->size = data.size()+4;
+	h->service = service;
+	std::memcpy(&buf[8],data.data(),data.size());
+	fakedata[sd] = std::string(&buf[0], 8+data.size());
+	
+	//std::cout << "HEX SEND2: " << hexStr(fakedata[sd]) << std::endl;
+}
+
+extern ssize_t recv(int sd, void *buf, size_t n, int f) {
+	if (fakedata.count(sd) == 0) {
+		std::cout << "Unrecognised socket" << std::endl;
+		return 0;
+	}
+	
+	int l = fakedata[sd].size();
+	
+	std::memcpy(buf, fakedata[sd].c_str(), l);
+	fakedata.erase(sd);
+	return l;
+}
+
+extern ssize_t writev(int sd, const struct iovec *v, int cnt) {
+	size_t len = v[0].iov_len+v[1].iov_len;
+	char buf[len];
+	std::memcpy(&buf[0],v[0].iov_base,v[0].iov_len);
+	std::memcpy(&buf[v[0].iov_len],v[1].iov_base,v[1].iov_len);
+	fakedata[sd] = std::string(&buf[0], len);
+	return 0;
+}
+
+static std::function<void()> waithandler;
+
+bool ftl::net::wait() {
+	if (waithandler) waithandler();
+	//waithandler = nullptr;
+	return true;
+}
 
 TEST_CASE("Socket::bind()", "[rpc]") {
-	SECTION("no argument bind") {
+	SECTION("no argument bind fake dispatch") {
 		auto s = new ftl::net::Socket(0);
 		bool called = false;
 		
@@ -20,7 +82,7 @@ TEST_CASE("Socket::bind()", "[rpc]") {
 		REQUIRE( called );
 	}
 	
-	SECTION("one argument bind") {
+	SECTION("one argument bind fake dispatch") {
 		auto s = new ftl::net::Socket(0);
 		bool called = false;
 		
@@ -38,7 +100,7 @@ TEST_CASE("Socket::bind()", "[rpc]") {
 		REQUIRE( called );
 	}
 	
-	SECTION("two argument bind") {
+	SECTION("two argument bind fake dispatch") {
 		auto s = new ftl::net::Socket(0);
 		bool called = false;
 		
@@ -56,5 +118,108 @@ TEST_CASE("Socket::bind()", "[rpc]") {
 		s->dispatch(buf.str());
 		REQUIRE( called );
 	}
+	
+	SECTION("no argument bind fake data") {
+		auto s = new ftl::net::Socket(0);
+		bool called = false;
+		
+		s->bind("test1", [&]() {
+			called = true;
+		});
+		
+		auto args_obj = std::make_tuple();
+		auto call_obj = std::make_tuple(0,0,"test1",args_obj);
+		std::stringstream buf;
+		msgpack::pack(buf, call_obj);
+		
+		fake_send(0, FTL_PROTOCOL_RPC, buf.str());
+		REQUIRE( s->data() );
+		REQUIRE( called );
+	}
+	
+	SECTION("non-void bind fake data") {
+		auto s = new ftl::net::Socket(0);
+		bool called = false;
+		
+		s->bind("test1", [&]() -> int {
+			called = true;
+			return 55;
+		});
+		
+		auto args_obj = std::make_tuple();
+		auto call_obj = std::make_tuple(0,0,"test1",args_obj);
+		std::stringstream buf;
+		msgpack::pack(buf, call_obj);
+		
+		fake_send(0, FTL_PROTOCOL_RPC, buf.str());
+		REQUIRE( s->data() );
+		REQUIRE( called );
+		
+		// TODO Require that a writev occurred with result value
+	}
+}
+
+TEST_CASE("Socket::call()", "[rpc]") {
+	SECTION("no argument call") {
+		auto s = new ftl::net::Socket(0);
+		
+		waithandler = [&]() {
+			// Read fakedata sent
+			// TODO Validate data
+			
+			// Do a fake send
+			auto res_obj = std::make_tuple(1,0,msgpack::object(),66);
+			std::stringstream buf;
+			msgpack::pack(buf, res_obj);
+		
+			fake_send(0, FTL_PROTOCOL_RPCRETURN, buf.str());
+			s->data();
+		};
+		
+		int res = s->call<int>("test1");
+		
+		REQUIRE( res == 66 );
+	}
+	
+	SECTION("one argument call") {
+		auto s = new ftl::net::Socket(0);
+		
+		waithandler = [&]() {
+			// Read fakedata sent
+			// TODO Validate data
+			
+			// Do a fake send
+			auto res_obj = std::make_tuple(1,1,msgpack::object(),43);
+			std::stringstream buf;
+			msgpack::pack(buf, res_obj);
+		
+			fake_send(0, FTL_PROTOCOL_RPCRETURN, buf.str());
+			s->data();
+		};
+		
+		int res = s->call<int>("test1", 78);
+		
+		REQUIRE( res == 43 );
+	}
+	
+	waithandler = nullptr;
+}
+
+TEST_CASE("Socket::call+bind loop", "[rpc]") {
+	auto s = new ftl::net::Socket(0);
+
+	// Just loop the send back to the recv
+	waithandler = [&]() {
+		s->data();
+	};
+	
+	s->bind("test1", [](int a) -> int {
+		std::cout << "Bind test1 called" << std::endl;
+		return a*2;
+	});
+	
+	int res = s->call<int>("test1", 5);
+	
+	REQUIRE( res == 10 );
 }