Vehicle-cpp/EgoJC/EgoRemote/socket_remote.cpp

#include "pch.h"
#include <thread>
#include "../protocol/win/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_add, &SocketRemote::OnAdd));
	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(20916);
	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::OnAdd(int8_t* Data, int16_t Size)
{
	remote::SCAdd Rep;
	Rep.ParseFromArray(Data, Size);
	if (Rep.ret() == true)
	{
		_uid = Rep.uid();
	}
	_notify->OnAdd(_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(20916);
			sin.sin_addr.s_addr = inet_addr(_ip.c_str());
			if (connect(sockfd, (const sockaddr*)&sin, sizeof(sin)) == SOCKET_ERROR)
			{
 
				closesocket(sockfd);
 
				sockfd = INVALID_SOCKET;
				continue;

			}
			_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);
	}

}