Cockpit-cpp/EgoRemote/socket_remote.cpp

#include "pch.h"
#include <thread>
#include "api/video/video_frame.h"
#include "remote.pb.h"
#include "socket_remote.h"
#include "../common/iobuffer.h"


SocketRemote::SocketRemote(IRemoteNotify* n) :_notify(n)
{
	_connected = false;
}
bool SocketRemote::Start(const char* ip)
{
	_ip = ip;

	FnMap.insert(std::make_pair(remote::sc_robot, &SocketRemote::OnRobotRep));
	FnMap.insert(std::make_pair(remote::sc_NotifyRep, &SocketRemote::OnNotifyRep));
	FnMap.insert(std::make_pair(remote::sc_NotifyAdd, &SocketRemote::OnNotifyAdd));
	FnMap.insert(std::make_pair(remote::sc_NotifyDel, &SocketRemote::OnNotifyDel));

	//FnMap.insert(std::make_pair(remote::sc_NotifyState, &SocketRemote::OnNotifyState));
	FnMap.insert(std::make_pair(remote::sc_sigin, &SocketRemote::OnSigin));
	FnMap.insert(std::make_pair(remote::sc_NotifyReq, &SocketRemote::OnNotifyReq));
	FnMap.insert(std::make_pair(remote::sc_NotifyLeave, &SocketRemote::OnNotifyLeave));
	FnMap.insert(std::make_pair(remote::sc_NotifyOffer, &SocketRemote::OnNotifyOffer));
	FnMap.insert(std::make_pair(remote::sc_NotifyAnswer, &SocketRemote::OnNotifyAnswer));
	FnMap.insert(std::make_pair(remote::sc_NotifyCandidate, &SocketRemote::OnNotifyCandidate));
	WSAData data;
	WSAStartup(MAKEWORD(2, 2), &data);
	sockfd = socket(AF_INET, SOCK_STREAM, 0);
	sockaddr_in sin;
	sin.sin_family = AF_INET;
	sin.sin_port = htons(20410);
	sin.sin_addr.s_addr = inet_addr(ip);
	if (connect(sockfd, (const sockaddr*)&sin, sizeof(sin)) == SOCKET_ERROR)
	{
		DWORD error = WSAGetLastError();
		closesocket(sockfd);
		sockfd = INVALID_SOCKET;
		_connected = false;
	}
	else
	{
		_connected = true;
	}
	_notify->OnConnected(_connected);
	_thread = std::thread(&SocketRemote::Run, this);
	return true;

}
void SocketRemote::OnRobotRep(int8_t* Data, int16_t Size)
{
	remote::SCRobot Rep;
	Rep.ParseFromArray(Data, Size);
	for (int32_t i = 0; i < Rep.robot_size(); i++)
	{
		auto& node = Rep.robot(i);
		_notify->OnRobot(node);
	}
}
void SocketRemote::OnSigin(int8_t* Data, int16_t Size)
{
	remote::SCSigin Rep;
	Rep.ParseFromArray(Data, Size);
	if (Rep.ret() == true)
	{
		_uid = Rep.uid();
	}
	_notify->OnLogin(_uid, Rep.ret());
}
void SocketRemote::OnNotifyAnswer(int8_t* Data, int16_t Size)
{

	remote::Answer Rep;
	Rep.ParseFromArray(Data, Size);
	_notify->OnVideoAnswer(Rep.index(), Rep.type().c_str(), Rep.sdp().c_str());
}
void SocketRemote::OnNotifyCandidate(int8_t* Data, int16_t Size)
{

	remote::Candidate Rep;
	Rep.ParseFromArray(Data, Size);
	_notify->OnVideoCandidate(Rep.index(), Rep.candidate().c_str(), Rep.sdpmlineindex(), Rep.sdpmid().c_str());

}
 
void SocketRemote::OnNotifyOffer(int8_t* Data, int16_t Size)
{

	remote::Offer Rep;
	Rep.ParseFromArray(Data, Size);
	_notify->OnVideoOffer(Rep.index(), Rep.type().c_str(), Rep.sdp().c_str());
}
void SocketRemote::OnNotifyRep(int8_t* Data, int16_t Size)
{

	remote::CSRep Rep;
	Rep.ParseFromArray(Data, Size);
	auto ok = Rep.desc() == remote::VideoDesc::OK;
	_notify->OnVideoRep(ok, Rep.index(), Rep.peer());
}
void SocketRemote::OnNotifyAdd(int8_t* Data, int16_t Size)
{
	remote::SCAddRobot Rep;
	Rep.ParseFromArray(Data, Size);
	_notify->OnRobot(Rep.robot());
}
void SocketRemote::OnNotifyDel(int8_t* Data, int16_t Size)
{
	remote::SCDelRobot Rep;
	Rep.ParseFromArray(Data, Size);
	_notify->OnNotifyDel(Rep.peer(), (EgoType)(Rep.egotype()));
}
void SocketRemote::OnNotifyReq(int8_t* Data, int16_t Size)
{


	remote::CSReq Rep;
	Rep.ParseFromArray(Data, Size);
	_notify->OnVideoReq(Rep.index(), Rep.peer());


}
void SocketRemote::OnNotifyLeave(int8_t* Data, int16_t Size)
{

	remote::Leave Req;
	Req.ParseFromArray(Data, Size);
	int32_t peer = Req.peer();
	EgoType type = static_cast<EgoType>(Req.egotype());
	_notify->OnVideoLeave(peer, type);
}
void SocketRemote::Run()
{
	_run = true;
	int32_t Offset = 0;
	CIOBuffer Buffer;
	while (_run)
	{
		if (_connected)
		{
			auto ret = recv(sockfd, (char*)&Buffer.Buffer[Offset], CIOBuffer::IO_BUFFER_SIZE - Offset, 0);
			if (ret <= 0)
			{

 
				DWORD error = WSAGetLastError();
				closesocket(sockfd);
 
				_connected = false;
				_notify->OnConnected(_connected);

			}
			else
			{
				Offset += ret;
				if (Offset >= MessageHead::Size())
				{
					bool bNeedMove = false;
					MessageHead head;
					int8_t* ptr = Buffer.Buffer;
					while (true)
					{
						if (MessageHead::Size() <= Offset)
						{
							head.Deserialize(ptr);
							int32_t length = MessageHead::Size() + head.Length;
							if (Offset >= length)
							{
								int8_t* Data = ptr + MessageHead::Size();

								NetProcess(head.Command, Data, head.Length);

								ptr += length;
								Offset -= length;
							}
							else
							{
								bNeedMove = Offset > 0;
								if (bNeedMove)
								{
									std::cout << "need Move " << Offset << std::endl;
								}
								break;
							}
						}
						else
						{
							break;
						}

					}
					if (bNeedMove)
					{
						memmove(Buffer.Buffer, ptr, Offset);
					}
				}
			}
		}
		else
		{
			std::this_thread::sleep_for(std::chrono::seconds(1));
			sockfd = socket(AF_INET, SOCK_STREAM, 0);
			sockaddr_in sin;
			sin.sin_family = AF_INET;
			sin.sin_port = htons(20410);
			sin.sin_addr.s_addr = inet_addr(_ip.c_str());
			if (connect(sockfd, (const sockaddr*)&sin, sizeof(sin)) == SOCKET_ERROR)
			{
 
				closesocket(sockfd);
				_connected = false;
				sockfd = INVALID_SOCKET;
				continue;

			}
			else
			{
				_connected = true;
			}
			_notify->OnConnected(_connected);
		}
	}


}
void SocketRemote::NetProcess(int16_t cmd, int8_t* Data, int16_t Size)
{


	auto it = FnMap.find(cmd);
	if (it != FnMap.end())
	{
		(this->*it->second)(Data, Size);
	}

}
void SocketRemote::Stop()
{
	_run = false;
 
	closesocket(sockfd);
 
	sockfd = INVALID_SOCKET;
	_thread.join();
}
void SocketRemote::WriteSigin(EgoType type, std::string& account, std::string& pass, std::string& name)
{
	remote::CSSigin Req;
	Req.set_account(account);
	Req.set_pass(pass);
	Req.set_name(name);
	Req.set_type(type);
	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_sigin;
	Head.Length = Req.ByteSizeLong();
	Head.Serialize(pBuffer.Buffer);
	auto ptr = pBuffer.Buffer + MessageHead::Size();
	Req.SerializeToArray(ptr, Head.Length);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);

}
void SocketRemote::Write(CIOBuffer* pBuffer)
{
	if (_connected)
	{

		int32_t ret = ::send(sockfd, (const char*)pBuffer->Buffer, pBuffer->Length, 0);
		if (ret <= 0)
		{
 
			closesocket(sockfd);
 
			_connected = false;
		}
	}
}
void SocketRemote::WriteVideoReq(int32_t peer, int32_t index)
{
	//_peer = peer;
	remote::CSReq Req;
	Req.set_peer(peer);
	Req.set_index(index);
	Req.set_egotype(EgoType::User);
	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_req;
	Head.Length = Req.ByteSizeLong();
	Head.Serialize(pBuffer.Buffer);
	auto ptr = pBuffer.Buffer + MessageHead::Size();
	Req.SerializeToArray(ptr, Head.Length);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);
}
void SocketRemote::WriteUserList(void)
{
	remote::CSRobot Req;
	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_robot;
	Head.Length = Req.ByteSizeLong();
	Head.Serialize(pBuffer.Buffer);
	auto ptr = pBuffer.Buffer + MessageHead::Size();
	Req.SerializeToArray(ptr, Head.Length);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);
}
void SocketRemote::WriteVideoRep(int32_t peer, remote::VideoDesc desc, int32_t index)
{
	//	_peer = peer;
	remote::CSRep Req;
	Req.set_peer(peer);
	Req.set_desc(desc);
	Req.set_index(index);

	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_rep;
	Head.Length = Req.ByteSizeLong();
	Head.Serialize(pBuffer.Buffer);
	auto ptr = pBuffer.Buffer + MessageHead::Size();
	Req.SerializeToArray(ptr, Head.Length);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);
}
void SocketRemote::WriteKeepAlive()
{
	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_keepAlive;
	Head.Length = 0;
	Head.Serialize(pBuffer.Buffer);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);
}
void SocketRemote::WriteOffer(int32_t peer, int32_t index, const char* type, const char* sdp)
{
	remote::Offer Req;
	Req.set_peer(peer);
	Req.set_sdp(sdp);
	Req.set_type(type);
	Req.set_index(index);

	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_offer;
	Head.Length = Req.ByteSizeLong();
	Head.Serialize(pBuffer.Buffer);
	auto ptr = pBuffer.Buffer + MessageHead::Size();
	Req.SerializeToArray(ptr, Head.Length);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);
}
void SocketRemote::WriteAnswer(int32_t peer, int32_t index, const char* type, const char* sdp)
{
	remote::Answer Req;
	Req.set_peer(peer);
	Req.set_sdp(sdp);
	Req.set_type(type);
	Req.set_index(index);

	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_answer;
	Head.Length = Req.ByteSizeLong();
	Head.Serialize(pBuffer.Buffer);
	auto ptr = pBuffer.Buffer + MessageHead::Size();
	Req.SerializeToArray(ptr, Head.Length);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);
}
void SocketRemote::WriteCandidate(int32_t peer, int32_t index, const char* candidate, int32_t sdp_mline_index, const char* sdp_mid)
{
	remote::Candidate Req;
	Req.set_peer(peer);
	Req.set_candidate(candidate);
	Req.set_index(index);
	Req.set_sdpmid(sdp_mid);
	Req.set_sdpmlineindex(sdp_mline_index);


	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_candidate;
	Head.Length = Req.ByteSizeLong();
	Head.Serialize(pBuffer.Buffer);
	auto ptr = pBuffer.Buffer + MessageHead::Size();
	Req.SerializeToArray(ptr, Head.Length);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);
}
void SocketRemote::WriteVideoLeave(int32_t peer)
{
	remote::Leave Req;
	Req.set_peer(peer);
	 
	MessageHead Head;
	CIOBuffer pBuffer;
	Head.Command = remote::cs_leave;
	Head.Length = Req.ByteSizeLong();
	Head.Serialize(pBuffer.Buffer);
	auto ptr = pBuffer.Buffer + MessageHead::Size();
	Req.SerializeToArray(ptr, Head.Length);
	pBuffer.Length = MessageHead::Size() + Head.Length;
	Write(&pBuffer);
}