path: root/modules/remote/c++/transfer_loopback.hpp

#pragma once

#include <forstio/reduce_templates.hpp>

#include "remote_loopback.hpp"
#include "transfer.hpp"

namespace saw {

template<>
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 
	using type = std::unordered_map<uint64_t, data<Schema, Encoding>>;
	type values_;

	ptr<remote<rmt::Loopback>> remote_;
	remote_address<rmt::Loopback> rmt_address_;
public:
	static constexpr std::pair<uint32_t,uint32_t> class_id{schema_hash<Schema>::apply(), schema_hash<Encoding>::apply()};

	data_server(ptr<remote<rmt::Loopback>> remote__, const remote_address<rmt::Loopback>& addr):
		remote_{remote__},
		rmt_address_{addr}
	{
		remote_().register_data_server(addr, *this);
	}

	~data_server(){
		remote_().deregister_data_server(rmt_address_, *this);
	}

	SAW_FORBID_COPY(data_server);
	SAW_FORBID_MOVE(data_server);

	/**
	 * Return the schema id
	 */
	std::pair<uint32_t,uint32_t> get_class_id() const override {
		return class_id;
	}
	
	error_or<void> send(const data<Schema, Encoding>& dat, id<Schema> store_id){
		try {
			auto insert_res = values_.emplace(std::make_pair(store_id.get_value(), dat));
			if(!insert_res.second){
				return make_error<err::already_exists>();
			}
		}catch(std::exception& ){
			return make_error<err::out_of_memory>();
		}
		return void_t{};
	}

	error_or<void> allocate(data<typename meta_schema<Schema>::MetaSchema, Encoding> meta, id<Schema> store_id){
		try{
			auto insert_res = values_.emplace(std::make_pair(store_id.get_value(), data<typename meta_schema<Schema>::MetaSchema, Encoding>{std::move(meta)}));
			if(!insert_res.second){
				return make_error<err::already_exists>();
			}
		}catch(const std::exception&){
			return make_error<err::out_of_memory>();
		}

		return make_void();
	}

	error_or<data<Schema, Encoding>> receive(id<Schema> store_id){
		auto find_res = values_.find(store_id.get_value());
		if(find_res == values_.end()){
			return make_error<err::not_found>();
		}

		auto& dat = find_res->second;
		// here
		return dat;// boy
	}

	error_or<void> erase(id<Schema> store_id){
		auto erase_res = values_.erase(store_id.get_value());
		if(erase_res == 0u){
			return make_error<err::not_found>();
		}

		return make_void();
	}

	error_or<ptr<data<Schema, Encoding>>> find(id<Schema> store_id){
		auto find_res = values_.find(store_id.get_value());
		if(find_res == values_.end()){
			return make_error<err::not_found>();
		}

		auto& dat = find_res->second;
		return ptr<data<Schema,Encoding>>{dat};
	}
};

template<typename... Schema, typename Encoding>
class data_server<tmpl_group<Schema...>, Encoding, rmt::Loopback> final {
private:
	ptr<remote<rmt::Loopback>> remote_;
	remote_address<rmt::Loopback> rmt_address_;
	std::tuple<data_server<Schema,Encoding,rmt::Loopback>...> values_;

	struct inner_register_helper {
		template<uint64_t i, typename Func>
		static error_or<uint64_t> reg_i(ptr<remote<rmt::Loopback>> rmt, ref<remote_address<rmt::Loopback>> addr, ref<std::tuple<data_server<Schema,Encoding,rmt::Loopback>...>> vals){

			{
				auto eov = rmt().register_data_server(addr,std::get<i>(vals));
				if(eov.is_error()){
					return i;
				}
			}

			if constexpr ( (i+1u) < sizeof...(Schema) ){
				return reg_i<i+1u>(rmt, addr, vals);
			}
			// This only happens if we reached the last element
			return i+1u;
		}

		static error_or<uint64_t> reg(ptr<remote<rmt::Loopback>> rmt, ref<remote_address<rmt::Loopback>> addr, 
				ref<std::tuple<data_server<Schema,Encoding,rmt::Loopback>...>> vals){
			if ( 0u < sizeof...(Schema) ){
				return reg_i<0u>(rmt,addr,vals);
			}
			return 1u;
		}

		static error_or<uint64_t> dereg(ptr<remote<rmt::Loopback>> rmt, ref<remote_address<rmt::Loopback>> addr, ref<std::tuple<data_server<Schema,Encoding,rmt::Loopback>...>> vals){
			return 1u;
		}
	};
public:
	data_server(ptr<remote<rmt::Loopback>> remote__, const remote_address<rmt::Loopback>& addr, std::tuple<data_server<Schema,Encoding>...>&& values__):
		remote_{remote__},
		rmt_address_{addr},
		values_{std::move(values__)}
	{
	}

	~data_server(){
		remote_().deregister_data_server(rmt_address_,*this);
	}

	SAW_FORBID_COPY(data_server);
	SAW_FORBID_MOVE(data_server);

	/**
	 * Get data from client
	 */
	template<typename Sch>
	error_or<void> send(const data<Sch, Encoding>& dat, id<Sch> store_id){
		auto& vals = std::get<std::unordered_map<uint64_t, data<Sch,Encoding>>>(values_);

		try {
			auto insert_res = vals.emplace(std::make_pair(store_id.get_value(), std::move(dat)));
			if(!insert_res.second){
				return make_error<err::already_exists>();
			}
		}catch(std::exception& ){
			return make_error<err::out_of_memory>();
		}
		return void_t{};
	}

	/**
	 *
	 */
	template<typename Sch>
	error_or<void> allocate(data<typename meta_schema<Sch>::MetaSchema, Encoding> meta, id<Sch> store_id){
		auto& vals = std::get<std::unordered_map<uint64_t, data<Sch,Encoding>>>(values_);
		try {
			auto insert_res = vals.emplace(std::make_pair(store_id.get_value(), data<typename meta_schema<Sch>::MetaSchema, Encoding>{std::move(meta)}));
			if(!insert_res.second){
				return make_error<err::already_exists>();
			}
		}catch(std::exception& ){
			return make_error<err::out_of_memory>();
		}
		return make_void();
	}

	template<typename Sch>
	error_or<data<Sch, Encoding>> receive(id<Sch> store_id){
		auto& vals = std::get<std::unordered_map<uint64_t, data<Sch,Encoding>>>(values_);

		auto find_res = vals.find(store_id.get_value());
		if(find_res == vals.end()){
			return make_error<err::not_found>();
		}

		auto& dat = find_res->second;
		return dat;
	}

	template<typename Sch>
	error_or<void> erase(id<Sch> store_id){
		auto& vals = std::get<std::unordered_map<uint64_t, data<Sch,Encoding>>>(values_);
		auto erase_op = vals.erase(store_id.get_value());
		if(erase_op == 0u){
			return make_error<err::not_found>();
		}
		return make_void();
	}

	template<typename Sch>
	error_or<data<Sch, Encoding>*> find(id<Sch> store_id){
		auto& vals = std::get<std::unordered_map<uint64_t, data<Sch,Encoding>>>(values_);
		auto find_res = vals.find(store_id.get_value());
		if(find_res == vals.end()){
			return make_error<err::not_found>();
		}

		auto& dat = find_res->second;
		return &dat;
	}
};

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}
	{}
	
	/**
	 * Send data to the remote.
	 */
	error_or<id<Schema>> send(const data<Schema, Encoding>& dat){
		id<Schema> dat_id{next_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;
	}
	
	/**
	 * Preallocate data
	 */
	error_or<id<Schema>> allocate(const data<typename meta_schema<Schema>::MetaSchema, Encoding>& meta){
		id<Schema> dat_id{next_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;
	}

	/**
	 * Receive data
	 */
	conveyor<data<Schema, Encoding>> receive(id<Schema> dat_id){
		auto eov = srv_().receive(dat_id);
		if(eov.is_error()){
			auto& err = eov.get_error();
			return std::move(err);
		}

		auto& val = eov.get_value();
		return std::move(val);
	}

	/**
	 * Erase data
	 */
	error_or<void> erase(id<Schema> dat_id){
		return srv_().erase(dat_id);
	}

	/**
	 * An exception for the Loopback backend. Here we can safely use find from
	 * the client side.
	 */
	error_or<ptr<data<Schema, Encoding>>> find(id<Schema> dat_id){
		auto eov = srv_().find(dat_id);
		if(eov.is_error()){
			auto& err = eov.get_error();
			return std::move(err);
		}

		auto val = eov.get_value();
		return val;
	}
};

/**
 * Client for transporting data to remote and receiving data back
 */
template<typename... Schema, typename Encoding>
class data_client<tmpl_group<Schema...>, Encoding, rmt::Loopback> {
private:
	/**
	 * Corresponding server for this client
	 */
	ptr<data_server<tmpl_group<Schema...>, Encoding, rmt::Loopback>> srv_;

	/**
	 * The next id for identifying issues on the remote side.
	 */
	uint64_t next_id_;
public:
	/**
	 * Main constructor
	 */
	data_client(ptr<data_server<tmpl_group<Schema...>, Encoding, rmt::Loopback>> srv__):
		srv_{srv__},
		next_id_{0u}
	{}

	SAW_FORBID_COPY(data_client);
	SAW_DEFAULT_MOVE(data_client);

	/**
	 * Send data to the remote.
	 */
	template<typename Sch>
	error_or<id<Sch>> send(const data<Sch, Encoding>& dat){
		id<Sch> dat_id{next_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;
	}
	
	/**
	 * Preallocate data
	 */
	template<typename Sch>
	error_or<id<Sch>> allocate(const data<typename meta_schema<Sch>::MetaSchema, Encoding>& meta){
		id<Sch> dat_id{next_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;
	}

	/**
	 * Receive data
	 */
	template<typename Sch>
	conveyor<data<Sch, Encoding>> receive(id<Sch> dat_id){
		auto eov = srv_().receive(dat_id);
		if(eov.is_error()){
			auto& err = eov.get_error();
			return std::move(err);
		}

		auto& val = eov.get_value();
		return std::move(val);
	}

	/**
	 * Erase data
	 */
	template<typename Sch>
	error_or<void> erase(id<Sch> dat_id){
		return srv_().erase(dat_id);
	}

	/**
	 * An exception for the Loopback backend. Here we can safely use find from
	 * the client side.
	 */
	template<typename Sch>
	error_or<data<Sch, Encoding>*> find(id<Sch> dat_id){
		auto eov = srv_().find(dat_id);
		if(eov.is_error()){
			auto& err = eov.get_error();
			return std::move(err);
		}

		auto val = eov.get_value();
		return val;
	}
};

}