多播的概念与组播地址范围及协议描述参考：
https://support.huawei.com/enterprise/zh/doc/EDOC1000017255/17fa1d44
int32_t ret = IndustrialSocketCreate(AF_INET, SOCK_DGRAM, 0, &softbus->publishFd_);
if (ret != 0) {
IBHILOGE("IndustrialSocketCreate send cast udp failed");
return ;
}
while (softbus->GetPublishState()) {
struct ip_mreq mreq;
// 设置多播地址
mreq.imr_multiaddr.s_addr = inet_addr(MULTICAST_IP);
// 设置使用该多播地址发送数据时使用的网卡ip
mreq.imr_interface.s_addr = inet_addr(softbus->localIp_.c_str());
// 调用setsockopt
ret = setsockopt(softbus->publishFd_, IPPROTO_IP, IP_MULTICAST_IF, &mreq,sizeof(mreq));
if ( ret < 0) {
IBHILOGE("Fail to join udp group, err: %{public}s", strerror(errno));
usleep(PUBLISH_INTERVAL);
} else {
break;
}
}
while (softbus->GetPublishState()) {
// 禁止组播回送（防止收到自己发送的组播包）
int op = 0;
ret = setsockopt(softbus->publishFd_, IPPROTO_IP, IP_MULTICAST_LOOP, &op, sizeof(op));
if (ret < 0) {
IBHILOGE("Fail to disable multicast loop, err: %{public}s", strerror(errno));
usleep(PUBLISH_INTERVAL);
} else {
break;
}
}
// 指定发送的组播地址和端口
struct sockaddr_in mcast_addr;
memset(&mcast_addr, 0, sizeof(mcast_addr));
mcast_addr.sin_family = AF_INET;
mcast_addr.sin_addr.s_addr = inet_addr(MULTICAST_IP);
mcast_addr.sin_port = htons(MULTICAST_PORT);
softbus->publishThread.SetThreadState(true);
int32_t stopPublishNum = 0;
while (softbus->publishThread.GetThreadState()) {
if (softbus->GetPublishState() && softbus->publishCount_ < MAX_PUBLISH_TIME) {
std::string sendMulicastData(MULTICAST_DATA);
sendMulicastData = sendMulicastData + softbus->localMac_;
sendMulicastData.append(buffer);
IBHILOGI("publishFd_ :%{public}d multicast data:%{public}s begin",
softbus->publishFd_,sendMulicastData.c_str());
ret = sendto(softbus->publishFd_, sendMulicastData.c_str(),
sendMulicastData.length(), 0, (const struct sockaddr *)&mcast_addr, sizeof(mcast_addr));
if (ret < 0) {
IBHILOGE("sendto publish msg:%{public}d fail,sendLength:%{public}d", softbus->publishFd_, ret);
} else {
// IBHILOGI("sendto publish msg:%{public}d succ,sendLength:%{public}d", softbus->publishFd_, ret);
}
softbus->publishCount_++;
usleep(PUBLISH_INTERVAL);
} else {
stopPublishNum++;
usleep(PUBLISH_INTERVAL);
if (stopPublishNum >= STOP_PUBLISH_TIME) {
softbus->publishCount_ = 0;
stopPublishNum = 0;
}
}
}
DestroyBusSocket(&softbus->publishFd_);
int32_t ret = IndustrialSocketCreate(AF_INET, SOCK_DGRAM, 0, &softbus->discoveryFd_);
if (ret != 0)
IBHILOGE("IndustrialSocketCreate recv cast udp failed");
return ;
}
while (softbus->GetDiscoveryState()) {
int reuse_addr_opt = 1;
int error = setsockopt(softbus->discoveryFd_, SOL_SOCKET, SO_REUSEADDR, &reuse_addr_opt, sizeof(int));
if (error < 0) {
IBHILOGE("setsockopt fail error:%{public}d,errno:%{public}d", error, errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
while (softbus->GetDiscoveryState()) {
// 设置超时
struct timeval timeout;
timeout.tv_sec = 0;//秒
timeout.tv_usec = USER_TIMEOUT_MS;//微秒
if (setsockopt(softbus->discoveryFd_, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) < 0) {
IBHILOGE("setsockopt SO_RCVTIMEO failed errno:%{public}d", errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
struct sockaddr_in server_socket;
while (softbus->GetDiscoveryState()) {
bzero(&server_socket, sizeof(server_socket));
server_socket.sin_family = AF_INET;
server_socket.sin_addr.s_addr = htonl(INADDR_ANY);
server_socket.sin_port = htons(MULTICAST_PORT);
// 调用bind绑定socket地址
if (bind(softbus->discoveryFd_, (struct sockaddr*)&server_socket, sizeof(struct sockaddr)) == -1) {
IBHILOGE("bind error errno:%{public}d", errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
struct ip_mreq mreq;
while (softbus->GetDiscoveryState()) {
// 加入组播，接收组播信息
mreq.imr_multiaddr.s_addr = inet_addr(MULTICAST_IP);
mreq.imr_interface.s_addr = inet_addr(softbus->localIp_.c_str());
int error = setsockopt(softbus->discoveryFd_, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
if (error < 0) {
IBHILOGE("setsockopt fail error:%{public}d,errno:%{public}d", error, errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
char recvBuff[100] = {0};
softbus->discoveryThread.SetThreadState(true);
while (softbus->discoveryThread.GetThreadState()) {
if (softbus->GetDiscoveryState()) {
struct sockaddr_in recvAddr;
memset(recvBuff, 0, sizeof(recvBuff));
memset(&recvAddr, 0, sizeof(struct sockaddr));
int sinSize = sizeof(struct sockaddr_in);
int ret = recvfrom(softbus->discoveryFd_, recvBuff, sizeof(recvBuff) - 1, 0,
(struct sockaddr*)&recvAddr, (socklen_t *)&sinSize);
if (ret > 0 ) {
IBHILOGI("recvfrom client IP:%{public}s cast:%{public}s ret:%{public}d",
inet_ntoa(recvAddr.sin_addr), recvBuff, ret);
std::string recvMulticastData = recvBuff;
if (recvMulticastData.length() != (MULTICAST_LEN + MAC_ADDR_LEN + TIME_STAMP_LEN)) {
usleep(DISCOVERY_INTERVAL);
continue;
} else {
std::string auth = recvMulticastData.substr(0, MULTICAST_LEN);
std::string peerMac = recvMulticastData.substr(MULTICAST_LEN, MAC_ADDR_LEN);
if (auth.compare(MULTICAST_DATA) == 0 && softbus->localMac_.compare(peerMac) != 0) { // 校验组播数据
std::string peerStamp =
recvMulticastData.substr(MULTICAST_LEN + MAC_ADDR_LEN, TIME_STAMP_LEN);
std::string peerAddr = inet_ntoa(recvAddr.sin_addr);
int32_t iFound = softbus->DiscoveryUpdateDeviceInfo(peerMac, peerAddr, peerStamp);
if (iFound >=0) {
softbus->NotifyDeviceFound(peerMac, peerAddr, peerStamp);
if (iFound > 0) {
softbus->UpdatePeerJoinInfo(peerAddr);
}
}
}
usleep(DISCOVERY_INTERVAL);
}
} else {
IBHILOGI("recvfrom error:%{public}d ret:%{public}d, retry recvfrom cast", errno, ret);
usleep(DISCOVERY_INTERVAL);
continue;
}
} else {
IBHILOGI("UnDiscoverySoftbus................");
usleep(DISCOVERY_INTERVAL);
}
}
DestroyBusSocket(&softbus->discoveryFd_);
广播方式可以通过本地地址与掩码方式计算出广播地址，本例中采用255.255.255.255地址，不需要路由器直接全网发送
int32_t ret = IndustrialSocketCreate(AF_INET, SOCK_DGRAM, 0, &softbus->publishFd_);
if (ret != 0) {
IBHILOGE("IndustrialSocketCreate send cast udp failed");
return ;
}
while (softbus->GetPublishState()) {
// 允许发送广播
int on_off = 1;
ret = setsockopt(softbus->publishFd_, SOL_SOCKET, SO_BROADCAST, &on_off, sizeof(on_off));
if ( ret < 0) {
IBHILOGE("Fail to send broadcast, err: %{public}s", strerror(errno));
usleep(PUBLISH_INTERVAL);
} else {
break;
}
}
// 指定发送的广播地址和端口
struct sockaddr_in mcast_addr;
memset(&mcast_addr, 0, sizeof(mcast_addr));
mcast_addr.sin_family = AF_INET;
mcast_addr.sin_addr.s_addr = inet_addr("255.255.255.255"); // 不经由路由器全网发送
mcast_addr.sin_port = htons(MULTICAST_PORT);
softbus->publishThread.SetThreadState(true);
int32_t stopPublishNum = 0;
while (softbus->publishThread.GetThreadState()) {
if (softbus->GetPublishState() && softbus->publishCount_ < MAX_PUBLISH_TIME) {
std::string sendMulicastData(MULTICAST_DATA);
sendMulicastData = sendMulicastData + softbus->localMac_;
sendMulicastData.append(buffer);
IBHILOGI("publishFd_ :%{public}d multicast data:%{public}s begin",
softbus->publishFd_,sendMulicastData.c_str());
ret = sendto(softbus->publishFd_, sendMulicastData.c_str(),
sendMulicastData.length(), 0, (const struct sockaddr *)&mcast_addr, sizeof(mcast_addr));
if (ret < 0) {
IBHILOGE("sendto publish msg:%{public}d fail,sendLength:%{public}d", softbus->publishFd_, ret);
} else {
// IBHILOGI("sendto publish msg:%{public}d succ,sendLength:%{public}d", softbus->publishFd_, ret);
}
softbus->publishCount_++;
usleep(PUBLISH_INTERVAL);
} else {
stopPublishNum++;
usleep(PUBLISH_INTERVAL);
if (stopPublishNum >= STOP_PUBLISH_TIME) {
softbus->publishCount_ = 0;
stopPublishNum = 0;
}
}
}
DestroyBusSocket(&softbus->publishFd_);
int32_t ret = IndustrialSocketCreate(AF_INET, SOCK_DGRAM, 0, &softbus->discoveryFd_);
if (ret != 0) {
IBHILOGE("IndustrialSocketCreate recv cast udp failed");
return ;
}
while (softbus->GetDiscoveryState()) {
int reuse_addr_opt = 1;
int error = setsockopt(softbus->discoveryFd_, SOL_SOCKET, SO_REUSEADDR, &reuse_addr_opt, sizeof(int));
if (error < 0) {
IBHILOGE("setsockopt fail error:%{public}d,errno:%{public}d", error, errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
while (softbus->GetDiscoveryState()) {
// 设置超时
struct timeval timeout;
timeout.tv_sec = 0;//秒
timeout.tv_usec = USER_TIMEOUT_MS;//微秒
if (setsockopt(softbus->discoveryFd_, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) < 0) {
IBHILOGE("setsockopt SO_RCVTIMEO failed errno:%{public}d", errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
struct sockaddr_in server_socket;
while (softbus->GetDiscoveryState()) {
bzero(&server_socket, sizeof(server_socket));
server_socket.sin_family = AF_INET;
server_socket.sin_addr.s_addr = htonl(INADDR_ANY);
server_socket.sin_port = htons(MULTICAST_PORT);
// 调用bind绑定socket地址
if (bind(softbus->discoveryFd_, (struct sockaddr*)&server_socket, sizeof(struct sockaddr)) == -1) {
IBHILOGE("bind error errno:%{public}d", errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
char recvBuff[100] = {0};
softbus->discoveryThread.SetThreadState(true);
while (softbus->discoveryThread.GetThreadState()) {
if (softbus->GetDiscoveryState()) {
struct sockaddr_in recvAddr;
memset(recvBuff, 0, sizeof(recvBuff));
memset(&recvAddr, 0, sizeof(struct sockaddr));
int sinSize = sizeof(struct sockaddr_in);
int ret = recvfrom(softbus->discoveryFd_, recvBuff, sizeof(recvBuff) - 1, 0,
(struct sockaddr*)&recvAddr, (socklen_t *)&sinSize);
if (ret > 0 ) {
IBHILOGI("recvfrom client IP:%{public}s cast:%{public}s ret:%{public}d",
inet_ntoa(recvAddr.sin_addr), recvBuff, ret);
std::string recvMulticastData = recvBuff;
if (recvMulticastData.length() != (MULTICAST_LEN + MAC_ADDR_LEN + TIME_STAMP_LEN)) {
usleep(DISCOVERY_INTERVAL);
continue;
} else {
std::string auth = recvMulticastData.substr(0, MULTICAST_LEN);
std::string peerMac = recvMulticastData.substr(MULTICAST_LEN, MAC_ADDR_LEN);
if (auth.compare(MULTICAST_DATA) == 0 && softbus->localMac_.compare(peerMac) != 0) { // 校验组播数据
std::string peerStamp =
recvMulticastData.substr(MULTICAST_LEN + MAC_ADDR_LEN, TIME_STAMP_LEN);
std::string peerAddr = inet_ntoa(recvAddr.sin_addr);
int32_t iFound = softbus->DiscoveryUpdateDeviceInfo(peerMac, peerAddr, peerStamp);
if (iFound >=0) {
softbus->NotifyDeviceFound(peerMac, peerAddr, peerStamp);
if (iFound > 0) {
softbus->UpdatePeerJoinInfo(peerAddr);
}
}
}
usleep(DISCOVERY_INTERVAL);
}
} else {
IBHILOGI("recvfrom error:%{public}d ret:%{public}d, retry recvfrom cast", errno, ret);
usleep(DISCOVERY_INTERVAL);
continue;
}
} else {
IBHILOGI("UnDiscoverySoftbus................");
usleep(DISCOVERY_INTERVAL);
}
}
DestroyBusSocket(&softbus->discoveryFd_);
// man connect说明
// The socket is nonblocking and the connection cannot be completed immediately. (UNIX domain sockets failed with EAGAIN // instead.) It is possible to select(2) or poll(2) for completion by selecting the socket for writing. After select(2)
// indicates writability, use getsockopt(2) to read the SO_ERROR option at level SOL_SOCKET to determine whether connect() // completed successfully (SO_ERROR is zero) or unsuccessfully (SO_ERROR is one of the usual error codes listed here, explaining // the reason for the failure).
// 示例代码如下：
int32_t fd = -1;
int32_t ret = IndustrialSocketCreate(AF_INET, SOCK_STREAM, 0, &fd);
struct sockaddr_in sendAddr;
bzero(&sendAddr, sizeof(sendAddr));
sendAddr.sin_family = AF_INET;
sendAddr.sin_port = htons(SEND_DATA_PORT);
sendAddr.sin_addr.s_addr = inet_addr(peerIp.c_str());
ChangeNonBlockMode(fd, true);
ret = connect(fd, (struct sockaddr*)&sendAddr, sizeof(sendAddr));
if (ret < 0 && errno != EINPROGRESS) {
IBHILOGI("connect server fd:%{public}d fail, errno=%{public}d", fd, errno);
DestroyBusSocket(&fd);
return -1;
}
// 判断是否可写
fd_set wset;
FD_ZERO(&wset);
FD_SET(fd,&wset);
struct timeval tv;
tv.tv_sec = 0;
tv.tv_usec = USER_TIMEOUT_MS;
if (select(fd+1, NULL, &wset, NULL, &tv) == 0) {
IBHILOGI("select server fd:%{public}d fail, errno=%{public}d", fd, errno);
DestroyBusSocket(&fd);
return -1;
}
// at level SOL_SOCKET to determine whether connect() completed successfully (SO_ERROR is zero) or unsuccessfully (SO_ERROR // is one of the usual error codes listed here, explaining the reason for the failure).
if (FD_ISSET(fd,&wset))
{
int error;
socklen_t len = sizeof(error);
ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len);
if (ret < 0 || (ret == 0 && error != 0)) {
IBHILOGI("getsockopt server fd:%{public}d fail, "
"error:%{public}d, errno=%{public}d", fd, error, errno);
DestroyBusSocket(&fd);
return -1;
}
}
// 通过SO_RCVTIMEO设置接收超时时间
// 设置超时
while (softbus->GetDiscoveryState()) {
// 设置超时
struct timeval timeout;
timeout.tv_sec = 0;//秒
timeout.tv_usec = USER_TIMEOUT_MS;//微秒
if (setsockopt(softbus->discoveryFd_, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) < 0) {
IBHILOGE("setsockopt SO_RCVTIMEO failed errno:%{public}d", errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
struct sockaddr_in server_socket;
while (softbus->GetDiscoveryState()) {
bzero(&server_socket, sizeof(server_socket));
server_socket.sin_family = AF_INET;
server_socket.sin_addr.s_addr = htonl(INADDR_ANY);
server_socket.sin_port = htons(MULTICAST_PORT);
// 调用bind绑定socket地址
if (bind(softbus->discoveryFd_, (struct sockaddr*)&server_socket, sizeof(struct sockaddr)) == -1) {
IBHILOGE("bind error errno:%{public}d", errno);
usleep(DISCOVERY_INTERVAL);
} else {
break;
}
}
softbus->discoveryThread.SetThreadState(true);
while (softbus->discoveryThread.GetThreadState()) {
if (softbus->GetDiscoveryState()) {
struct sockaddr_in recvAddr;
memset(recvBuff, 0, sizeof(recvBuff));
memset(&recvAddr, 0, sizeof(struct sockaddr));
int sinSize = sizeof(struct sockaddr_in);
int ret = recvfrom(softbus->discoveryFd_, recvBuff, sizeof(recvBuff) - 1, 0,
(struct sockaddr*)&recvAddr, (socklen_t *)&sinSize);
if (ret > 0 ) {
IBHILOGI("recvfrom client IP:%{public}s cast:%{public}s ret:%{public}d",
inet_ntoa(recvAddr.sin_addr), recvBuff, ret);
std::string recvMulticastData = recvBuff;
if (recvMulticastData.length() != (MULTICAST_LEN + MAC_ADDR_LEN + TIME_STAMP_LEN)) {
usleep(DISCOVERY_INTERVAL);
continue;
} else {
std::string auth = recvMulticastData.substr(0, MULTICAST_LEN);
std::string peerMac = recvMulticastData.substr(MULTICAST_LEN, MAC_ADDR_LEN);
if (auth.compare(MULTICAST_DATA) == 0 && softbus->localMac_.compare(peerMac) != 0) { // 校验组播数据
std::string peerStamp =
recvMulticastData.substr(MULTICAST_LEN + MAC_ADDR_LEN, TIME_STAMP_LEN);
std::string peerAddr = inet_ntoa(recvAddr.sin_addr);
int32_t iFound = softbus->DiscoveryUpdateDeviceInfo(peerMac, peerAddr, peerStamp);
if (iFound >=0) {
softbus->NotifyDeviceFound(peerMac, peerAddr, peerStamp);
if (iFound > 0) {
softbus->UpdatePeerJoinInfo(peerAddr);
}
}
}
usleep(DISCOVERY_INTERVAL);
}
} else {
IBHILOGI("recvfrom error:%{public}d ret:%{public}d, retry recvfrom cast", errno, ret);
usleep(DISCOVERY_INTERVAL);
continue;
}
} else {
IBHILOGI("UnDiscoverySoftbus................");
usleep(DISCOVERY_INTERVAL);
}
}
// recv与recvfrom处理相同
if (IndustrialWaitEvent(fd, INDUSTRIAL_SOCKET_IN, USER_TIMEOUT_MS) > 0) {
int32_t bytesReceived = IndustrialSocketRecv(fd, buffer + valread, dataLength-valread, 0);
if (bytesReceived == -1) {
if (errno == EINTR || errno == EAGAIN) {
retry++;
} else {
IBHILOGE("processData:read failed");
retry++;
}
} else if (bytesReceived == 0) {
IBHILOGE("processData:socket error");
return;
} else {
valread += bytesReceived;
}
} else{
IBHILOGE("processData:IndustrialWaitEvent failed");
retry++;
}
void DestroyBusSocket(int32_t *socketFd)
{
IBHILOGE("DestroyBusSocket socketF:%{public}d", *socketFd);
if(*socketFd > 0) {
close(*socketFd);
*socketFd = -1;
}
}
int32_t IndustrialSocketSelect(int32_t nfds, fd_set *readFds, fd_set *writeFds, fd_set *exceptFds,
struct timeval *timeOut)
{
if (timeOut == NULL) {
IBHILOGE("timeOut is null");
return -1;
}
int32_t ret = select(nfds, readFds, writeFds, exceptFds, timeOut);
if (ret < 0) {
IBHILOGE("select : %{public}s", strerror(errno));
}
return ret;
}
int32_t IndustrialSocketCreate(int32_t domain, int32_t type, int32_t protocol, int32_t *socketFd)
{
if (socketFd == NULL) {
IBHILOGE("socketFd is null");
return -1;
}
int32_t ret;
ret = socket(domain, type, protocol);
if (ret < 0) {
IBHILOGE("socket %{public}s", strerror(errno));
return -1;
} else {
*socketFd = ret;
return 0;
}
}
int32_t ChangeNonBlockMode(int32_t fd, bool isNonBlock)
{
if (fd < 0) {
return -1;
}
int32_t flags = fcntl(fd, F_GETFL, 0);
if (flags < 0) {
IBHILOGE("fd=%{public}d,fcntl get flag failed, errno=%{public}d", fd, errno);
return -1;
}
if (isNonBlock && ((uint32_t)flags & O_NONBLOCK) == 0) {
flags = (int32_t)((uint32_t)flags | O_NONBLOCK);
IBHILOGI("fd=%{public}d set to nonblock", fd);
} else if (!isNonBlock && ((uint32_t)flags & O_NONBLOCK) != 0) {
flags = (int32_t)((uint32_t)flags & ~O_NONBLOCK);
IBHILOGI("fd=%{public}d set to block", fd);
} else {
IBHILOGI("fd=%{public}d nonblock state is already ok", fd);
return 0;
}
return fcntl(fd, F_SETFL, flags);
}
int32_t IndustrialSocketSend(int32_t socketFd, const void *buf, uint32_t len, int32_t flags)
{
int32_t wrapperFlag = flags | MSG_NOSIGNAL;
int32_t ret = send(socketFd, buf, len, wrapperFlag);
if (ret < 0) {
IBHILOGI("send : %{public}s, errno:%{public}d", strerror(errno), errno);
return ret;
}
return ret;
}
int32_t IndustrialSocketRecv(int32_t socketFd, void *buf, uint32_t len, int32_t flags)
{
int32_t ret = recv(socketFd, buf, len, flags);
if (ret < 0) {
IBHILOGI("recv : %{public}s errno:%{public}d", strerror(errno), errno);
return ret;
}
return ret;
}
int IndustrialWaitEvent(int fd, short events, int timeout)
{
if (fd < 0) {
IBHILOGI("%{public}s:%{public}d:fd=%{public}d invalid params", __func__, __LINE__, fd);
return -1;
}
struct timeval tv = {0};
tv.tv_sec = 0;
tv.tv_usec = timeout;
int rc = 0;
switch (events) {
case INDUSTRIAL_SOCKET_OUT: {
fd_set writeSet;
FD_ZERO(&writeSet);
FD_SET(fd, &writeSet);
rc = IndustrialSocketSelect(fd + 1, NULL, &writeSet, NULL, &tv);
}
case INDUSTRIAL_SOCKET_IN: {
fd_set readSet;
FD_ZERO(&readSet);
FD_SET(fd, &readSet);
rc = IndustrialSocketSelect(fd + 1, &readSet, NULL, NULL, &tv);
}
default:
break;
}
return rc;
}