udpstack.cxx

RTP协议

    struct sockaddr dummyAddr;

    memset((char*) &dummyAddr, 0, sizeof(dummyAddr));
    dummyAddr.sa_family = NetworkConfig::instance().getAddrFamily();

    int result;

    if ((result = connect(data->socketFd,
                          (struct sockaddr*) & dummyAddr,
                          sizeof(dummyAddr))) != 0)
    {
        int err = errno;
        strstream errMsg;
        errMsg << "UdpStack<" << getLclName() << " " << getRmtName()
        << ">::UdpStack error during socket connect: ";
        errMsg << strerror(err);
        errMsg << char(0);

        cpLog(LOG_ERR,  errMsg.str());
        errMsg.freeze(false);
#if 0
        throw UdpStackException(errMsg.str());
        assert(0);
#endif
    }

    dummyAddr.sa_family = AF_UNSPEC;
    if ((result = connect(data->socketFd,
                          (struct sockaddr*) & dummyAddr,
                          sizeof(dummyAddr))) != 0)
    {
        int err = errno;
        strstream errMsg;
        errMsg << "UdpStack<" << getLclName() << " " << getRmtName()
        << ">::UdpStack error during socket connect: ";
        errMsg << strerror(err);
        errMsg << char(0);

        cpLog(LOG_ERR, errMsg.str());
        errMsg.freeze(false);
#if 0
        throw UdpStackException(errMsg.str());
        assert(0);
#endif
    }
}

/// set the local ports
/// The first time change from inactive to recvonly/sendrecv or
/// change the local ports, this method needs to be called.
void
UdpStack::setLocal (const int minPort, int maxPort )
{
    cpLog (LOG_DEBUG_STACK, "UdpStack::setLocal");
    cpLog (LOG_DEBUG_STACK, "minPort = %d, maxPort = %d", minPort, maxPort);

    if ((mode == sendonly) || (mode == inactive))
    {
        cpLog(LOG_ERR, "The UdpStack is sendonly or inactive.");
        return ;
    }
    // To reopen a new socket, since after sendto(), bind() will fail
    //    if (localPort != -1)
    {
        int newFd;
        newFd = socket(NetworkConfig::instance().getAddrFamily(), 
                       SOCK_DGRAM, IPPROTO_UDP);
        if(NetworkConfig::instance().getAddrFamily() == AF_INET6)
        {
            //Set the sockoption so that we get get source IP
            //when running on the same host
            int on=1;
            setsockopt(data->socketFd, IPPROTO_IPV6, IPV6_PKTINFO, &on, sizeof(on));
        }
#ifndef WIN32
        if ( close (data->socketFd) != 0 )
#else
	if ( closesocket(data->socketFd) )
#endif
	{
            cpLog(LOG_ERR, "close socketFd error!");
	}
        data->socketFd = newFd;
        if ( data->socketFd < 0 )
        {
            int err = errno;
            strstream errMsg;
            errMsg << "UdpStack<" /* << getLclName() */
            << ">::UdpStack error during socket creation: ";
            errMsg << strerror(err);
            errMsg << char(0);

            cpLog(LOG_ERR,  errMsg.str());
            throw UdpStackException(errMsg.str());
            errMsg.freeze(false);
            assert(0);
        }
#ifdef __linux__
        int buf1 = 1;
        int len1 = sizeof(buf1);
        struct protoent * protoent;
        protoent = getprotobyname("icmp");

        if (!protoent)
        {
            fprintf(stderr, "Cannot get icmp protocol\n");
        }
        else
        {
            if (setsockopt(data->socketFd, protoent->p_proto, SO_BSDCOMPAT,
                           (char*)&buf1, len1)
                    == -1)
            {
                fprintf(stderr, "setsockopt error SO_BSDCOMPAT :%s",
                        strerror(errno));
            }
        }
#endif


    }
    doServer(minPort, maxPort);
}

/// set the default destination
void
UdpStack::setDestination ( const NetworkAddress* host )
{
    if ((mode == recvonly) || (mode == inactive))
    {
        cpLog(LOG_ERR, "The UdpStack is recvonly or inactive.");
        return ;
    }
    doClient(host);
}

void
UdpStack::setDestination ( const char* host, int port )
{
    assert(host);
    NetworkAddress netAddress;

    if ( host )
    {
        string rHostName = host;
        netAddress.setPort(port);
        netAddress.setHostName(rHostName);
    }

    setDestination (&netAddress);
}

int
UdpStack::getRxPort() {
        	cpLog(LOG_DEBUG_STACK, "getRxPort()");
		char service[6];
		if(getnameinfo((struct sockaddr *)data->localAddr, sizeof(*data->localAddr), NULL, 0, service, sizeof(service), NI_NUMERICSERV | NI_NUMERICHOST))
			perror("getnameinfo");
		return atoi(service);
}

int
UdpStack::getTxPort() {
        	cpLog(LOG_DEBUG_STACK, "getTxPort()");
		char service[6];
		if(getnameinfo((struct sockaddr *)data->remoteAddr, sizeof(*data->remoteAddr), NULL, 0, service, sizeof(service), NI_NUMERICSERV | NI_NUMERICHOST))
			perror("getnameinfo");
		return atoi(service);
        };


/// need to delete the ne wobject after the object is used
NetworkAddress *
UdpStack::getDestinationHost () const
{
	char host[256];
	char port[6];

    if(getnameinfo((struct sockaddr *)data->remoteAddr, sizeof(*data->remoteAddr), host, sizeof(host), port, sizeof(port), NI_NUMERICSERV | NI_NUMERICHOST))
	    perror("getnameinfo");

    NetworkAddress* desHost = new NetworkAddress(host, atoi(port)) ;
    return desHost;

}


int
UdpStack::getSocketFD ()
{
    assert(data);
    return data->socketFd;
}

void
UdpStack::addToFdSet ( fd_set* set )
{
    assert(set);
    FD_SET(data->socketFd, set);
}


int
UdpStack::getMaxFD ( int prevMax )
{
    return ( getSocketFD() > prevMax ) ? getSocketFD() : prevMax;
}


bool
UdpStack::checkIfSet ( fd_set* set )
{
    assert(set);
    return ( FD_ISSET(data->socketFd, set) ? true : false );
}


int
UdpStack::receive ( const char* buf, const int bufSize )
{
    if ((mode == sendonly) || (mode == inactive))
    {
        cpLog(LOG_ERR, "The stack is not in a state capable of receiving.");
        return -1;
    }

    int len = recv( data->socketFd,
                    (char *)buf, bufSize,
                    0 /*flags */);
    if ( len < 0 )
    {
        int err = errno;
	cpLog(LOG_ERR, "UdpStack: receive error: %s", strerror(err));
#if 0
        throw UdpStackException(errMsg.str());
        assert(0);
#endif
        //        cpLog(LOG_DEBUG_STACK, "UdpStack::receive pkt len=0");
    }
    else if (len == 0)
    {
	// received no bytes
	cpLog(LOG_DEBUG, "did not receive any data");
    }
    else
    {
        numBytesReceived += len;
        numPacketsReceived += 1;
    }

    if ( (logFlag) && (len > 0) )
    {
        strstream lenln1;
        lenln1 << ++rcvCount << " " << len << "\n" << char(0);
        in_log->write(lenln1.str(), strlen(lenln1.str()));
        in_log->write(buf, len);
        in_log->write(separator, 6);
        lenln1.freeze(false);
    }

    return len;
}


int
UdpStack::receiveFrom ( const char* buffer,
                        const int bufSize,
                        NetworkAddress* sender ) // returns bytes read
{
    if ((mode == sendonly) || (mode == inactive))
    {
        cpLog(LOG_ERR, "The stack is not capable to receive. ");
        return -1;
    }

    struct sockaddr_storage xSrc;

    int srcLen = sizeof(xSrc);
    if(sender) sender->getSockAddr(&xSrc);

#if defined(WIN32)

    /* WSAECONNRESET (10054) Connection reset by peer.
     * A existing connection was forcibly closed by the remote host. This
     * normally results if the peer application on the remote host is suddenly
     * stopped, the host is rebooted, or the remote host used a "hard close" (see
     * setsockopt for more information on the SO_LINGER option on the remote
     * socket.) This error may also result if a connection was broken due to
     * "keep-alive" activity detecting a failure while one or more operations are
     * in progress. Operations that were in progress fail with WSAENETRESET.
     * Subsequent operations fail with WSAECONNRESET. Fix suggested by Alexandr Kuzmin [e-mail].
     */

    int len = 0;
    do
    {
        len = recvfrom( data->socketFd,
                            (char *)buffer,
                            bufSize,
                            0 /*flags */ ,
                            (sockaddr*) &xSrc,
                            (socklen_t*) &srcLen);
    } while( (len == -1) && (WSAGetLastError() == WSAECONNRESET ) );
#else
    int len = 0;
    bool ipV6 = false;
    struct in6_pktinfo pktinfo;
    if(NetworkConfig::instance().getAddrFamily() == AF_INET)
    {
        len = recvfrom( data->socketFd,
                        (char *)buffer,
                        bufSize,
                        0 /*flags */ ,
                        (struct sockaddr*) &xSrc,
                        (socklen_t*) &srcLen);
    }
    else
    {
        ipV6 = true;
        int flags=0;
        len = recvfrom_flags( data->socketFd,
                        (char *)buffer,
                        bufSize,
                        &flags /*flags */ ,
                        (struct sockaddr*) &xSrc,
                        (socklen_t*) &srcLen,
                        &pktinfo);
    }

#endif
    if ( len <= 0 )
    {
        int err = errno;
        strstream errMsg;
        errMsg << "UdpStack<" << getLclName() << ">::receive error : ";
        errMsg << strerror(err);
        errMsg << char(0);

        cpLog(LOG_ERR, "%s", errMsg.str());
        errMsg.freeze(false);
    }
    else
    {
        if(sender)
        {
            struct sockaddr_storage xSrcT;
	    sender->getSockAddr(&xSrcT);
            if(memcmp(&xSrc, &xSrcT, srcLen)  == 0)
	    {
		//No need to set anything, return
                numBytesReceived += len;
                numPacketsReceived += 1;
		return len;
	    }
            //Do the heavy-weight work only if the caller shows interest
	    char hostname[256];
	    char port[64];
            hostname[0] = '\0';
            port[0] = '\0';
	    int err = getnameinfo((struct sockaddr *)&xSrc, srcLen, hostname, 256, port, 64, NI_NUMERICHOST | NI_NUMERICSERV);
	    if (err) {
                cpLog(LOG_ERR, "Failed to get the host name");
	    }

            string tmp = hostname;
            if(ipV6)
            {
                struct sockaddr_in6* sin6 = (struct sockaddr_in6*)&xSrc;
                if(IN6_IS_ADDR_LOOPBACK((struct in6_addr*)&sin6->sin6_addr) &&
                   (inet_ntop(AF_INET6, &pktinfo.ipi6_addr, hostname, 256)))
                {
	            tmp = "[";
	            tmp += hostname;
	            tmp += "]";
                }
                else
                {
	            tmp = "[";
	            tmp += hostname;
	            tmp += "]";
                }
            }
            cpLog(LOG_DEBUG, "***Received from:%s:%s", tmp.c_str(), port);
            sender->setPort(atoi(port));
            if((sender->getIpName() != Data(hostname)))
            {
                sender->setHostName(tmp.c_str());
            }
        }
        numBytesReceived += len;
        numPacketsReceived += 1;
    }

    if ( (logFlag) && (len > 0) )
    {
        strstream lenln2;
        lenln2 << ++rcvCount << " " << len << "\n" << char(0);
        in_log->write(lenln2.str(), strlen(lenln2.str()));
        in_log->write(buffer, len);
        in_log->write(separator, 6);
        lenln2.freeze(false);
    }

    return len;
}

int
UdpStack::receiveTimeout ( const char* buffer,
                           const int bufSize,
                           NetworkAddress* sender,
                           int sec,
                           int usec)
{
#ifndef __vxworks
    timeval tv;
    fd_set rset;
    int fd = getSocketFD();

// Make the coket non-blocking and then change it againb after the receivfrom
#ifndef WIN32
    int retVal;
    bool madeNonBlocking = false;
    if(blockingFlg)
    {
        madeNonBlocking = true;
        if(setModeBlocking(false) < 0)
            return -1;
    }
#else
    int retVal;
    unsigned long non_blocking = 1;
    if (ioctlsocket(fd, FIONBIO, &non_blocking))
        return -1;
#endif

    // select will return upon timeout, error or received message
    FD_ZERO(&rset);
    FD_SET(fd, &rset);
    tv.tv_sec = sec;
    tv.tv_usec = usec;
    retVal = select(fd + 1, &rset, NULL, NULL, &tv);
    // we don't care about no stinking error
    if (retVal <= 0)
        return retVal;


    retVal = receiveFrom( buffer,
                          bufSize,
                          sender);

#ifndef WIN32
    if(madeNonBlocking)
    {
        if(setModeBlocking(true) < 0);
            return -1;
    }
#else
	non_blocking = 0;
	if (ioctlsocket(fd, FIONBIO, &non_blocking) != 0)
	return -1;
#endif

    return retVal;
#else

    cpLog(LOG_ERR, "UdpStack::receiveTimeout  * not defined in vxworks *\n");
    return -1;
#endif
}

// uses send() which is better to get ICMP msg back
// function returns a 0  normally
void
UdpStack::transmit ( const char* buf, const int length )
{
    if ((mode == recvonly) || (mode == inactive))
    {
        cpLog(LOG_ERR, "The stack is not capable to transmit. ");
        return ;
    }

    assert(buf);
    assert(length > 0);

    if ( packetLossProbability > 0.0 )
    {
        static bool randInit = false;
        if (!randInit)
        {
            randInit = true;

            timeval tv;
            gettimeofday(&tv, NULL);

            long seed = tv.tv_sec + tv.tv_usec;

            srandom(seed);
        }

        double numerator( random() & 0x7FFFFFFF );
        double denominator( 0x7FFFFFFF );
        double prob = numerator / denominator;
        if ( prob < packetLossProbability )
        {
            // ok - just drop this packet
            return ;
        }
    }

    int count = send( data->socketFd,
                      (char *)buf, length,
                      0 /* flags */ );

    if ( count < 0 )
    {
        int err = errno;
        strstream errMsg;
        errMsg << "UdpStack<" << getRmtName() << ">::transmit ";

        switch (err)
        {
            case ECONNREFUSED:
            {
                // This is the most common error - you get it if the host
                // does not exist or is nor running a program to recevice
                // the packets. This is what you get with the other side
                // crashes.

                errMsg << "Connection refused by destination host";
                errMsg << char(0);
                cpLog(LOG_ERR, errMsg.str());

#if 0
                throw UdpStackExceptionConectionRefused(errMsg.str());
#endif
            }
            break;
            case EHOSTDOWN:
            {
                errMsg << "destination host is down";
                errMsg << char(0);
                cpLog(LOG_ERR, errMsg.str());

            }
            break;
            case EHOSTUNREACH:
            {