4 files changed, 259 insertions, 33 deletions

diff --git a/modules/remote/c++/remote.hpp b/modules/remote/c++/remote.hpp
index 7e846d3..83e6042 100644
--- a/modules/remote/c++/remote.hpp
+++ b/modules/remote/c++/remote.hpp
@@ -90,6 +90,14 @@ template<typename... T, typename Encoding, typename Remote>
 struct tmpl_group_to_data_server_ptr_std_tuple<tmpl_group<T...>, Encoding, Remote> {
 	using type = std::tuple<ptr<data_server<T,Encoding,Remote>>...>;
 };
+
+template<typename T, typename Encoding, typename Remote>
+struct tmpl_group_to_data_client_std_tuple;
+
+template<typename... T, typename Encoding, typename Remote>
+struct tmpl_group_to_data_client_std_tuple<tmpl_group<T...>, Encoding, Remote> {
+	using type = std::tuple<data_client<T,Encoding,Remote>...>;
+};
 }
 template<typename Remote>
 class i_rpc_server {
diff --git a/modules/remote/c++/remote_loopback.hpp b/modules/remote/c++/remote_loopback.hpp
index 0d25641..d8cd7fa 100644
--- a/modules/remote/c++/remote_loopback.hpp
+++ b/modules/remote/c++/remote_loopback.hpp
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <tuple>
 #include <variant>
 
 #include <forstio/codec/interface.hpp>
@@ -9,56 +10,169 @@
 
 namespace saw {
 
+/**
+ * Handle for data which is stored on the remote.
+ * It is the result of an rpc call and resolves to the actual data by
+ * going through the data transfer layer again.
+ */
 template<typename T, typename Encoding>
 class remote_data<T, Encoding, rmt::Loopback> {
 private:
 	id<T> id_;
+	ptr<data_server<T, Encoding, rmt::Loopback>> srv_;
 public:
-	remote_data(const id<T>& id):
-		id_{id}
+	remote_data(const id<T>& id__, ptr<data_server<T, Encoding, rmt::Loopback>> srv__):
+		id_{id__},
+		srv_{srv__}
 	{}
 
 	/**
-	 * Wait until data arrives
+	 * Id of the data on the remote.
+	 */
+	const id<T>& get_id() const {
+		return id_;
+	}
+
+	/**
+	 * Wait until data arrives.
+	 * The loopback backend answers calls inline, so the data is already
+	 * present if the call succeeded.
 	 */
-	error_or<data<T, Encoding>> wait(wait_scope& wait);
+	error_or<data<T, Encoding>> wait(wait_scope& wait){
+		(void) wait;
+		return srv_().receive(id_);
+	}
 
 	/**
 	 * Asynchronously wait for a result
 	 */
-	conveyor<data<T, Encoding>> on_receive();
+	conveyor<data<T, Encoding>> on_receive(){
+		auto eov = srv_().receive(id_);
+		if(eov.is_error()){
+			return std::move(eov.get_error());
+		}
+
+		return std::move(eov.get_value());
+	}
 };
 
 /**
- * Client RPC reference structure
+ * Client RPC reference structure.
+ *
+ * This is the overlay on top of the data transfer layer. Parameters and
+ * return values are moved by the data clients/servers. The rpc layer only
+ * ties ids to a function call on the remote.
  */
 template<typename Iface, typename Encoding>
 class rpc_client<Iface, Encoding, rmt::Loopback> {
+public:
+	using ServerT = rpc_server<Iface, Encoding, rmt::Loopback>;
+	using TypeGroup = typename schema_iface_type_group<Iface>::Type;
+	using DataClients = typename impl::tmpl_group_to_data_client_std_tuple<TypeGroup, Encoding, rmt::Loopback>::type;
+private:
+	ptr<ServerT> srv_;
+	DataClients dat_clients_;
+
+	/**
+	 * Every data client is just a handle to the data server the rpc server
+	 * is already bound to.
+	 */
+	static DataClients create_data_clients(const typename ServerT::DataServers& dat_srvs){
+		return std::apply([](const auto&... srv){
+			return DataClients{srv...};
+		}, dat_srvs);
+	}
+public:
+	rpc_client(ptr<ServerT> srv__):
+		srv_{srv__},
+		dat_clients_{create_data_clients(srv__().get_data_servers())}
+	{}
+
+	/**
+	 * Access the data client of one of the interface types.
+	 * Useful if parameters are meant to be reused in multiple calls.
+	 */
+	template<typename IdT>
+	data_client<IdT, Encoding, rmt::Loopback>& get_data_client(){
+		return std::get<data_client<IdT, Encoding, rmt::Loopback>>(dat_clients_);
+	}
+
 	/**
-	 * request the data from the remote
+	 * Request the data from the remote
 	 */
 	template<typename IdT>
-	remote_data<IdT, Encoding, rmt::Loopback> request_data(id<IdT> data);
-		
-	/** @todo
-	 * Determine type based on Name
+	remote_data<IdT, Encoding, rmt::Loopback> request_data(id<IdT> dat_id){
+		return {dat_id, std::get<ptr<data_server<IdT, Encoding, rmt::Loopback>>>(srv_().get_data_servers())};
+	}
+
+	/**
+	 * Call a function on the remote.
+	 *
+	 * The parameter is either already present on the remote, in which case
+	 * only its id is handed over, or it is sent through the data layer
+	 * first.
 	 */
-	/*
 	template<string_literal Name>
 	error_or<
-		id<
-			typename schema_member_type<Name, Iface>::type
-		>
-	> call(data_or_id<Input> inp);
-	*/
+		remote_data<typename schema_member_type<Name, Iface>::type::ResponseT, Encoding, rmt::Loopback>
+	> call(data_or_id<typename schema_member_type<Name, Iface>::type::RequestT, Encoding> input){
+		using Req = typename schema_member_type<Name, Iface>::type::RequestT;
+		using Resp = typename schema_member_type<Name, Iface>::type::ResponseT;
+
+		auto& req_client = get_data_client<Req>();
+		auto& resp_client = get_data_client<Resp>();
+
+		/**
+		 * Resolve the parameter to an id on the remote.
+		 */
+		auto eo_req_id = [&]() -> error_or<id<Req>> {
+			if(input.is_id()){
+				return input.get_id();
+			}
+
+			return req_client.send(input.get_data());
+		}();
+		if(eo_req_id.is_error()){
+			return std::move(eo_req_id.get_error());
+		}
+
+		/**
+		 * Reserve the slot the response is supposed to end up in.
+		 */
+		auto eo_resp_id = resp_client.next_free_id();
+		if(eo_resp_id.is_error()){
+			return std::move(eo_resp_id.get_error());
+		}
+		auto resp_id = eo_resp_id.get_value();
+
+		{
+			auto eov = srv_().template call<Name>(eo_req_id.get_value(), resp_id);
+			if(eov.is_error()){
+				return std::move(eov.get_error());
+			}
+		}
+
+		return request_data<Resp>(resp_id);
+	}
 };
 
+/**
+ * Server side of the rpc overlay.
+ * It translates a call on an id to a call on the local interface by
+ * retrieving the parameter from the data server and storing the response in
+ * the data server again.
+ */
 template<typename Iface, typename Encode>
-class rpc_server<Iface, Encode, rmt::Loopback> {
+class rpc_server<Iface, Encode, rmt::Loopback> final : public i_rpc_server<rmt::Loopback> {
 public:
 	using InterfaceT = interface<Iface, Encode>;
 	using TypeGroup = typename schema_iface_type_group<Iface>::Type;
 	using DataServers = typename impl::tmpl_group_to_data_server_ptr_std_tuple<TypeGroup, Encode, rmt::Loopback>::type;
+
+	/**
+	 * Identifies the interface this server provides.
+	 */
+	static constexpr std::pair<uint32_t,uint32_t> class_id{schema_hash<Iface>::apply(), schema_hash<Encode>::apply()};
 private:
 	ptr<remote<rmt::Loopback>> remote_;
 	ptr<remote_address<rmt::Loopback>> addr_;
@@ -71,7 +185,31 @@ public:
 		addr_{addr__},
 		iface_{std::move(iface__)},
 		dat_srvs_{dat_srvs__}
-	{}
+	{
+		remote_().register_rpc_server(addr_(), *this);
+	}
+
+	~rpc_server(){
+		remote_().deregister_rpc_server(addr_(), *this);
+	}
+
+	SAW_FORBID_COPY(rpc_server);
+	SAW_FORBID_MOVE(rpc_server);
+
+	/**
+	 * Return the interface id
+	 */
+	std::pair<uint32_t,uint32_t> get_class_id() const override {
+		return class_id;
+	}
+
+	/**
+	 * The data servers this rpc server moves its parameters and responses
+	 * through.
+	 */
+	const DataServers& get_data_servers() const {
+		return dat_srvs_;
+	}
 
 	template<string_literal Lit>
 	error_or<void> call(id<typename schema_member_type<Lit, Iface>::type::RequestT> id_param, id<typename schema_member_type<Lit,Iface>::type::ResponseT> id_resp){
@@ -103,4 +241,24 @@ public:
 	}
 };
 
+template<typename Iface, typename Encode>
+conveyor<rpc_client<Iface, Encode, rmt::Loopback>> remote<rmt::Loopback>::rpc_connect(const remote_address<rmt::Loopback>& addr){
+	constexpr auto class_id = rpc_server<Iface, Encode, rmt::Loopback>::class_id;
+	key_t key;
+	key.data = std::array<uint64_t,3>{addr.get_address_id().get(), class_id.first, class_id.second};
+
+	auto find = registered_rpc_servers_.find(key);
+	if(find == registered_rpc_servers_.end()){
+		return make_error<err::not_found>("Rpc server not found");
+	}
+
+	auto eo_rpc_srv = find->second().template cast_to<rpc_server<Iface, Encode, rmt::Loopback>>();
+	if(eo_rpc_srv.is_error()){
+		auto& err = eo_rpc_srv.get_error();
+		return std::move(err);
+	}
+
+	return rpc_client<Iface, Encode, rmt::Loopback>{eo_rpc_srv.get_value()};
+}
+
 }
diff --git a/modules/remote/c++/remote_loopback_base.hpp b/modules/remote/c++/remote_loopback_base.hpp
index 5224d95..f2076c6 100644
--- a/modules/remote/c++/remote_loopback_base.hpp
+++ b/modules/remote/c++/remote_loopback_base.hpp
@@ -108,6 +108,9 @@ public:
 
 	/**
 	 * Connect to a remote
+	 *
+	 * Defined in remote_loopback.hpp, since the rpc client needs to be
+	 * a complete type at that point.
 	 */
 	template<typename Iface, typename Encode>
 	conveyor<rpc_client<Iface, Encode, rmt::Loopback>> rpc_connect(const remote_address<rmt::Loopback>& addr);
@@ -175,9 +178,9 @@ public:
 		auto class_id = srv.get_class_id();
 		key_t key;
 		key.data = std::array<uint64_t,3>{addr.get_address_id().get(), class_id.first, class_id.second};
-		
+
 		auto insert = registered_data_servers_.emplace(std::make_pair(key, ptr<i_data_server<rmt::Loopback>>{srv}));
-		if(insert.second){
+		if(!insert.second){
 			return make_error<err::already_exists>("Server already bound to this address");
 		}
 
@@ -195,5 +198,34 @@ public:
 
 		return make_void();
 	}
+
+	/**
+	 * Internal function
+	 */
+	error_or<void> register_rpc_server(const remote_address<rmt::Loopback>& addr, i_rpc_server<rmt::Loopback>& srv){
+		auto class_id = srv.get_class_id();
+		key_t key;
+		key.data = std::array<uint64_t,3>{addr.get_address_id().get(), class_id.first, class_id.second};
+
+		auto insert = registered_rpc_servers_.emplace(std::make_pair(key, ptr<i_rpc_server<rmt::Loopback>>{srv}));
+		if(!insert.second){
+			return make_error<err::already_exists>("Server already bound to this address");
+		}
+
+		return make_void();
+	}
+
+	error_or<void> deregister_rpc_server(const remote_address<rmt::Loopback>& addr, i_rpc_server<rmt::Loopback>& srv){
+		auto class_id = srv.get_class_id();
+		key_t key;
+		key.data = std::array<uint64_t,3>{addr.get_address_id().get(), class_id.first, class_id.second};
+
+		auto erase = registered_rpc_servers_.erase(key);
+		if(erase == 0u){
+			return make_error<err::not_found>("Server not found");
+		}
+
+		return make_void();
+	}
 };
 }
diff --git a/modules/remote/c++/transfer_loopback.hpp b/modules/remote/c++/transfer_loopback.hpp
index e45abb3..1cd7e91 100644
--- a/modules/remote/c++/transfer_loopback.hpp
+++ b/modules/remote/c++/transfer_loopback.hpp
@@ -13,12 +13,18 @@ class remote<rmt::Loopback>;
 template<typename Schema, typename Encoding>
 class data_server<Schema, Encoding, rmt::Loopback> final : public i_data_server<rmt::Loopback> {
 private:
-	//typename 
+	//typename
 	using type = std::unordered_map<uint64_t, data<Schema, Encoding>>;
 	type values_;
 
 	ptr<remote<rmt::Loopback>> remote_;
 	remote_address<rmt::Loopback> rmt_address_;
+
+	/**
+	 * Counter for handing out ids. Ids are owned by the server, since it is
+	 * the only instance knowing about every client.
+	 */
+	uint64_t next_id_ = 0u;
 public:
 	static constexpr std::pair<uint32_t,uint32_t> class_id{schema_hash<Schema>::apply(), schema_hash<Encoding>::apply()};
 
@@ -42,7 +48,20 @@ public:
 	std::pair<uint32_t,uint32_t> get_class_id() const override {
 		return class_id;
 	}
-	
+
+	/**
+	 * Hand out the next id which is not in use yet.
+	 * The id is not reserved by this call. It is only in use as soon as
+	 * something is stored under it.
+	 */
+	id<Schema> next_free_id(){
+		while(values_.find(next_id_) != values_.end()){
+			++next_id_;
+		}
+
+		return id<Schema>{next_id_};
+	}
+
 	error_or<void> send(ref<data<Schema, Encoding>> dat, id<Schema> store_id){
 		try {
 			auto insert_res = values_.emplace(std::make_pair(store_id.get_value(), dat()));
@@ -233,40 +252,49 @@ template<typename Schema, typename Encoding>
 class data_client<Schema, Encoding, rmt::Loopback> final {
 private:
 	ptr<data_server<Schema, Encoding, rmt::Loopback>> srv_;
-
-	uint64_t next_id_;
 public:
 	data_client(ptr<data_server<Schema, Encoding, rmt::Loopback>> srv__):
-		srv_{srv__},
-		next_id_{0u}
+		srv_{srv__}
 	{}
-	
+
+	/**
+	 * Ask the remote for an id which is not in use yet.
+	 */
+	error_or<id<Schema>> next_free_id(){
+		return srv_().next_free_id();
+	}
+
 	/**
 	 * Send data to the remote.
 	 */
 	error_or<id<Schema>> send(const data<Schema, Encoding>& dat){
-		id<Schema> dat_id{next_id_};
+		id<Schema> dat_id = srv_().next_free_id();
 		auto eov = srv_().send(dat, dat_id);
 		if(eov.is_error()){
 			auto& err = eov.get_error();
 			return std::move(err);
 		}
 
-		++next_id_;
 		return dat_id;
 	}
-	
+
+	/**
+	 * Send data to the remote and store it under a specific id.
+	 */
+	error_or<void> send(const data<Schema, Encoding>& dat, id<Schema> dat_id){
+		return srv_().send(dat, dat_id);
+	}
+
 	/**
 	 * Preallocate data
 	 */
 	error_or<id<Schema>> allocate(const data<typename meta_schema<Schema>::MetaSchema, Encoding>& meta){
-		id<Schema> dat_id{next_id_};
+		id<Schema> dat_id = srv_().next_free_id();
 		auto eov = srv_().allocate(meta, dat_id);
 		if(eov.is_error()){
 			auto& err = eov.get_error();
 			return std::move(err);
 		}
-		++next_id_;
 		return dat_id;
 	}