rfly_udp.h

#pragma once
 
#if defined(__linux__) || defined(__APPLE__)
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <unistd.h>
#include <netdb.h>
#include <fcntl.h>
#include <ifaddrs.h>
#define nonblockingsocket(s) {unsigned long ctl = 1;}
#elif _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#pragma comment(lib,"ws2_32.lib")
#include <iphlpapi.h>
#pragma comment(lib, "iphlpapi.lib")
#define nonblockingsocket(s) {unsigned long ctl = 1;ioctlsocket( s, FIONBIO, &ctl );}
#endif
 
#include <string>
#include <functional>
#include <cerrno>
#include <thread>
#include <iostream>
#include <algorithm>
#include <cstring>
#include <stdexcept>
#include <vector>
 
#define FDR_UNUSED(expr){ (void)(expr); } 
#define FDR_ON_ERROR std::function<void(int, std::string)> onError = [](int errorCode, std::string errorMessage){FDR_UNUSED(errorCode); FDR_UNUSED(errorMessage)}
 
#ifndef BUFFER_SIZE_RFLY
#define BUFFER_SIZE_RFLY 1024
#endif
 
// ===== 序列化工具 =====
 
// 计算序列化后的大小
template<typename T>
size_t calculateSize(const T& obj);
 
// 基本数据类型的大小计算
template<typename T>
inline size_t calculateSize(const T& value) {
 static_assert(std::is_trivially_copyable<T>::value, 
 "Type must be trivially copyable for serialization");
 return sizeof(T);
}
 
// std::string的大小计算
inline size_t calculateSize(const std::string& str) {
 return sizeof(uint32_t) + str.size();
}
 
// 普通数组的大小计算
template<typename T, size_t N>
inline size_t calculateSize(const T (&array)[N]) {
 size_t total = 0;
 for (size_t i = 0; i < N; ++i) {
        total += calculateSize(array[i]);
    }
 return total;
}
 
// std::vector的大小计算
template<typename T>
inline size_t calculateSize(const std::vector<T>& vec) {
 size_t total = sizeof(uint32_t);  // 存储vector大小
 for (const auto& element : vec) {
        total += calculateSize(element);
    }
 return total;
}
 
// ===== 序列化到缓冲区 =====
 
template<typename T>
void serializeToBuffer(uint8_t* buffer, const T& value);
 
// 基本数据类型的序列化
template<typename T>
inline void serializeToBuffer(uint8_t* buffer, const T& value) {
 static_assert(std::is_trivially_copyable<T>::value, 
 "Type must be trivially copyable for serialization");
    memcpy(buffer, &value, sizeof(value));
}
 
// std::string的序列化
inline void serializeToBuffer(uint8_t* buffer, const std::string& str) {
    uint32_t length = static_cast<uint32_t>(str.size());
    memcpy(buffer, &length, sizeof(length));
    buffer += sizeof(length);
    memcpy(buffer, str.data(), length);
}
 
// 普通数组的序列化
template<typename T, size_t N>
inline void serializeToBuffer(uint8_t* buffer, const T (&array)[N]) {
 for (size_t i = 0; i < N; ++i) {
        serializeToBuffer(buffer, array[i]);
        buffer += calculateSize(array[i]);
    }
}
 
// std::vector的序列化
template<typename T>
inline void serializeToBuffer(uint8_t* buffer, const std::vector<T>& vec) {
    uint32_t size = static_cast<uint32_t>(vec.size());
    memcpy(buffer, &size, sizeof(size));
    buffer += sizeof(size);
 
 for (const auto& element : vec) {
        serializeToBuffer(buffer, element);
        buffer += calculateSize(element);
    }
}
 
// ===== 从缓冲区反序列化 =====
 
template<typename T>
size_t deserializeFromBuffer(const uint8_t* buffer, T& value);
 
// 基本数据类型的反序列化
template<typename T>
inline size_t deserializeFromBuffer(const uint8_t* buffer, T& value) {
 static_assert(std::is_trivially_copyable<T>::value, 
 "Type must be trivially copyable for deserialization");
    memcpy(&value, buffer, sizeof(value));
 return sizeof(value);
}
 
// std::string的反序列化
inline size_t deserializeFromBuffer(const uint8_t* buffer, std::string& str) {
    uint32_t length;
    memcpy(&length, buffer, sizeof(length));
    buffer += sizeof(length);
 
    str.assign(reinterpret_cast<const char*>(buffer), length);
 return sizeof(length) + length;
}
 
// 普通数组的反序列化
template<typename T, size_t N>
inline size_t deserializeFromBuffer(const uint8_t* buffer, T (&array)[N]) {
 size_t total = 0;
 for (size_t i = 0; i < N; ++i) {
        total += deserializeFromBuffer(buffer + total, array[i]);
    }
 return total;
}
 
// std::vector的反序列化
template<typename T>
inline size_t deserializeFromBuffer(const uint8_t* buffer, std::vector<T>& vec) {
    uint32_t size;
    memcpy(&size, buffer, sizeof(size));
    buffer += sizeof(size);
 
    vec.resize(size);
 size_t total = sizeof(size);
 
 for (size_t i = 0; i < size; ++i) {
        total += deserializeFromBuffer(buffer + total - sizeof(size), vec[i]);
    }
 return total;
}
 
// ===== 打包/解包函数 =====
 
// 打包函数 - 返回堆上分配的字节流和长度
template<typename T>
size_t packStruct(const T& obj, uint8_t* buffer) {
 // 检查缓冲区是否有效
 if (buffer == nullptr) {
 throw std::invalid_argument("buffer cannot be null");
    }
 // 计算所需大小（用于返回，不分配内存）
 size_t size = calculateSize(obj);
 // 直接在传入的缓冲区中序列化
    serializeToBuffer(buffer, obj);
 // 返回实际序列化的大小（与计算的size一致）
 return size;
}
 
// 解包函数 - 从字节流恢复结构体
template<typename T>
size_t unpackStruct(const uint8_t* buffer, T& obj) {
 return deserializeFromBuffer(buffer, obj);
}
 
struct GenericInOut28 {
    int32_t ioSilInts[8];   // 非静态成员：整数数组
 float ioSilFloats[20];  // 非静态成员：浮点数数组
 
 // 构造函数：初始化全部为0
    GenericInOut28() {
        memset(this, 0, sizeof(*this));
    }
};
 
// 手动实现GenericInOut28的序列化/反序列化（不依赖宏）
// 1. 计算大小
inline size_t calculateSize(const GenericInOut28& obj) {
 // 分别计算两个成员的大小并累加
 return calculateSize(obj.ioSilInts) + calculateSize(obj.ioSilFloats);
}
 
// 2. 序列化到缓冲区
inline void serializeToBuffer(uint8_t* buffer, const GenericInOut28& obj) {
 // 先序列化整数数组，再移动指针
    serializeToBuffer(buffer, obj.ioSilInts);
    buffer += calculateSize(obj.ioSilInts);  // 移动指针到整数数组末尾
 
 // 再序列化浮点数数组
    serializeToBuffer(buffer, obj.ioSilFloats);
 // 指针无需再移动（后续操作由调用者处理）
}
 
// 3. 从缓冲区反序列化
inline size_t deserializeFromBuffer(const uint8_t* buffer, GenericInOut28& obj) {
 const uint8_t* start = buffer;  // 记录起始位置
 
 // 先反序列化整数数组，移动指针
    buffer += deserializeFromBuffer(buffer, obj.ioSilInts);
 
 // 再反序列化浮点数数组，移动指针
    buffer += deserializeFromBuffer(buffer, obj.ioSilFloats);
 
 // 返回总消耗的字节数
 return buffer - start;
}
 
 
class BaseSocket
{
public:
 enum SocketType
    {
        TCP = SOCK_STREAM,
        UDP = SOCK_DGRAM
    };
    sockaddr_in address;
 //char udpBuf[BUFFER_SIZE_RFLY+1];
 //std::string RecvIP;
 //uint16_t RecvPort;
    std::string remoteAddress() const { return ipToString(this->address); }
 int remotePort() const { return ntohs(this->address.sin_port); }
 int fileDescriptor() const { return this->sock; }
 
protected:
 int sock = 0;
 
 bool hasSetNoblock=false;
 
 
 
 //bool isStopRec=false;
 
 // Get std::string value of the IP from a `sockaddr_in` address struct
 static std::string ipToString(const sockaddr_in& addr)
    {
 char ip[INET_ADDRSTRLEN];
 
 struct sockaddr_in in;
        memset(&in, 0, sizeof(in));
        in.sin_family = AF_INET;
        memcpy(&in.sin_addr, &(addr.sin_addr), sizeof(struct in_addr));
        getnameinfo((struct sockaddr *)&in, sizeof(struct sockaddr_in), ip, INET_ADDRSTRLEN, NULL, 0, NI_NUMERICHOST);
 
 return std::string(ip);
    }
 
 static bool isVirtualMachineAdapter(const std::string& adapterName) {
        std::vector<std::string> vmKeywords = {
 "VMware", "VirtualBox", "Hyper-V", "QEMU"
        };
 for (const auto& keyword : vmKeywords) {
 if (adapterName.find(keyword) != std::string::npos) {
 return true;
            }
        }
 return false;
    }
 
 // 函数用于获取静态 IP 地址，排除虚拟机网络
 static std::vector<std::string> getStaticIPAddresses() {
        std::vector<std::string> staticIPs;
#ifdef _WIN32
        ULONG outBufLen = 0;
 // 首次调用以获取所需的缓冲区大小
 if (GetAdaptersInfo(NULL, &outBufLen) == ERROR_BUFFER_OVERFLOW) {
            PIP_ADAPTER_INFO pAdapterInfo = (PIP_ADAPTER_INFO)malloc(outBufLen);
 if (pAdapterInfo) {
 // 再次调用以获取适配器信息
 if (GetAdaptersInfo(pAdapterInfo, &outBufLen) == NO_ERROR) {
                    PIP_ADAPTER_INFO pAdapter = pAdapterInfo;
 while (pAdapter) {
                        std::string adapterName = pAdapter->Description;
 // 检查是否为虚拟机网络
 if (!isVirtualMachineAdapter(adapterName)) {
 // 检查是否为静态分配的 IP
 if (pAdapter->DhcpEnabled == FALSE) {
                                IP_ADDR_STRING* pIpAddr = &pAdapter->IpAddressList;
 while (pIpAddr) {
                                    staticIPs.emplace_back(pIpAddr->IpAddress.String);
                                    pIpAddr = pIpAddr->Next;
                                }
                            }
                        }
                        pAdapter = pAdapter->Next;
                    }
                }
                free(pAdapterInfo);
            }
        }
 
#endif
 return staticIPs;
    }
 
    BaseSocket(FDR_ON_ERROR, SocketType sockType = TCP, int socketId = -1)
    {
 if (socketId == -1)
        {
            this->sock = (int)socket(AF_INET, sockType, 0);
 
 if ( this->sock == -1 )
            {
                onError(errno, "Socket creating error.");
            }
        }
 else
        {
            this->sock = socketId;
        }
 int rcvBufSize=102400*10;
#ifndef __linux__
        setsockopt(this->sock, SOL_SOCKET, SO_RCVBUF, (char*)&rcvBufSize, sizeof(rcvBufSize));
        rcvBufSize=102400*10;
        setsockopt(this->sock, SOL_SOCKET, SO_SNDBUF, (char*)&rcvBufSize, sizeof(rcvBufSize));
#else
        setsockopt(this->sock, SOL_SOCKET, SO_RCVBUF, &rcvBufSize, sizeof(rcvBufSize));
        rcvBufSize=102400*10;
        setsockopt(this->sock, SOL_SOCKET, SO_SNDBUF, &rcvBufSize, sizeof(rcvBufSize));
#endif
 
 
 
    }
 virtual ~BaseSocket(){}
};
 
 
class UDPSocket : public BaseSocket
{
public:
    std::function<void(std::string, std::string, uint16_t)> onMessageReceived;
    std::function<void(const char*, size_t, std::string, uint16_t)> onRawMessageReceived;
 bool isUseConnect = false;
 //std::thread* mT;
 
 void Close() {
 
#ifndef __linux__
        closesocket(this->sock);
#else
        shutdown(this->sock, SHUT_RDWR);
        close(this->sock);
#endif
    }
 
 
 explicit UDPSocket(bool useConnect = false, FDR_ON_ERROR, int socketId = -1): BaseSocket(onError, SocketType::UDP, socketId)
    {
        this->isUseConnect=useConnect;
 
    }
 
 void StartRecvThread()
    {
 
 if (this->isUseConnect)
        {
            std::thread t(Receive, this); // usage with Connect()
            t.detach();
        }
 else
        {
            std::thread t(ReceiveFrom, this);
            t.detach();
        }
 
 //this->isStopRec=false;
    }
 
 // Send raw bytes to a spesific `host` & `port` with no connection
 int SendTo(const char* bytes, size_t byteslength, const char* host, uint16_t port, FDR_ON_ERROR)
    {
        sockaddr_in hostAddr;
 
 //int status;
 if((hostAddr.sin_addr.s_addr =inet_addr(host))==INADDR_NONE){
           onError(errno, "SendTo: Invalid address. Address type not supported.");
        }
 
 // struct addrinfo hints, *res, *it;
 // memset(&hints, 0, sizeof(hints));
 // hints.ai_family = AF_INET;
 // hints.ai_socktype = SOCK_DGRAM;
 
 // int status;
 // if ((status = getaddrinfo(host, NULL, &hints, &res)) != 0)
 // {
 //     onError(errno, "Invalid address." + std::string(gai_strerror(status)));
 //     return -1;
 // }
 
 // for (it = res; it != NULL; it = it->ai_next)
 // {
 //     if (it->ai_family == AF_INET)
 //     { // IPv4
 //         memcpy((void* )(&hostAddr), (void* )it->ai_addr, sizeof(sockaddr_in));
 //         break; // for now, just get first ip (ipv4).
 //     }
 // }
 
 // freeaddrinfo(res);
 
        hostAddr.sin_port = htons(port);
        hostAddr.sin_family = AF_INET;
 
 int sent_length = sendto(this->sock, bytes, (int)byteslength, 0, (sockaddr*)&hostAddr, sizeof(hostAddr));
 if (sent_length == -1)
        {
            onError(errno, "Cannot send message to the address.");
 return -1;
        }
 
 return sent_length;
    }
 // Send raw bytes to a spesific `host` & `port` with no connection
 int SendTo(const char* bytes, size_t byteslength, const std::string& host, uint16_t port, FDR_ON_ERROR)
    {
 return this->SendTo(bytes, byteslength, host.c_str(), port, onError);
    }
 // Send std::string to a spesific `host` & `port` with no connection
 int SendTo(const std::string& message, const char* host, uint16_t port, FDR_ON_ERROR)
    {
 return this->SendTo(message.c_str(), message.length(), host, port, onError);
    }
 // Send std::string to a spesific `host` & `port` with no connection
 int SendTo(const std::string& message, const std::string& host, uint16_t port, FDR_ON_ERROR)
    {
 return this->SendTo(message.c_str(), message.length(), host.c_str(), port, onError);
    }
 
 // Send raw bytes to the `Connect()`ed server.
 int Send(const char* bytes, size_t byteslength) { return send(this->sock, bytes, (int)byteslength, 0); }
 // Send std::string to the `Connect()`ed server.
 int Send(const std::string& message) { return this->Send(message.c_str(), message.length()); }
 
 // Connect to a server with raw `uint32_t ipv4` and `uint16_t port` values.
 void Connect(uint32_t ipv4, uint16_t port, FDR_ON_ERROR)
    {
        this->address.sin_family = AF_INET;
        this->address.sin_port = htons(port);
        this->address.sin_addr.s_addr = ipv4;
 
 // Try to connect.
 int status = connect(this->sock, (const sockaddr* )&this->address, sizeof(sockaddr_in));
 if (status == -1)
        {
            onError(errno, "Connection failed to the host.");
 return;
        }
    }
 // Connect to a server with `host` address and `port` values.
 void Connect(const char* host, uint16_t port, FDR_ON_ERROR)
    {
 // struct addrinfo hints, *res, *it;
 // memset(&hints, 0, sizeof(hints));
 // hints.ai_family = AF_INET;
 // hints.ai_socktype = SOCK_DGRAM;
 
 // int status = getaddrinfo(host, NULL, &hints, &res);
 // if (status != 0)
 // {
 //     onError(errno, "Invalid address." + std::string(gai_strerror(status)));
 //     return;
 // }
 
 // for (it = res; it != NULL; it = it->ai_next)
 // {
 //     if (it->ai_family == AF_INET)
 //     { // IPv4
 //         memcpy((void* )(&this->address), (void*)it->ai_addr, sizeof(sockaddr_in));
 //         break; // for now, just get first ip (ipv4).
 //     }
 // }
 
 // freeaddrinfo(res);
 
 if((this->address.sin_addr.s_addr =inet_addr(host))==INADDR_NONE){
           onError(errno, "SendTo: Invalid address. Address type not supported.");
        }
        this->Connect((uint32_t)this->address.sin_addr.s_addr, port, onError);
    }
 // Connect to a server with `host` address and `port` values.
 void Connect(const std::string& host, uint16_t port, FDR_ON_ERROR) { this->Connect(host.c_str(), port, onError); }
 
private:
 static void Receive(UDPSocket* udpSocket)
    {
 char tempBuffer[BUFFER_SIZE_RFLY+1];
 size_t messageLength;
 //std::cout<<"Receive thread.."<<std::endl;
 while ((messageLength = recv(udpSocket->sock, tempBuffer, BUFFER_SIZE_RFLY, 0)) != -1)
        {
 //std::cout<<"Receive Msg.."<<std::endl;
            tempBuffer[messageLength] = '\0';
 if (udpSocket->onMessageReceived)
                udpSocket->onMessageReceived(std::string(tempBuffer, messageLength), ipToString(udpSocket->address), ntohs(udpSocket->address.sin_port));
 
 if (udpSocket->onRawMessageReceived)
                udpSocket->onRawMessageReceived(tempBuffer, messageLength, ipToString(udpSocket->address), ntohs(udpSocket->address.sin_port));
        }
    }
 
 static void ReceiveFrom(UDPSocket* udpSocket)
    {
 
        sockaddr_in hostAddr;
        socklen_t hostAddrSize = sizeof(hostAddr);
 
 char tempBuffer[BUFFER_SIZE_RFLY+1];
 size_t messageLength;
 //std::cout<<"ReceiveFrom thread.."<<std::endl;
 while ((messageLength = recvfrom(udpSocket->sock, tempBuffer, BUFFER_SIZE_RFLY, 0, (sockaddr* )&hostAddr, &hostAddrSize)) != -1)
        {
 //std::cout<<"Receive Msg.."<<std::endl;
            tempBuffer[messageLength] = '\0';
 if (udpSocket->onMessageReceived)
                udpSocket->onMessageReceived(std::string(tempBuffer, messageLength), ipToString(hostAddr), ntohs(hostAddr.sin_port));
 
 if (udpSocket->onRawMessageReceived)
                udpSocket->onRawMessageReceived(tempBuffer, messageLength, ipToString(hostAddr), ntohs(hostAddr.sin_port));
 //std::cout<<"ReceiveFrom thread.."<<tempBuffer<<messageLength<<std::endl;
        }
    }
};
 
 
class UDPServer : public UDPSocket
{
public:
 // filter udp data according to ip
 bool needFilter=false;
 // Bind the custom address & port of the server.
 void Bind(const char* addressin, uint16_t port, FDR_ON_ERROR)
    {
 // int status = inet_pton(AF_INET, addressin, &this->address.sin_addr);
 // switch (status) {
 //     case -1:
 //         onError(errno, "Invalid address. Address type not supported.");
 //         return;
 //     case 0:
 //         onError(errno, "AF_INET is not supported. Please send message to developer.");
 //         return;
 //     default:
 //         break;
 // }
 
 
 if((this->address.sin_addr.s_addr =inet_addr(addressin))==INADDR_NONE){
           onError(errno, "Invalid address. Address type not supported.");
        }
 //this->address.sin_addr =inet_addr(addressin);
        this->address.sin_family = AF_INET;
        this->address.sin_port = htons(port);
 
 
 int status;
 //set recv buffer and reuseIP
#ifndef __linux__
 bool bOpt = true;
        status = setsockopt(this->sock, SOL_SOCKET, SO_REUSEADDR, (char*)&bOpt, sizeof(bOpt));
 if (status == -1)
        {
            onError(errno, "Cannot set SO_REUSEADDR the socket.");
 return;
        }
 // int rcvBufSize=602400;
 // status = setsockopt(this->sock, SOL_SOCKET, SO_RCVBUF, (char*)&rcvBufSize, sizeof(rcvBufSize));
 // {
 //     onError(errno, "Cannot set SO_RCVBUF the socket.");
 //     return;
 // }
 
#else
 int flag0 = 1;
        setsockopt(this->sock, SOL_SOCKET, SO_REUSEADDR, &flag0, sizeof(flag0) );
 // int rcvBufSize=602400;
 // setsockopt(this->sock, SOL_SOCKET, SO_RCVBUF, &rcvBufSize, sizeof(rcvBufSize));
#endif
 
        status = bind(this->sock, (const sockaddr*)&this->address, sizeof(this->address));
 if (status == -1)
        {
            onError(errno, "Cannot bind the socket.");
 return;
        }
 
 //this->StartRecvThread();
    }
 
 // Bind the address(0.0.0.0) & port of the server.
 void Bind(uint16_t port, FDR_ON_ERROR)
    {
 //std::cout<<"Bind.."<<std::endl;
        this->Bind("0.0.0.0", port, onError);
    }
 
 // For Sim, Bind 127.0.0.1 for Sim.
 // For Real, if static ip "192.168.151.168" exist, bind it;
 //           else, bind a non "0.0.0.0" static ip;
 //           else, bind "0.0.0.0".
 void BindAuto(bool isReal, uint16_t port, FDR_ON_ERROR)
    {
        std::cout<<"Bind Auto, with ip:\t";
 if (isReal){
            needFilter = true;
            std::vector<std::string> staticIPs = getStaticIPAddresses();
            std::string targetIP = "192.168.151.168";
 auto it = std::find(staticIPs.begin(), staticIPs.end(), targetIP);
 if (it != staticIPs.end()) {
                this->Bind("192.168.151.168", port, onError);
                std::cout<<targetIP<<std::endl;
            } else {
                std::string bindIP = "0.0.0.0";
 for (const auto& ip : staticIPs) {
 if (ip != "0.0.0.0") {
                        bindIP = ip;
 break;
                    }
                }
                this->Bind(bindIP.c_str(), port, onError);
                std::cout<<bindIP<<std::endl;
            }
        } else{
            this->Bind("127.0.0.1", port, onError);
            std::cout<<"127.0.0.1"<<std::endl;
        }
    }
 
 // SET Multicast IP and add to group
 void setMulticast(const char* GROUP_IP, FDR_ON_ERROR)
    {
 
 struct ip_mreq vmreq;
 
 if((vmreq.imr_multiaddr.s_addr =inet_addr(GROUP_IP))==INADDR_NONE){
           onError(errno, "setMulticast: Invalid address. Address type not supported.");
        }
 if((vmreq.imr_interface.s_addr =inet_addr("0.0.0.0"))==INADDR_NONE){
           onError(errno, "setMulticast: Invalid address. Address type not supported.");
        }
 
 // inet_pton(AF_INET, GROUP_IP, &vmreq.imr_multiaddr); 
 // inet_pton(AF_INET, "0.0.0.0", &vmreq.imr_interface);  
 
#ifndef __linux__
 int status = setsockopt(this->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char*)&vmreq, sizeof(vmreq));
#else
 int status = setsockopt(this->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &vmreq, sizeof(vmreq));
#endif
 
 if(status == -1)
        {
            onError(errno, "setsockopt(IP_ADD_MEMBERSHIP) failed.");
 return;
        }        
    }
 
 // Enable or disable the SO_BROADCAST flag
 void setBroadcast(bool value, FDR_ON_ERROR)
    {
 int broadcast = static_cast<int>(value);
#ifndef __linux__
 int status = setsockopt(this->sock, SOL_SOCKET, SO_BROADCAST, (char*)&broadcast, sizeof broadcast);
#else
 int status = setsockopt(this->sock, SOL_SOCKET, SO_BROADCAST, &broadcast, sizeof broadcast);
#endif
 if (status == -1)
        {
            onError(errno, "setsockopt(SO_BROADCAST) failed.");
 return;
        }
    }
 
 int RecvNoblock(char * buf, std::string &ip, int &port,int maxrecvlen=BUFFER_SIZE_RFLY)
    {
 if(!this->hasSetNoblock){
            nonblockingsocket(this->sock);
            this->hasSetNoblock=true;
        }
 int req_len=0;
 int recvlen=0;
        sockaddr_in hostAddr;
        socklen_t hostAddrSize = sizeof(hostAddr);
 
#ifndef __linux__
        recvlen=recvfrom(this->sock, buf, maxrecvlen, 0, (struct sockaddr *)&hostAddr, &hostAddrSize);
#else
        recvlen=recvfrom(this->sock, buf, maxrecvlen, MSG_DONTWAIT, (struct sockaddr *)&hostAddr, (socklen_t *)(&hostAddrSize));
#endif
 if(recvlen>0){
            ip=ipToString(hostAddr);
            port = ntohs(hostAddr.sin_port);
            buf[recvlen] = '\0';
        }        
 return recvlen;
    }
 
 
};