boost asio를 이용한 iptuner 제작

1. 목적

boost asio를 몇 번을 봤는데, 볼 때마다 기억이 잘 나지 않는다. boost asio 예제를 하나 만들면 기억에 도움이 되지 않을까 하는 마음으로 작성한다. 예제는 내 Job과 관련이 있는 부분인 IPTV의 한 부분인 간단한 IPTuner를 만들어 본다.

IPTuner는 일반적으로 Multicast Stream을 수신하는 부분으로 간단한 예제이지만 큰 틀은 변하지 않는다.

boost buffer 부분은 좀 더 공부를 해야 할 듯 하다.

2. 소스코드

windsend나 VLC를 이용해서 server는 대체한다. 다만 VLC는 TS Stream를 VLC 맞게 변경하는 부분이 있어서 정확한 값을 얻기는 힘들다. 일단 windsend를 기반으로 동작 테스트를 진행한다. 

전에 올렸던 소스가 CTRL-C를 누르지 않으면 종료되지 않은 문제가 있어서 수정해서 다시 올린다. 전에 CTRL-C를 누르지 않으면 종료하지 않는 이유는 io_service에서 run을 호출하는 thread에서 block되기 때문에 그 thread가 main함수가 돌고 있는 thread여서 동작하지 않았다. 따라서 실제 socket에서 receive 동작을 하는 io_service 동작은 별도의 thread로 분리한다.

소스상에서 한가지 동작 중에 문제가 있는 부분은 io_service work에 관한 것이다. 실제 work는 io_service가 stop을 호출하지 전까지 io_service에 계속 block되어져야 하는데 block되지 않고 종료되는 문제가 있다. 계속 삽질 중이지만 생각되로 동작하지 않는 문제가 있다. 여기까지 오는데 엄청난 삽질을 해 됐다. 몇 가지 문제 빼 놓고는 정상적으로 동작하는 것으로 보인다.

#ifndef __IPTUNER_HPP__ #define __IPTUNER_HPP__ #include <iostream> #include <string> #include <boost/asio.hpp> #include <boost/thread.hpp> #include <boost/bind.hpp> #include <boost/array.hpp> #include <functional> #include <boost/smart_ptr.hpp> using boost::asio::ip::udp; class CIPTuner { public : #ifdef USE_BOOST_ARRAY const static int BUFFER_SIZE = 188*7; using pCallBackFuncType = std::function<void(boost::array<unsigned char, BUFFER_SIZE>)>; #else using pCallBackFuncType = std::function<void(char*, size_t)>; #endif public : CIPTuner(); CIPTuner(boost::asio::io_service &io); virtual ~CIPTuner(); virtual bool startTune(std::string addr, unsigned short port); virtual bool stopTune(void); virtual void do_receive(void); virtual void handle_receiver(const boost::system::error_code& error, size_t bytes_transferred); virtual void registerCallBackFunc(pCallBackFuncType pFunc); private : std::string addr_; unsigned short port_; boost::asio::io_service io_service_; boost::asio::ip::udp::socket socket; boost::scoped_ptr<boost::thread> receiveTrId; pCallBackFuncType pCallBack; #ifdef USE_BOOST_ARRAY boost::array<unsigned char, BUFFER_SIZE> recv_buffer; #else enum { BUFFER_SIZE = 188*7 }; char recv_buffer[BUFFER_SIZE]; #endif }; #endif enum class DebugLevel4IPTuner { DEBUG_LEVEL_NORMAL, // No Debug Message from class DEBUG_LEVEL_ERROR, // Print Error Message DEBUG_LEVEL_DEBUG, // Print Debug Related Message DEBUG_LEVEL_TRACE, // All Debug Message Trace including class member function DEBUG_LEVEL_MAX }; static DebugLevel4IPTuner gDebugLevel = DebugLevel4IPTuner::DEBUG_LEVEL_TRACE; #define DEBUG_PRINT(debugPrint, debugMsg)\ {\ if((debugPrint) <= gDebugLevel)\ {\ switch(debugPrint) \ {\ case DebugLevel4IPTuner::DEBUG_LEVEL_NORMAL:\ std::printf("[%s,%d] %s \n", __FUNCTION__, __LINE__, debugMsg);\ break;\ case DebugLevel4IPTuner::DEBUG_LEVEL_ERROR:\ std::printf("\033[0;31m Error in [%s,%d] %s \n", __FUNCTION__, __LINE__, debugMsg);\ break;\ case DebugLevel4IPTuner::DEBUG_LEVEL_DEBUG:\ std::printf("\033[1;32m Debug in [%s,%d] %s \n", __FUNCTION__, __LINE__, debugMsg);\ break;\ case DebugLevel4IPTuner::DEBUG_LEVEL_TRACE:\ std::printf("\033[1;33m Trace in [%s,%d] %s \n", __FUNCTION__, __LINE__, debugMsg);\ break;\ default :\ break;\ }\ std::printf("\033[0m");\ }\ } CIPTuner::CIPTuner() : socket(io_service_), receiveTrId() { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); pCallBack = nullptr; } CIPTuner::~CIPTuner() { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); stopTune(); } bool CIPTuner::startTune(std::string addr, unsigned short port) { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); if( (addr_.compare(addr) == 0) && (port_ == port) ) { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_ERROR, "same address and port"); return false; } if(receiveTrId) { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_ERROR, "Already Task Started"); return false; } addr_ = addr; port_ = port; boost::asio::ip::address multicast_address = boost::asio::ip::address::from_string(addr); boost::asio::ip::udp::endpoint endpoint = boost::asio::ip::udp::endpoint(multicast_address, port); socket.open(udp::v4()); socket.set_option(boost::asio::ip::udp::socket::reuse_address(true)); socket.set_option(boost::asio::ip::multicast::enable_loopback(true)); socket.bind(endpoint); // Join the multicast group. socket.set_option(boost::asio::ip::multicast::join_group(multicast_address)); //socket.non_blocking(true); //boost::scoped_ptr<boost::asio::io_service::work> work_(new boost::asio::io_service::work(io_service_)); io_service_.post(boost::bind(&CIPTuner::do_receive, this)); receiveTrId.reset(new boost::thread(boost::bind(&boost::asio::io_service::run, &io_service_))); DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); return true; } void CIPTuner::handle_receiver(const boost::system::error_code& error, size_t bytes_transferred) { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); if (!error || error == boost::asio::error::message_size) { #ifdef USE_BOOST_ARRAY if(pCallBack != nullptr) pCallBack(recv_buffer); #else if(pCallBack != nullptr) pCallBack(recv_buffer, bytes_transferred); #endif do_receive(); } } void CIPTuner::do_receive(void) { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); boost::asio::ip::udp::endpoint sender_endpoint; #ifdef USE_BOOST_ARRAY socket.async_receive_from(boost::asio::buffer(recv_buffer), sender_endpoint, boost::bind(&CIPTuner::handle_receiver, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); #else socket.async_receive_from(boost::asio::buffer(recv_buffer, BUFFER_SIZE), sender_endpoint, boost::bind(&CIPTuner::handle_receiver, this, boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); #endif } void CIPTuner::registerCallBackFunc(pCallBackFuncType pFunc) { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); pCallBack = pFunc; } bool CIPTuner::stopTune(void) { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); boost::asio::ip::address multicast_address = boost::asio::ip::address::from_string(addr_); if(!receiveTrId) { DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_ERROR, "Task is Null"); return false; } socket.set_option(boost::asio::ip::multicast::leave_group(multicast_address)); socket.close(); io_service_.stop(); receiveTrId->join(); io_service_.reset(); receiveTrId.reset(); DEBUG_PRINT(DebugLevel4IPTuner::DEBUG_LEVEL_TRACE, ""); return true; } #include "iptuner.hpp" #define PRINT_HEADER_MARK(c, l) {for(int i = 0; i < (l); i++) std::printf("%c", (c)); std::printf("\n");} #define PRINT_HEADER_INFO(s, x) std::printf("%-30s : %10d [%#10x]\n", #s, x, x); struct TSfilterInfo_t { bool transport_error_indicator; bool payload_unit_start_indicator; bool transport_priority; unsigned short pid; // 13 bit unsigned char transport_scrambling_control; // 2 bit unsigned char adaptation_field_control; // 2 bit unsigned char continuity_counter; // 4 bit unsigned char pointer_field; // 8 bit unsigned char payload_length; unsigned long long ts_packet_number; // 64 bit }; std::function<void(char*, size_t)> pCallBack = [&](char* _buffer, size_t num) { static unsigned long long cnt = 0; for(int i = 0; i < 7; i++) { char *buffer = &_buffer[188*i]; TSfilterInfo_t tsInfo; if(buffer[0] != 0x47) { std::cout << "Not Header Start " << std::endl; } tsInfo.ts_packet_number = ++cnt; tsInfo.transport_error_indicator = (buffer[1] >> 7) & 0x01; tsInfo.payload_unit_start_indicator = (buffer[1] >> 6) & 0x01; tsInfo.pid = ((buffer[1] & 0x1f) << 8) | buffer[2]; tsInfo.transport_scrambling_control = (buffer[3] >> 6) & 0x03; tsInfo.adaptation_field_control = (buffer[3] >> 4) & 0x03; tsInfo.continuity_counter = buffer[3] & 0x0f; PRINT_HEADER_MARK('-', 56) PRINT_HEADER_INFO(ts_packet_number , tsInfo.ts_packet_number) PRINT_HEADER_INFO(transport_error_indicator , tsInfo.transport_error_indicator) PRINT_HEADER_INFO(payload_unit_start_indicator , tsInfo.payload_unit_start_indicator) PRINT_HEADER_INFO(pid , tsInfo.pid) PRINT_HEADER_INFO(transport_scrambling_control , tsInfo.transport_scrambling_control) PRINT_HEADER_INFO(adaptation_field_control , tsInfo.adaptation_field_control) PRINT_HEADER_INFO(continuity_counter , tsInfo.continuity_counter) } }; int main() { CIPTuner iptuner; while(1) { char option; std::cout << "Enter Debug Menu : "; std::cin.get(option); if( (option == 'x') || (option == 'X') ) break; if( (option == 'b') || (option == 'B') ) { std::cout << "Start Tune !!" << std::endl; std::string addr = "234.1.1.1"; unsigned short port = 5000; iptuner.registerCallBackFunc(pCallBack); iptuner.startTune(addr, port); } else if( (option == 'r') || (option == 'R') ) { std::cout << "Restart Tune !!" << std::endl; std::string addr = "234.1.1.2"; unsigned short port = 5000; iptuner.registerCallBackFunc(pCallBack); iptuner.startTune(addr, port); } else if( (option == 'e') || (option == 'E') ) { std::cout << "Stop Tune !!" << std::endl; iptuner.stopTune(); } } std::cout << " ... end of main ... " << std::endl; return 0; }

3. 실행 결과

몇가지 스트림을 테스트 해 봤는데 값은 정상적으로 얻어온다.

-------------------------------------------------------- ts_packet_number : 24203 [ 0x5e8b] transport_error_indicator : 0 [ 0] payload_unit_start_indicator : 0 [ 0] pid : 8191 [ 0x1fff] transport_scrambling_control : 0 [ 0] adaptation_field_control : 1 [ 0x1] continuity_counter : 15 [ 0xf] -------------------------------------------------------- ts_packet_number : 24204 [ 0x5e8c] transport_error_indicator : 0 [ 0] payload_unit_start_indicator : 0 [ 0] pid : 8191 [ 0x1fff] transport_scrambling_control : 0 [ 0] adaptation_field_control : 1 [ 0x1] continuity_counter : 15 [ 0xf] -------------------------------------------------------- ts_packet_number : 24205 [ 0x5e8d] transport_error_indicator : 0 [ 0] payload_unit_start_indicator : 0 [ 0] pid : 8191 [ 0x1fff] transport_scrambling_control : 0 [ 0] adaptation_field_control : 1 [ 0x1] continuity_counter : 15 [ 0xf] -------------------------------------------------------- ts_packet_number : 24206 [ 0x5e8e] transport_error_indicator : 0 [ 0] payload_unit_start_indicator : 0 [ 0] pid : 8191 [ 0x1fff] transport_scrambling_control : 0 [ 0] adaptation_field_control : 1 [ 0x1] continuity_counter : 15 [ 0xf] do_receive handle_receiver