<commit_before>/*
 * qianqians
 * 2020-1-10
 * gatemanager.cpp
 */
#include <enetacceptservice.h>
#include <modulemng_handle.h>

#include "gatemanager.h"
#include "hub_service.h"

namespace hub {
	
gateproxy::gateproxy(std::shared_ptr<abelkhan::Ichannel> ch, std::shared_ptr<hub_service> hub, std::string gate_name) {
	_ch = ch;
	_hub_call_gate_caller = std::make_shared<abelkhan::hub_call_gate_caller>(ch, service::_modulemng);
	_hub = hub;
	_gate_name = gate_name;
}

void gateproxy::reg_hub() {
	_hub_call_gate_caller->reg_hub(_hub->hub_name, _hub->hub_type)->callBack([this]() {
		spdlog::trace("hub reg_hub to gate:{0} sucessed", _gate_name);
		sig_reg_hub_sucessed.emit();
	}, [this]() {
		spdlog::trace("hub reg_hub to gate:{0} faild", _gate_name);
	})->timeout(5 * 1000, [this]() {
		spdlog::trace("hub reg_hub to gate:{0} timeout", _gate_name);
	});
}
void gateproxy::disconnect_client(std::string& cuuid) {
	_hub_call_gate_caller->disconnect_client(cuuid);
}

void gateproxy::forward_hub_call_client(const std::string& cuuid, const std::vector<uint8_t>& _data_bin) {
	_hub_call_gate_caller->forward_hub_call_client(cuuid, _data_bin);
}

void gateproxy::forward_hub_call_group_client(const std::vector<std::string>& cuuids, const std::vector<uint8_t>& rpc_argv) {
	_hub_call_gate_caller->forward_hub_call_group_client(cuuids, rpc_argv);
}

void gateproxy::forward_hub_call_global_client(const std::vector<uint8_t>& rpc_argv) {
	_hub_call_gate_caller->forward_hub_call_global_client(rpc_argv);
}

directproxy::directproxy(std::string cuuid_, std::shared_ptr<abelkhan::Ichannel> direct_ch) {
	_hub_call_client_caller = std::make_shared<abelkhan::hub_call_client_caller>(direct_ch, service::_modulemng);
	
	_cuuid = cuuid_;
	_direct_ch = direct_ch;
}

void directproxy::call_client(const std::vector<uint8_t>& rpc_argv) {
	_hub_call_client_caller->call_client(rpc_argv);
}

gatemanager::gatemanager(std::shared_ptr<service::enetacceptservice> conn_, std::shared_ptr<hub_service> hub_) {
	_conn = conn_;
	_hub = hub_;

	_hub->sig_svr_be_closed.connect([this](std::string svr_type, std::string svr_name) {
		if (svr_type != "gate") {
			return;
		}

		for (auto it = clients.begin(); it != clients.end(); it++) {
			if (it->second->_gate_name == svr_name) {
				clients.erase(it);
				break;
			}
		}
		
		auto it = gates.find(svr_name);
		if (it != gates.end()) {
			ch_gates.erase(it->second->_ch);
			gates.erase(it);

			_hub->sig_gate_closed.emit(svr_name);
		}
	});

	_data_size = 8 * 1024;
	_data = (unsigned char*)malloc(_data_size);
}

void gatemanager::connect_gate(std::string gate_name, std::string ip, uint16_t port) {
	spdlog::trace("connect_gate ip:{0} port:{1}", ip, port);
	_conn->connect(ip, port, [this, ip, port, gate_name](std::shared_ptr<abelkhan::Ichannel> ch) {
		spdlog::trace("gate ip:{0}  port:{1} reg_hub", ip, port);
		auto _gate_proxy = std::make_shared<gateproxy>(ch, _hub, gate_name);
		_gate_proxy->sig_reg_hub_sucessed.connect([this, gate_name, ch, _gate_proxy]() {
			gates[gate_name] = _gate_proxy;
			ch_gates[ch] = _gate_proxy;
		});
		_gate_proxy->reg_hub();
	});
}

void gatemanager::client_connect(std::string client_uuid, std::shared_ptr<abelkhan::Ichannel> gate_ch) {
	if (ch_gates.find(gate_ch) == ch_gates.end()) {
		spdlog::trace("unreg gate channel!");
		return;
	}

	spdlog::trace("reg client:{0}", client_uuid);

	clients[client_uuid] = ch_gates[gate_ch];
}

void gatemanager::client_disconnect(std::string client_uuid) {
	if (clients.find(client_uuid) == clients.end()) {
		return;
	}

	clients.erase(client_uuid);
	_hub->sig_client_disconnect.emit(client_uuid);
}

void gatemanager::client_direct_connect(std::string client_uuid, std::shared_ptr<abelkhan::Ichannel> direct_ch) {
	auto it = direct_clients.find(client_uuid);
	if (it != direct_clients.end()) {
		ch_direct_clients.erase(it->second->_direct_ch);
		direct_clients.erase(it);
		return;
	}

	spdlog::trace("reg direct client:{0}", client_uuid);

	auto _directproxy = std::make_shared<directproxy>(client_uuid, direct_ch);
	direct_clients.insert(std::make_pair(client_uuid, _directproxy));

	ch_direct_clients.insert(std::make_pair(direct_ch, _directproxy));
}

void gatemanager::client_direct_disconnect(std::shared_ptr<abelkhan::Ichannel> direct_ch) {
	auto it = ch_direct_clients.find(direct_ch);
	if (it == ch_direct_clients.end()) {
		return;
	}

	direct_clients.erase(it->second->_cuuid);
	ch_direct_clients.erase(it);

	_hub->sig_direct_client_disconnect.emit(it->second->_cuuid);

	spdlog::trace("client_direct_disconnect direct_clients.size:{0}, ch_direct_clients.size:{1}!", direct_clients.size(), ch_direct_clients.size());
}

std::shared_ptr<directproxy> gatemanager::get_direct_client(std::shared_ptr<abelkhan::Ichannel> direct_ch) {
	auto it = ch_direct_clients.find(direct_ch);
	if (it == ch_direct_clients.end()) {
		return nullptr;
	}

	return it->second;
}

void gatemanager::disconnect_client(std::string uuid) {
	auto it_gate = clients.find(uuid);
	if (it_gate != clients.end()) {
		it_gate->second->disconnect_client(uuid);
		clients.erase(it_gate);
	}

	auto it_direct = direct_clients.find(uuid);
	if (it_direct != direct_clients.end()) {
		it_direct->second->_direct_ch->disconnect();
		ch_direct_clients.erase(it_direct->second->_direct_ch);
		direct_clients.erase(it_direct);
	}
}

void gatemanager::heartbeat_client(int64_t ticktime) {
	std::vector<std::shared_ptr<directproxy> > remove_client;<commit_msg>bug fixed:gatemanager<commit_after>/*
 * qianqians
 * 2020-1-10
 * gatemanager.cpp
 */
#include <enetacceptservice.h>
#include <modulemng_handle.h>

#include "gatemanager.h"
#include "hub_service.h"

namespace hub {
	
gateproxy::gateproxy(std::shared_ptr<abelkhan::Ichannel> ch, std::shared_ptr<hub_service> hub, std::string gate_name) {
	_ch = ch;
	_hub_call_gate_caller = std::make_shared<abelkhan::hub_call_gate_caller>(ch, service::_modulemng);
	_hub = hub;
	_gate_name = gate_name;
}

void gateproxy::reg_hub() {
	_hub_call_gate_caller->reg_hub(_hub->hub_name, _hub->hub_type)->callBack([this]() {
		spdlog::trace("hub reg_hub to gate:{0} sucessed", _gate_name);
		sig_reg_hub_sucessed.emit();
	}, [this]() {
		spdlog::trace("hub reg_hub to gate:{0} faild", _gate_name);
	})->timeout(5 * 1000, [this]() {
		spdlog::trace("hub reg_hub to gate:{0} timeout", _gate_name);
	});
}
void gateproxy::disconnect_client(std::string& cuuid) {
	_hub_call_gate_caller->disconnect_client(cuuid);
}

void gateproxy::forward_hub_call_client(const std::string& cuuid, const std::vector<uint8_t>& _data_bin) {
	_hub_call_gate_caller->forward_hub_call_client(cuuid, _data_bin);
}

void gateproxy::forward_hub_call_group_client(const std::vector<std::string>& cuuids, const std::vector<uint8_t>& rpc_argv) {
	_hub_call_gate_caller->forward_hub_call_group_client(cuuids, rpc_argv);
}

void gateproxy::forward_hub_call_global_client(const std::vector<uint8_t>& rpc_argv) {
	_hub_call_gate_caller->forward_hub_call_global_client(rpc_argv);
}

directproxy::directproxy(std::string cuuid_, std::shared_ptr<abelkhan::Ichannel> direct_ch) {
	_hub_call_client_caller = std::make_shared<abelkhan::hub_call_client_caller>(direct_ch, service::_modulemng);
	
	_cuuid = cuuid_;
	_direct_ch = direct_ch;
}

void directproxy::call_client(const std::vector<uint8_t>& rpc_argv) {
	_hub_call_client_caller->call_client(rpc_argv);
}

gatemanager::gatemanager(std::shared_ptr<service::enetacceptservice> conn_, std::shared_ptr<hub_service> hub_) {
	_conn = conn_;
	_hub = hub_;

	_hub->sig_svr_be_closed.connect([this](std::string svr_type, std::string svr_name) {
		if (svr_type != "gate") {
			return;
		}

		for (auto it = clients.begin(); it != clients.end(); it++) {
			if (it->second->_gate_name == svr_name) {
				clients.erase(it);
				break;
			}
		}
		
		auto it = gates.find(svr_name);
		if (it != gates.end()) {
			ch_gates.erase(it->second->_ch);
			gates.erase(it);

			_hub->sig_gate_closed.emit(svr_name);
		}
	});

	_data_size = 8 * 1024;
	_data = (unsigned char*)malloc(_data_size);
}

void gatemanager::connect_gate(std::string gate_name, std::string ip, uint16_t port) {
	spdlog::trace("connect_gate ip:{0} port:{1}", ip, port);
	_conn->connect(ip, port, [this, ip, port, gate_name](std::shared_ptr<abelkhan::Ichannel> ch) {
		spdlog::trace("gate ip:{0}  port:{1} reg_hub", ip, port);
		auto _gate_proxy = std::make_shared<gateproxy>(ch, _hub, gate_name);
		_gate_proxy->sig_reg_hub_sucessed.connect([this, gate_name, ch, _gate_proxy]() {
			gates[gate_name] = _gate_proxy;
			ch_gates[ch] = _gate_proxy;
		});
		_gate_proxy->reg_hub();
	});
}

void gatemanager::client_connect(std::string client_uuid, std::shared_ptr<abelkhan::Ichannel> gate_ch) {
	if (ch_gates.find(gate_ch) == ch_gates.end()) {
		spdlog::trace("unreg gate channel!");
		return;
	}

	spdlog::trace("reg client:{0}", client_uuid);

	clients[client_uuid] = ch_gates[gate_ch];
}

void gatemanager::client_disconnect(std::string client_uuid) {
	if (clients.find(client_uuid) == clients.end()) {
		return;
	}

	clients.erase(client_uuid);
	_hub->sig_client_disconnect.emit(client_uuid);
}

void gatemanager::client_direct_connect(std::string client_uuid, std::shared_ptr<abelkhan::Ichannel> direct_ch) {
	auto it = direct_clients.find(client_uuid);
	if (it != direct_clients.end()) {
		ch_direct_clients.erase(it->second->_direct_ch);
		direct_clients.erase(it);
		return;
	}

	spdlog::trace("reg direct client:{0}", client_uuid);

	auto _directproxy = std::make_shared<directproxy>(client_uuid, direct_ch);
	direct_clients.insert(std::make_pair(client_uuid, _directproxy));

	ch_direct_clients.insert(std::make_pair(direct_ch, _directproxy));
}

void gatemanager::client_direct_disconnect(std::shared_ptr<abelkhan::Ichannel> direct_ch) {
	auto it = ch_direct_clients.find(direct_ch);
	if (it == ch_direct_clients.end()) {
		return;
	}

	auto cuuid = it->second->_cuuid;

	direct_clients.erase(cuuid);
	ch_direct_clients.erase(it);

	_hub->sig_direct_client_disconnect.emit(cuuid);

	spdlog::trace("client_direct_disconnect direct_clients.size:{0}, ch_direct_clients.size:{1}!", direct_clients.size(), ch_direct_clients.size());
}

std::shared_ptr<directproxy> gatemanager::get_direct_client(std::shared_ptr<abelkhan::Ichannel> direct_ch) {
	auto it = ch_direct_clients.find(direct_ch);
	if (it == ch_direct_clients.end()) {
		return nullptr;
	}

	return it->second;
}

void gatemanager::disconnect_client(std::string uuid) {
	auto it_gate = clients.find(uuid);
	if (it_gate != clients.end()) {
		it_gate->second->disconnect_client(uuid);
		clients.erase(it_gate);
	}

	auto it_direct = direct_clients.find(uuid);
	if (it_direct != direct_clients.end()) {
		it_direct->second->_direct_ch->disconnect();
		ch_direct_clients.erase(it_direct->second->_direct_ch);
		direct_clients.erase(it_direct);
	}
}

void gatemanager::heartbeat_client(int64_t ticktime) {
	std::vector<std::shared_ptr<directproxy> > remove_client;
{'fraction_non_alphanumeric': 0.15605241455145238, 'fraction_numerical': 0.006597635847154769, 'mean_word_length': 7.072485207100592, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 119, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2968494', 'n_tokens_mistral': 4399, 'n_tokens_neox': 4248, 'n_words': 581}