boost_multicast_2.cpp

#include "stdafx.h"

#define BOOST_ASIO_ENABLE_CANCELIO
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/thread.hpp>
#include <boost/shared_ptr.hpp>

using boost::asio::ip::udp;

// --------------------------------------------------------------------------------
// Multicast Server
// --------------------------------------------------------------------------------
class CMulticastServer : public boost::enable_shared_from_this<CMulticastServer>
{
private:
	CMulticastServer(
		LPCSTR multicast_ip,
		CONST USHORT open_port,
		LPCSTR data);
public:
	typedef boost::shared_ptr<CMulticastServer> SP;
	static SP CreateSP(
		LPCSTR multicast_ip,
		CONST USHORT open_port,
		LPCSTR data)
	{
		return SP(new CMulticastServer(multicast_ip, open_port, data));
	}
	~CMulticastServer();

public:
	DWORD Start();
	VOID Stop();
protected:
	VOID handle_receive_from(CONST boost::system::error_code& error, size_t receiveBytes);

private:
	enum
	{
		MAX_LENGTH = 32,
		MAX_TTL = 2
	};

	boost::asio::io_service			m_io_service;
	boost::thread_group				m_thread_group;

	string							m_multicast_ip;
	udp::socket						m_socket;
	udp::endpoint					m_sender_endpoint;
	CHAR							m_recvData[MAX_LENGTH + 1];
	CHAR							m_sendData[MAX_LENGTH + 1];

	LPCSTR							m_data;
	boost::system::error_code		m_ec;
};

CMulticastServer::CMulticastServer(
	LPCSTR multicast_ip,
	CONST USHORT open_port,
	LPCSTR data
	)
	:m_socket(m_io_service, udp::endpoint(udp::v4(), open_port)),
	m_multicast_ip(multicast_ip)
{
	ZeroMemory(m_recvData, sizeof(m_recvData));
	strcpy_s(m_sendData, data);
}

CMulticastServer::~CMulticastServer(void)
{
}

DWORD CMulticastServer::Start()
{
	// STEP 1 : Multicast Group Join
	boost::asio::ip::address multicast_group_address(boost::asio::ip::address::from_string(m_multicast_ip));
	m_socket.set_option(boost::asio::ip::udp::socket::reuse_address(true));
	m_socket.set_option(boost::asio::ip::multicast::join_group(multicast_group_address), m_ec); // join
	if (m_ec){ return m_ec.value(); }
	m_socket.set_option(boost::asio::ip::multicast::hops(MAX_TTL), m_ec); // set hops
	if (m_ec){ return m_ec.value(); }

	// STEP 2 : Wait & Receive
	m_socket.async_receive_from(
		boost::asio::buffer(m_recvData, MAX_LENGTH),
		m_sender_endpoint,
		boost::bind(&CMulticastServer::handle_receive_from, shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)
		);

	// STEP 3 : IoService에 Thread 할당
	m_thread_group.create_thread(boost::bind(&boost::asio::io_service::run, &m_io_service));

	return NO_ERROR;
}

VOID CMulticastServer::Stop()
{
	// 소켓을 닫는다.
	m_socket.close();

	// IoService를 중단한다.
	m_io_service.stop();

	// 작업 Thread가 모두 종료되기를 기다린다.
	m_thread_group.join_all();
}

VOID CMulticastServer::handle_receive_from(CONST boost::system::error_code& error, size_t receiveBytes)
{
	if (!error && receiveBytes > 0)
	{
		if (_stricmp(m_recvData, "multicast_sample") == 0)
		{
			size_t sendDataLength = m_socket.send_to(
				boost::asio::buffer(m_sendData, strlen(m_sendData)),
				m_sender_endpoint
				);
		}

		ZeroMemory(m_recvData, sizeof(m_recvData));
		m_socket.async_receive_from(
			boost::asio::buffer(m_recvData, MAX_LENGTH),
			m_sender_endpoint,
			boost::bind(&CMulticastServer::handle_receive_from, shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)
			);
	}
}

// --------------------------------------------------------------------------------
// Multicast Client
// --------------------------------------------------------------------------------
class CMulticastClient
{
public:
	CMulticastClient(
		std::string dest_ip /* multicast group ip */,
		std::string dest_port /* multicast group port */);
	~CMulticastClient();

public:
	DWORD Start(unsigned int maxttl = MAX_TTL, unsigned int timeout = TIME_OUT);
	VOID Stop();
	vector<std::string>& GetRecvDataList();

protected:
	VOID handle_receive_from(CONST boost::system::error_code& error);
	VOID handle_timeout(CONST boost::system::error_code& error);

private:
	enum
	{
		MAX_LENGTH = 32,
		MAX_TTL = 2,
		TIME_OUT = 3
	};

	boost::asio::io_service m_io_service;
	boost::thread_group	m_thread_group;
	udp::socket	m_socket;
	std::string	m_dest_ip;
	std::string	m_dest_port;

	vector<std::string> m_recvDatas;
	CHAR m_recvData[MAX_LENGTH + 1];
	CHAR m_sendData[MAX_LENGTH + 1];

	boost::system::error_code m_ec;
};


CMulticastClient::CMulticastClient(std::string dest_ip, std::string dest_port)
	: m_socket(m_io_service, udp::endpoint(udp::v4(), 0)),
	m_dest_ip(dest_ip), m_dest_port(dest_port)
{
	ZeroMemory(m_recvData, sizeof(m_recvData));
	ZeroMemory(m_sendData, sizeof(m_sendData));
	strcpy_s(m_sendData, "multicast_sample");
}


CMulticastClient::~CMulticastClient()
{
	printf("destroy\n");
}

VOID CMulticastClient::Stop()
{
	// 소켓을 닫는다.
	m_socket.close();
	printf("m_socket.close()\n");

	// IoService를 중단한다.
	m_io_service.stop();
	printf("m_io_service.stop()\n");

	// 작업 Thread가 모두 종료되기를 기다린다.
	m_thread_group.join_all();
	printf("m_thread_group.join_all()\n");
}

DWORD CMulticastClient::Start(unsigned int maxttl, unsigned int timeout)
{
	udp::resolver resolver(m_io_service);
	udp::resolver::query query(udp::v4(), m_dest_ip, m_dest_port);
	udp::resolver::iterator iterator = resolver.resolve(query, m_ec);
	if (m_ec) { return m_ec.value(); }

	// --------------------------------------------------------------------------------
	// STEP 1 : send 데이터 전송 
	// --------------------------------------------------------------------------------
	// Set TTL (Hop) 
	m_socket.set_option(boost::asio::ip::multicast::hops(maxttl));

	// 생성자에서 얻어온 multicast group ip로 udp패킷을 전송
	size_t nSendResultLength = m_socket.send_to(boost::asio::buffer(m_sendData, strlen(m_sendData)), *iterator, 0, m_ec);
	if (m_ec) { return m_ec.value(); }

	// --------------------------------------------------------------------------------
	// STEP 2 : receive udp (non-blocking) 
	// --------------------------------------------------------------------------------
	if (nSendResultLength > 0) {
		// 서버로부터 ip주소가 담긴 stream을 m_recvData로 가져온다.
		udp::endpoint sendEndpoint;
		m_socket.async_receive_from(boost::asio::buffer(m_recvData), sendEndpoint, boost::bind(&CMulticastClient::handle_receive_from, this, boost::asio::placeholders::error));
		m_thread_group.create_thread(boost::bind(&boost::asio::io_service::run, &m_io_service));
	}
	return NO_ERROR;
}

VOID CMulticastClient::handle_timeout(CONST boost::system::error_code& error)
{
	printf("handle_timeout : %d\n", error);
	if (!error)
	{
		// 타임아웃 기다리는 동안 서버로부터 메시지를 받지 못했다면, 커넥션을 닫는다.
		if (m_socket.is_open()){			
			m_socket.close();
		}
	}
}

VOID CMulticastClient::handle_receive_from(CONST boost::system::error_code& error)
{
	printf("handle_receive_from : %d\n", error);
	if (!error)
	{
		printf("receive data : %s\n", m_recvData);
		m_recvDatas.push_back(m_recvData);

		ZeroMemory(m_recvData, sizeof(m_recvData));
		udp::endpoint sendEndpoint;
		m_socket.async_receive_from(boost::asio::buffer(m_recvData), sendEndpoint, boost::bind(&CMulticastClient::handle_receive_from, this, boost::asio::placeholders::error));
	}
}

vector<std::string>& CMulticastClient::GetRecvDataList()
{
	return m_recvDatas;
}

// --------------------------------------------------------------------------------
// Main
// --------------------------------------------------------------------------------
int main()
{
	{
		DWORD dwResult = NO_ERROR;

		//CMulticastServer::SP server = CMulticastServer::CreateSP("224.4.35.183", 13701, "hello world");
		//dwResult = server->Start();

		//Sleep(1000);

		CMulticastClient client1("224.4.35.183", "13701");
		client1.Start();

		boost::thread t([&](){
			Sleep(1000 * 2);
			client1.Stop();
		});
		t.join();
		//server->Stop();
	}

	printf("exit\n");
	getchar();

	return 0;
}