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>
#define nonblockingsocket(s) {unsigned long ctl = 1;}
#elif _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#pragma comment(lib,"ws2_32.lib")
#define nonblockingsocket(s) {unsigned long ctl = 1;ioctlsocket( s, FIONBIO, &ctl );}
#endif
 
#include <string>
#include <functional>
#include <cerrno>
#include <thread>
#include <iostream>
 
#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
 
 
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);
    }  
 
 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:
 // 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);
    }
 
 // 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;
    }
 
 
};