今回は、マルチキャスト受信プログラム(UDPMulticastReceiver for Windows)を作ります。
UDPMulticastSenderの作成で記述したように、マルチキャストグループへの「参加」、「離脱」
機能をUDPServerに組み込みます。
UDPMultiRecveiverに必要な設定
マルチキャストグループへの参加・離脱
【使用する構造体】 struct group_req { uint32_t gr_interface; // インターフェース番号 struct sockaddr_storage gr_group; // グループアドレス };
(＊)通常gr_interface=0を指定するとすべてのインターフェースが対象となりますが
使用するインターフェース番号を指定しないと動作しないことがあります。
(＊)IPv4とIPv6で異なる構造体を使用する方法(古い方法)がありますが
ここではでは同じ構造体で制御を行います(新しい方法)。
これは、RFC 3678:Socket Interface Extensions for Multicast Source Filtersに
記述されています。
マルチキャストグループへの参加
IPv4
group_req mreq
setsockopt(socket, IPPROTO_IP, MCAST_JOIN_GROUP, (char *)&mreq, sizeof(mreq));
IPv6
group_req mreq
setsockopt(socket, IPPROTO_IPV6, MCAST_JOIN_GROUP, (char *)&mreq, sizeof(mreq));
マルチキャストグループからの離脱
IPv4
group_req mreq
setsockopt(socket, IPPROTO_IP, MCAST_LEAVE_GROUP, (char *)&mreq, sizeof(mreq));
IPv6
group_req mreq
setsockopt(socket, IPPROTO_IPV6, MCAST_LEAVE_GROUP, (char *)&mreq, sizeof(mreq));
それではUDPMulticastReceiver for Windowsを作ることにしましょう。
プログラム起動時に参加するマルチキャスト(マルチキャストグループ、ポート）と
使用するインターフェースを指定できるようにします。
起動後メニューで参加/離脱を指定できるようにします。
プログラム起動引数 <Multicast IP> <Port> [<InterfaceName>] メニュー +：参加 -：離脱 q：終了
UDPMulticastReceiverの作成（Windows）
UDPServer for Windowsをコピーしてフォルダ名をUDPMulticastReceiverに変更します。
以下のファイルはそのまま（変更なく）使用します。
define.h
MySyncObject.cpp, MySyncObject.h
RingBuff.cpp, RingBuff.h
stdThread.cpp, stdThread.h
ThreadJob.cpp, ThreadJob.h
プロジェクト名、ソリューション名をUDPMulticastReceiverに変更します。
UDPServer.cppをUDPMulticastReceiver.cppに変更します。
SendRecvThread.cpp, SendRecvThread.hはメッセージを送り返す部分を削除します。
UDPMulticastReceiver.cppのUDPServer.cppからの変更点は次の通りです。
(1)受信ソケット、受信スレッドは1つ（指定したマルチキャストアドレスで決まる）
(2)起動パラメータの変更、マルチキャスト対応に変更した機能を呼び出す
main
(3)メニュー対応
DispMenu、GetKeyString
(4)ソケット作成に指定されたマルチキャストアドレスを指定する
bindはAI_PASSIVEで行う(アドレスを指定しない)
CreateAndBindSocket、SetRecvSocketOption
(5)マルチキャストグループへの参加・離脱を行う
main、SetMutiticastInfo、GetInterfaceInfo、MGroupControl
変更を順番に行うことにしましょう。
(1)受信ソケット、受信スレッドは1つ（指定したマルチキャストアドレスで決まる）
MAX_SOCKET_NUMで指定している配列から1つの変数にします。
【UDPMulticastReceiver.cpp】 // 変数の宣言 SOCKET m_fdServer; // ★(1)bind済みUDPソケット CSendRecvThread *m_pCSendRecvThread = NULL; // ★(1) ConnectionInfoRec *m_pConInfo = NULL; // ★(1) CMySyncObject *m_pCMySyncObject = NULL; // ★(1) //============================================== // function // ★(1)ソケットの破棄、受信スレッドの破棄 // parameter // なし // return // なし //============================================== void Stop() { fprintf(stderr, “Stop()\n”); //受信スレッドの破棄, 接続済みソケットの破棄 if (m_pCSendRecvThread != NULL) { m_pCSendRecvThread->End(); m_pCSendRecvThread->WaitForEnd(); SAFE_DELETE(m_pCSendRecvThread); DestroySocket(m_pConInfo->fdClient); SAFE_FREE(m_pConInfo) } }
(3)メニュー対応
【UDPMulticastReceiver.cpp】 #define CMD_QUIT_CHAR ‘q’ // ★(3) #define CMD_JOIN_CHAR ‘+’ // ★(3) #define CMD_LEAVE_CHAR ‘-‘ // ★(3) // 関数の宣言 void DispMenu(); // ★(3) int GetKeyString(LPSTR pszString, int iSize); // ★(3) //============================================== // function // ★メニューの表示 // parameter // なし // return // なし //============================================== void DispMenu() { fprintf(stderr, “%c:quit %c:Join %c:Leave : “, CMD_QUIT_CHAR, CMD_JOIN_CHAR, CMD_LEAVE_CHAR); } //============================================== // function // キーボード入力(改行まで)の取得 // 入力がないときはすぐにリターン // parameter // LPSTR pszString [in/out]データ格納用バッファ // int iSize [in]バッファのサイズ // return // 0:入力なし CMD_QUIT_CHAR:終了 CMD_SEND_MSG_CHAR:メッセージ送信 //============================================== int GetKeyString(LPSTR pszString, int iSize) { int iRet = 0; memset(pszString, 0, iSize); if (_kbhit()) { switch (_getch()) { case CMD_QUIT_CHAR: iRet = CMD_QUIT_CHAR; break; case CMD_JOIN_CHAR: iRet = CMD_JOIN_CHAR; break; case CMD_LEAVE_CHAR: iRet = CMD_LEAVE_CHAR; break; default: break; } } return(iRet); }
(5)マルチキャストグループへの参加・離脱を行う
マルチキャスト制御用の変数の追加
起動引数で指定されたマルチキャストグループを変数に格納関数の追加
マルチキャスト制御関数の追加
【UDPMulticastReceiver.cpp】 // 関数の宣言 void SetMutiticastInfo(LPSTR szIFName, LPSTR szAddr, LPSTR szPort); // ★(5) sockaddr *GetInterfaceInfo(int iFamily, LPCSTR szIFName, DWORD *pdwIndex); // ★(5) BOOL MGroupControl(BOOL fJoin, SOCKET fdRecv, group_req *pmreq); // ★(5) // 変数の宣言 group_req m_mreq; // ★(5) //============================================== // function // ★(5)マルチキャスト参加・離脱用情報取得 // アドレスとポートがm_mreqに格納される // parameter // LPSTR szIFName [in]インターフェース/NULL // LPSTR szAddr [in]マルチキャストアドレス // LPCSTR szPort [in]ポート番号 // return // TRUE/FALSE //============================================== void SetMutiticastInfo(LPSTR szIFName, LPSTR szAddr, LPSTR szPort) { sockaddr *pAddr; addrinfo hints, *pres = NULL; DWORD dwIFIndex; memset(&m_mreq, 0, sizeof(group_req)); memset(&hints, 0, sizeof(hints)); hints.ai_socktype = SOCK_DGRAM; if (getaddrinfo(szAddr, szPort, &hints, &pres) != 0) { DispErrorMsg(“Err:getaddrinfo()”); goto L_END; } pAddr = GetInterfaceInfo(pres->ai_family, szIFName, &dwIFIndex); SAFE_FREE(pAddr) m_mreq.gr_interface = dwIFIndex; // インデックスを格納 memcpy(&(m_mreq.gr_group), pres->ai_addr, pres->ai_addrlen); // アドレスを格納 L_END: if (pres != NULL) freeaddrinfo(pres); return; } //==============================================——————- // Function // ★(5)マルチキャストグループ制御(IPv4/IPv6) // Parameter // BOOL fJoin [in]TRUE:参加 FALSE:離脱 // SOCKET fd [in]ソケット // struct group_req *pmreq [in]対象とするマルチキャストグループ // Return // TRUE/FALSE //==============================================——————- BOOL MGroupControl(BOOL fJoin, SOCKET fdRecv, group_req *pmreq) { int iRet = -1; BOOL fRet; if (fJoin == TRUE) { if (pmreq->gr_group.ss_family == AF_INET) iRet = setsockopt(fdRecv, IPPROTO_IP, MCAST_JOIN_GROUP, (char *)pmreq, sizeof(struct group_req)); else if (pmreq->gr_group.ss_family == AF_INET6) iRet = setsockopt(fdRecv, IPPROTO_IPV6, MCAST_JOIN_GROUP, (char *)pmreq, sizeof(struct group_req)); } else { if (pmreq->gr_group.ss_family == AF_INET) iRet = setsockopt(fdRecv, IPPROTO_IP, MCAST_LEAVE_GROUP, (char *)pmreq, sizeof(struct group_req)); else if (pmreq->gr_group.ss_family == AF_INET6) iRet = setsockopt(fdRecv, IPPROTO_IPV6, MCAST_LEAVE_GROUP, (char *)pmreq, sizeof(struct group_req)); } fRet = (iRet == 0) ? TRUE : FALSE; if (fRet == FALSE) DispErrorMsg(“Err:MGroupControl”); return(fRet); }
sockaddr *GetInterfaceInfo(int iFamily, LPCSTR szIFName, DWORD *pdwIndex)
はUDPMulticastSender.cppで使ったものと同じなのでコピーしてください。
(4)ソケット作成に指定されたマルチキャストアドレスを指定する
bindはAI_PASSIVEで行う(アドレスを指定しない)
【UDPMulticastReceiver.cpp】 // 関数の宣言 BOOL CreateAndBindSocket(LPCSTR szAddress, WORD wPort); // ★(4)ソケットの作成と名前付け BOOL SetRecvSocketOption(SOCKET fd, int iFamily, LPSTR szPort); // ★(4) //============================================== // function // ★(4)ソケットの作成とbindの実施 // m_fdServerに格納される // parameter // LPCSTR szAddress [in]マルチキャストアドレス // WORD wPort [in]受信ポート番号 // return // TRUE/FALSE //============================================== BOOL CreateAndBindSocket(LPCSTR szAddress, WORD wPort) { BOOL fRet = FALSE; SOCKET fd = INVALID_SOCKET; addrinfo hints, *pres = NULL, *pTemp = NULL; char szPort[NI_MAXSERV]; fprintf(stderr, “CreateAndBindSocket()\n”); memset(&hints, 0, sizeof(hints)); hints.ai_socktype = SOCK_DGRAM; DISABLE_C4996 sprintf(szPort, “%d”, wPort); ENABLE_C4996 // ソケット(IPv4/v6)を作るためにはアドレスがわかれば良い if (getaddrinfo(szAddress, NULL, &hints, &pres) == 0) { // 順番にpresの中に格納されている情報を使用する pTemp = pres; while (pTemp != NULL) { // ソケット作成 if((fd = socket(pTemp->ai_family, pTemp->ai_socktype, pTemp->ai_protocol)) == INVALID_SOCKET) goto L_NEXT; fprintf(stderr, “%d %d %d %d\n”, (int)fd, pTemp->ai_family, AF_INET, AF_INET6); // 受信ソケットの設定(bind) if (SetRecvSocketOption(fd, pTemp->ai_family, szPort) == FALSE) { DispErrorMsg(“Err:SetRecvSocketOption”); DestroySocket(fd); goto L_NEXT; } m_fdServer = fd; fRet = TRUE; break; L_NEXT: pTemp = pTemp->ai_next; } freeaddrinfo(pres); } if (m_fdServer == INVALID_SOCKET) // ソケット作成に失敗 fRet = FALSE; return(fRet); } //============================================== // function // ★(4)受信ソケットの設定 // bind // parameter // SOCKET fd [in]ソケット // int iFamily [in]ファミリ // LPCSTR szPort [in]ポート番号 // return // TRUE/FALSE //============================================== BOOL SetRecvSocketOption(SOCKET fd, int iFamily, LPSTR szPort) { BOOL fRet = FALSE; addrinfo hints, *pres = NULL, *ptemp = NULL; int on = 1; #ifdef IPV6_V6ONLY // IPv6ソケットでIPv4射影アドレスを使用しないように設定 if (iFamily == AF_INET6) { on = 1; if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&on, sizeof(on)) < 0) { DispErrorMsg(“Err:setsockopt()”); DestroySocket(fd); goto L_END; } } #endif // UDPではSO_REUSEADDR使わない // bind(ポートを指定し、アドレスは指定しない(PASSIVE)) memset(&hints, 0, sizeof(hints)); hints.ai_socktype = SOCK_DGRAM; // UDP hints.ai_family = iFamily; hints.ai_flags = AI_PASSIVE; if (getaddrinfo(NULL, szPort, &hints, &pres) != 0) { DispErrorMsg(“Err:getaddrinfo()”); goto L_END; } if (bind(fd, pres->ai_addr, (int)pres->ai_addrlen) == SOCKET_ERROR) { DispErrorMsg(“Err:bind()”); goto L_END; } fRet = TRUE; L_END: if (pres != NULL) freeaddrinfo(pres); return(fRet); }
(2)起動パラメータの変更、マルチキャスト対応に変更した機能を呼び出す
【UDPMulticastReceiver.cpp】 int main(int argc, char *argv[]) { int iRet = -1; char szKeyInBuff[81] = {0}; LPSTR pszIFName = NULL; // ★送信元インターフェース名 // EnableESC(stderr); // ★動作を確認するためECSシーケンスによる画面制御をOFFに Locate(1, 1, 2); m_pCMySyncObject = new CMySyncObject(); m_pCMySyncObject->Initialize(); // ★起動パラメータチェック if ((argc != 3) && (argc != 4)) { fprintf(stderr, “%s <Multicast IP> <Port> [<InterfaceName>]\n”, argv[0]); goto L_END; } if (argc == 4) pszIFName = argv[3]; // ソケットライブラリの初期化 if (InitSocketLib() == FALSE) goto L_END; // ★変数の初期化 m_fdServer = INVALID_SOCKET; // ★UDPソケットの作成とbind if (CreateAndBindSocket(argv[1], (WORD)atol(argv[2])) == FALSE) goto L_END; // ★マルチキャスト参加・離脱用情報取得 SetMutiticastInfo(pszIFName, argv[1], argv[2]); // ★送受信スレッドの作成 m_pConInfo = (ConnectionInfoRec *)calloc(1, sizeof(ConnectionInfoRec)); m_pConInfo->pCMySyncObject = m_pCMySyncObject; m_pConInfo->fdClient = m_fdServer; m_pCSendRecvThread = new CSendRecvThread(m_pConInfo); m_pCSendRecvThread->Begin(); // ★メニュー表示と処理の実行 DispMenu(); while (1) { Sleep(10); // Sleepを入れておく方がCPU負荷のことを考えると良いでしょう switch (GetKeyString(szKeyInBuff, sizeof(szKeyInBuff))) { case 0: break; case CMD_QUIT_CHAR: goto L_END; break; case CMD_JOIN_CHAR: MGroupControl(TRUE, m_fdServer, &m_mreq); DispMenu(); break; case CMD_LEAVE_CHAR: MGroupControl(FALSE, m_fdServer, &m_mreq); DispMenu(); break; } } iRet = 0; L_END: // すべてのソケットの破棄 Stop(); m_pCMySyncObject->Uninitialize(); SAFE_DELETE(m_pCMySyncObject) // ソケットライブラリの開放 UninitSocketLib(); return(iRet); }
最後にCSendRecvThreadからメッセージを送り返す部分を削除します。
【SendRecvThread.cpp】 //============================================== // function // メッセージコマンドの受信 // parameter // HeaderRec *pHeader [in]仮読みしたヘッダ // return // 0:パケットが揃っていない // 1:パケットを受信したので処理を行った // -1:エラーが発生した //============================================== BOOL CSendRecvThread::RecvMessagePacket(HeaderRec *pHeader) { … // SendMessagePacket(pszMsg, iMsgSize); … }
UDPMulticastSenderを動かして、メッセージ送信を開始します。
【UDPMulticastSender】 C:\>UDPMulticastSender.exe ff1e::1 10000 q:quit m:メッセージ送信 :m 送信メッセージを入力してください : aaaaaaaaaaa 【データ送信開始】
UDPMulticastReceiverを動かします。
JOINするまで、データは受信しません。
JOINしてデータを受信することがわかります。
【UDPMulticastReceiver】 C:\>UDPMulticastReceiver.exe ff1e::1 10000 CreateAndBindSocket() 276 23 2 23 q:quit +:Join -:Leave : + 【ここでJOINしてデータ受信開始】 recvfrom socket:276 Addr:2001:ce8:132:4332:3c16:217e:6aae:dc25 Port:56490 498660 :CMD:100Msg:recv 12aaaaaaaaaaa recvfrom socket:276 Addr:2001:ce8:132:4332:3c16:217e:6aae:dc25 Port:56490 498665 :CMD:100Msg:recv 13aaaaaaaaaaa recvfrom socket:276 Addr:2001:ce8:132:4332:3c16:217e:6aae:dc25 Port:56490 498670 :CMD:100Msg:recv 14aaaaaaaaaaa q:quit +:Join -:Leave : – 【ここでLEAVEしてデータ受信停止】 q:quit +:Join -:Leave : + 【ここでJOINしてデータ受信再開】 recvfrom socket:276 Addr:2001:ce8:132:4332:3c16:217e:6aae:dc25 Port:56490 498685 :CMD:100Msg:recv 17aaaaaaaaaaa recvfrom socket:276 Addr:2001:ce8:132:4332:3c16:217e:6aae:dc25 Port:56490 498690 :CMD:100Msg:recv 18aaaaaaaaaaa
Linux版も同様の対応で作成することができます。
以下はプロジェクトのリンクです。
UDPMulticastReceiver for Windows
UDPMulticastReceiver for Linux