Add UDP group

README-CN.md

@@ -420,6 +420,7 @@ int main(int argc, char** argv) {
 - TCP客户端: [evpp/TcpClient_test.cpp](evpp/TcpClient_test.cpp)
 - UDP服务端: [evpp/UdpServer_test.cpp](evpp/UdpServer_test.cpp)
 - UDP客户端: [evpp/UdpClient_test.cpp](evpp/UdpClient_test.cpp)
+- UDP组播接收:[evpp/UdpGroupDest_test.cpp](evpp/UdpGroupDest_test.cpp)
 - HTTP服务端: [examples/http_server_test.cpp](examples/http_server_test.cpp)
 - HTTP客户端: [examples/http_client_test.cpp](examples/http_client_test.cpp)
 - WebSocket服务端: [examples/websocket_server_test.cpp](examples/websocket_server_test.cpp)

README.md

@@ -360,6 +360,7 @@ int main(int argc, char** argv) {
 - [evpp/TcpClient_test.cpp](evpp/TcpClient_test.cpp)
 - [evpp/UdpServer_test.cpp](evpp/UdpServer_test.cpp)
 - [evpp/UdpClient_test.cpp](evpp/UdpClient_test.cpp)
+- [evpp/UdpGroupDest_test.cpp](evpp/UdpGroupDest_test.cpp)
 - [examples/http_server_test.cpp](examples/http_server_test.cpp)
 - [examples/http_client_test.cpp](examples/http_client_test.cpp)
 - [examples/websocket_server_test.cpp](examples/websocket_server_test.cpp)

base/hsocket.h

@@ -199,6 +199,31 @@ HV_INLINE int udp_broadcast(int sockfd, int on DEFAULT(1)) {
  return setsockopt(sockfd, SOL_SOCKET, SO_BROADCAST, (const char*)&on, sizeof(int));
 }
 
+HV_INLINE int udp_joingroupv4(int sockfd, const char* group, const char* local_host) {
+ struct ip_mreq mreq;
+ mreq.imr_multiaddr.s_addr = inet_addr(group);
+ mreq.imr_interface.s_addr = inet_addr(local_host);
+ return setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (const char*)&mreq, sizeof(mreq));
+}
+HV_INLINE int udp_leavegroupv4(int sockfd, const char* group, const char* local_host) {
+ struct ip_mreq mreq;
+ mreq.imr_multiaddr.s_addr = inet_addr(group);
+ mreq.imr_interface.s_addr = inet_addr(local_host);
+ return setsockopt(sockfd, IPPROTO_IP, IP_DROP_MEMBERSHIP, (const char*)&mreq, sizeof(mreq));
+}
+HV_INLINE int udp_joingroupv6(int sockfd, const char* group, ULONG interface) {
+ struct ipv6_mreq mreq;
+ mreq.ipv6mr_interface = interface;
+ inet_pton(AF_INET6, group, &mreq.ipv6mr_multiaddr);
+ return setsockopt(sockfd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, (const char*)&mreq, sizeof(mreq));
+}
+HV_INLINE int udp_leavegroupv6(int sockfd, const char* group, ULONG interface) {
+ struct ipv6_mreq mreq;
+ mreq.ipv6mr_interface = interface;
+ inet_pton(AF_INET6, group, &mreq.ipv6mr_multiaddr);
+ return setsockopt(sockfd, IPPROTO_IPV6, IPV6_DROP_MEMBERSHIP, (const char*)&mreq, sizeof(mreq));
+}
+
 HV_INLINE int ip_v6only(int sockfd, int on DEFAULT(1)) {
 #ifdef IPV6_V6ONLY
  return setsockopt(sockfd, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&on, sizeof(int));

evpp/README.md

@@ -19,6 +19,6 @@ hloop.h中的c接口被封装成了c++的类，参考了muduo和evpp。
 ├── TcpClient.h TCP客户端类
 ├── TcpServer.h TCP服务端类
 ├── UdpClient.h UDP客户端类
-└── UdpServer.h UDP服务端类
-
+├── UdpServer.h UDP服务端类
+└── UdpGroup.h UDP组播类
 ```

evpp/UdpGroup.h

@@ -0,0 +1,170 @@
+#ifndef HV_UDP_GROUP_HPP_
+#define HV_UDP_GROUP_HPP_
+#include "hsocket.h"
+
+#include "EventLoopThreadPool.h"
+#include "Channel.h"
+
+namespace hv {
+
+template<class TSocketChannel = SocketChannel>
+class UdpGroupEventLoopTmpl {
+public:
+ typedef std::shared_ptr<TSocketChannel> TSocketChannelPtr;
+
+ UdpGroupEventLoopTmpl(EventLoopPtr loop = NULL) {
+ loop_ = loop ? loop : std::make_shared<EventLoop>();
+ port = 0;
+#if WITH_KCP
+ kcp_setting = NULL;
+#endif
+ }
+
+ virtual ~UdpGroupEventLoopTmpl() {
+#if WITH_KCP
+ HV_FREE(kcp_setting);
+#endif
+ }
+
+ const EventLoopPtr& loop() {
+ return loop_;
+ }
+
+ //use createsocket() for server OR use createsocketRemote() for client
+ //@retval >=0 bindfd, <0 error
+ int createsocket(int port, const char* host = "0.0.0.0") {
+ hio_t* io = hloop_create_udp_server(loop_->loop(), host, port);
+ if (io == NULL) return -1;
+ this->host = host;
+ this->port = port;
+ channel = std::make_shared<TSocketChannel>(io);
+ return channel->fd();
+ }
+
+ // join group
+ int joinGroup(const char* g){
+ if(channel == NULL || channel->isClosed()){
+ return -1;
+ }
+ return udp_joingroupv4(channel->fd(), g, host.c_str());
+ }
+ // leave group
+ int leaveGroup(const char* g){
+ if(channel == NULL || channel->isClosed()){
+ return -1;
+ }
+ return udp_leavegroupv4(channel->fd(), g, host.c_str());
+ }
+
+ // closesocket thread-safe
+ void closesocket() {
+ if (channel) {
+ channel->close(true);
+ }
+ }
+
+ int startRecv() {
+ if (channel == NULL || channel->isClosed()) {
+ return -1;
+ }
+ channel->onread = [this](Buffer* buf) {
+ if (onMessage) {
+ onMessage(channel, buf);
+ }
+ };
+#if WITH_KCP
+ if (kcp_setting) {
+ hio_set_kcp(channel->io(), kcp_setting);
+ }
+#endif
+ return channel->startRead();
+ }
+
+ int stopRecv() {
+ if (channel == NULL) return -1;
+ return channel->stopRead();
+ }
+
+ // start thread-safe
+ void start() {
+ loop_->runInLoop(std::bind(&UdpGroupEventLoopTmpl::startRecv, this));
+ }
+
+#if WITH_KCP
+ void setKcp(kcp_setting_t* setting) {
+ if (setting == NULL) {
+ HV_FREE(kcp_setting);
+ return;
+ }
+ if (kcp_setting == NULL) {
+ HV_ALLOC_SIZEOF(kcp_setting);
+ }
+ *kcp_setting = *setting;
+ }
+#endif
+
+public:
+ std::string host;
+ int port;
+ TSocketChannelPtr channel;
+#if WITH_KCP
+ kcp_setting_t* kcp_setting;
+#endif
+ // Callback
+ std::function<void(const TSocketChannelPtr&, Buffer*)> onMessage;
+private:
+ EventLoopPtr loop_;
+};
+
+template<class TSocketChannel = SocketChannel>
+class UdpGroupTmpl : private EventLoopThread, public UdpGroupEventLoopTmpl<TSocketChannel> {
+public:
+ UdpGroupTmpl(EventLoopPtr loop = NULL)
+ : EventLoopThread(loop)
+ , UdpGroupEventLoopTmpl<TSocketChannel>(EventLoopThread::loop())
+ , is_loop_owner(loop == NULL)
+ {}
+ virtual ~UdpGroupTmpl() {
+ stop(true);
+ }
+
+ const EventLoopPtr& loop() {
+ return EventLoopThread::loop();
+ }
+
+ // join group
+ int joinGroup(const char* g){
+ return UdpGroupEventLoopTmpl<TSocketChannel>::joinGroup(g);
+ }
+
+ // leave group
+ int leaveGroup(const char* g){
+ return UdpGroupEventLoopTmpl<TSocketChannel>::leaveGroup(g);
+ }
+
+ // start thread-safe
+ void start(bool wait_threads_started = true) {
+ if (isRunning()) {
+ UdpGroupEventLoopTmpl<TSocketChannel>::start();
+ } else {
+ EventLoopThread::start(wait_threads_started, std::bind(&UdpGroupTmpl::startRecv, this));
+ }
+ }
+
+ // stop thread-safe
+ void stop(bool wait_threads_stopped = true) {
+ UdpGroupEventLoopTmpl<TSocketChannel>::closesocket();
+ if (is_loop_owner) {
+ EventLoopThread::stop(wait_threads_stopped);
+ }
+ }
+
+private:
+ bool is_loop_owner;
+};
+
+typedef UdpGroupTmpl<SocketChannel> UdpGroup;
+
+}
+
+#endif // HV_UDP_GROUP_HPP_

evpp/UdpGroupDest_test.cpp

@@ -0,0 +1,57 @@
+/*
+ * UdpGroupDest_test.cpp
+ *
+ * @build make evpp
+ * @client bin/UdpClient_test 9997 230.1.1.25
+ * @server bin/UdpGroupDest_test 9997 230.1.1.25
+ *
+ */
+
+#include <iostream>
+
+#include "UdpGroup.h"
+
+using namespace hv;
+
+int main(int argc, char* argv[]) {
+ if (argc < 2) {
+ printf("Usage: %s port\n", argv[0]);
+ return -10;
+ }
+ int port = atoi(argv[1]);
+ const char* group = "230.1.1.25";
+ if (argc > 2) {
+ group = argv[2];
+ }
+
+ UdpGroup ug;
+ int bindfd = ug.createsocket(port);
+ if (bindfd < 0) {
+ return -20;
+ }
+ int rst = ug.joinGroup(group);
+ printf("udp group bind on port %d, bindfd=%d rst=%d ...\n", port, bindfd, rst);
+ ug.onMessage = [](const SocketChannelPtr& channel, Buffer* buf) {
+ // echo
+ printf(" >>> %.*s\n", (int)buf->size(), (char*)buf->data());
+ };
+ ug.start();
+
+ std::string str;
+ while (std::getline(std::cin, str)) {
+ if (str == "close") {
+ ug.closesocket();
+ break;
+ } else if (str == "leave") {
+ int c = ug.leaveGroup(group);
+ printf("leave group rst=%d\n", c);
+ } else if (str == "join") {
+ int c = ug.joinGroup(group);
+ printf("join group rst=%d\n", c);
+ } else {
+ break;
+ }
+ }
+
+ return 0;
+}