#include "cloudserver.h"

CloudServer::CloudServer(DispatcherClient * c, const PIByteArray & sname): server(c) {
	setName(sname.toHex());
	server_uuid = sname;
	connects << CONNECTL(c, dataReadedServer, ([this](uint id, PIByteArray & ba) {
							 // keep the lookup and the send under the lock: the
		                     // sub-client may be deleted by the drainer concurrently
							 mutex_clients.lock();
							 last_ping.reset();
							 DispatcherClient * cl = index_clients.value(id, nullptr);
							 if (cl) {
								 // a healthy forward is fast: a slow one means
			                     // the peer is not draining, remember it and
			                     // drop further data instead of stalling
								 PITimeMeasurer tm;
								 cl->sendData(ba);
								 if (tm.elapsed_s() > 2.0) {
									 cl->markWriteStalled();
									 piCoutObj << "write stalled to client" << id;
								 }
							 }
							 mutex_clients.unlock();
						 }));
	connects << CONNECTL(c, pingReceived, [this]() {
		mutex_clients.lock();
		last_ping.reset();
		mutex_clients.unlock();
	});
	{
		PIMutexLocker locker(mutex_clients);
		last_ping.reset();
	}
}


CloudServer::~CloudServer() {}


PIByteArray CloudServer::serverUUID() const {
	return server_uuid;
}


void CloudServer::addClient(DispatcherClient * c) {
	mutex_clients.lock();
	last_ping.reset();
	clients << c;
	uint cid = c->clientId();
	index_clients.insert(cid, c);
	mutex_clients.unlock();
	c->sendConnected(1);
	server->sendConnected(cid);
	CONNECTL(c, dataReaded, ([this, cid](PIByteArray & ba) {
				 //		piCoutObj << c->clientId() << "dataReaded";
				 PITimeMeasurer tm;
				 server->sendDataToClient(ba, cid);
				 if (tm.elapsed_s() > 2.0) {
					 server->markWriteStalled();
					 piCoutObj << "write stalled to server" << cid;
				 }
			 }));
}


void CloudServer::removeClient(DispatcherClient * c) {
	mutex_clients.lock();
	last_ping.reset();
	clients.removeOne(c);
	index_clients.remove(c->clientId());
	mutex_clients.unlock();
	server->sendDisconnected(c->clientId());
}


PIVector<DispatcherClient *> CloudServer::getClients() {
	PIMutexLocker locker(mutex_clients);
	PIVector<DispatcherClient *> cl = clients;
	return cl;
}


double CloudServer::lastPing() {
	PIMutexLocker locker(mutex_clients);
	return last_ping.elapsed_s();
}


void CloudServer::close() {
	server->close();
}


void CloudServer::stop() {
	// copy the list under the lock: addClient()/removeClient() modify it
	// concurrently from other threads
	PIVector<DispatcherClient *> cl;
	{
		PIMutexLocker locker(mutex_clients);
		cl = clients;
	}
	for (auto c: cl)
		c->close();
	server->stop();
}


void CloudServer::printStatus() {
	PIMutexLocker locker(mutex_clients);
	piCout << "    " << "Clients for" << server->address() << server_uuid.toHex() << ":";
	for (auto c: clients) {
		piCout << "        " << c->address() << c->clientId();
	}
}