apps/astreamer/socket.cpp

Go to the documentation of this file.

#include <string>
#include <iostream>

#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <unistd.h>
#include <sys/poll.h>
#include <errno.h>
#include <fcntl.h>
#include <string.h>

using namespace std;

#include "socket.h"
#include "helpers.h"

#undef RTSP_DEBUG

#ifdef RTSP_DEBUG
        #define D(a) a
#else
        #define D(a)
#endif

#define BUF_SIZE        2048

TCP_Client::TCP_Client(bool enable_exception) {
        ex = enable_exception;
        fd = -1;
}

TCP_Client::~TCP_Client() {
        disconnect();
}

void TCP_Client::connect(string ip, string port) {
        if((fd = ::socket(PF_INET, SOCK_STREAM, 0)) == -1) {
                cerr << "fail socket()" << endl;
                throw(false);
        }

        struct sockaddr_in addr;
        memset(&addr, 0, sizeof(struct sockaddr_in));
        addr.sin_family = AF_INET;
        addr.sin_port = htons(::atol(port.c_str()));
        addr.sin_addr.s_addr = ::inet_addr(ip.c_str());

        if(::connect(fd, (struct sockaddr *)&addr, sizeof(struct sockaddr)) == -1) {
                cerr << "fail connect() to " << ip << ":" << port << endl;
                throw(false);
        }
        // TODO: we must use size as MTU for interface... and also - checking for not single frames...
        buf = (char *)malloc(1500);
}

void TCP_Client::disconnect(void) {
        if(fd > 0) {
                ::close(fd);
                free(buf);
                buf = NULL;
        }
        fd = -1;
}

void TCP_Client::send(const string *data, int opt) {
        if(::send(fd, data->c_str(), data->length(), opt) < 0) {
                cerr << "fail to send(const string *)" << endl;
                throw(false);
        }
}

void TCP_Client::send(const string data, int opt) {
        int er;
        if(::send(fd, data.c_str(), data.length(), opt) < 0) {
                er = errno;
                cerr << "fail to send(const string): " << strerror(er) << endl;
                throw(false);
        }
}

void TCP_Client::send(void *data, int len, int opt) {
        if(::send(fd, data, len, opt) < 0) {
                cerr << "fail to send(void *, len)" << endl;
                throw(false);
        }
}

void TCP_Client::recv(string &r, int opt) {
        int l, er;
        if((l = ::recv(fd, buf, BUF_SIZE - 1, opt)) > 0) {
                buf[l] = '\0';
                r = buf;
        } else {
                er = errno;
                cerr << "fail to recv(string &): " << strerror(er) << endl;
                throw (false);
        }
}

/*
 * Socket class...
 */

unsigned long Socket::ip_to_number(string ip) {
        string left;
        string right;
        unsigned long rez = 0;
        while(String::split(ip, '.', left, right)) {
                rez += atol(left.c_str());
                rez <<= 8;
                ip = right;
        }
        return rez;
}

bool Socket::_is_multicast(string ip) {
        bool rez = true;
        unsigned long a_min = ip_to_number("224.0.0.0");
        unsigned long a_max = ip_to_number("239.255.255.255");
        if(a_min > a_max) {
                unsigned long a = a_min;
                a_min = a_max;
                a_max = a;
        }
        unsigned long a_ip = ip_to_number(ip.c_str());
        if(a_ip > a_max)
                rez = false;
        if(a_ip < a_min)
                rez = false;
        return rez;
}

Socket::Socket(string _ip, int _port, stype _type) {
        ip = _ip;
        port = _port;
        is_multicast = false;

D(cerr << "new socket..." << endl;)
        _state = STATE_EMPTY;

        int t = 0;
        type = _type;

        struct sockaddr_in addr;
        memset(&addr, 0, sizeof(struct sockaddr_in));
        addr.sin_family = PF_INET;
        addr.sin_port = htons(port);

        switch(type) {
        case TYPE_TCP: {
                fd = socket(PF_INET, SOCK_STREAM, 0);
                int flag = 1;
                setsockopt(fd, IPPROTO_IP, SO_REUSEADDR, &flag, sizeof(flag));
                if(ip != "")
                        addr.sin_addr.s_addr = inet_addr(ip.c_str());
                else
                        addr.sin_addr.s_addr = htonl(INADDR_ANY);
                t = bind(fd, (struct sockaddr *)&addr, sizeof(struct sockaddr));
D(              cerr << "TCP ::bind() == " << t << endl;)
                break;
        }
        case TYPE_UDP: {
                fd = socket(PF_INET, SOCK_DGRAM, 0);
                struct sockaddr_in saddr;
                memset(&saddr, 0, sizeof(struct sockaddr_in));
                saddr.sin_family = PF_INET;
                saddr.sin_port = htons(0);
                saddr.sin_addr.s_addr = htonl(INADDR_ANY);
                ::bind(fd, (struct sockaddr *)&saddr, sizeof(struct sockaddr_in));
                addr.sin_addr.s_addr = inet_addr(ip.c_str());
                t = ::connect(fd, (struct sockaddr *)&addr, sizeof(struct sockaddr));
D(              cerr << "::connect() == " << t << endl;)
                if((is_multicast = _is_multicast(ip)) == true) {
                        struct ip_mreqn multiaddr;
                        multiaddr.imr_multiaddr.s_addr = inet_addr(ip.c_str());
                        multiaddr.imr_address.s_addr = htonl(INADDR_ANY);
                        multiaddr.imr_ifindex = 0;
                        setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &multiaddr, sizeof(struct sockaddr_in));
                }
                }
                break;
//      }
        }
}

Socket::Socket(int _fd, string _ip, int _port, stype _t) {
        _state = STATE_EMPTY;

        fd =_fd;
        ip = _ip;
        port = _port;
        type = _t;
D(      cerr << "socket: ip == " << ip << "; port == " << port << endl;)
}

Socket::~Socket() {
        close(fd);
}

int Socket::poll(list<Socket *> &s) {
        struct pollfd *pfd;
        ssize_t s_size = s.size();

D(cerr << "Socket::poll()..." << endl;)
        pfd = (struct pollfd *)malloc(sizeof(struct pollfd) * s_size);
        memset(pfd, 0, sizeof(struct pollfd) * s_size);
        int i = 0;
D(cerr << "socket.fd == ";)
        for(list<Socket *>::iterator it = s.begin(); it != s.end(); it++, i++) {
                pfd[i].fd = (*it)->fd;
D(cerr << pfd[i].fd << "; ";)
//              pfd[i].events = 0xFFFF;
                pfd[i].events = POLLIN | POLLHUP;
                pfd[i].revents = 0x00;
        }
D(cerr << endl;)
        int p = ::poll(pfd, s_size, -1);
        i = 0;
        for(list<Socket *>::iterator it = s.begin(); it != s.end(); it++, i++) {
                (*it)->_state = STATE_EMPTY;
D(cerr << "revents == " << pfd[i].revents << "; POLLIN == " << POLLIN << endl;)
                if(pfd[i].revents & POLLIN) {
                        (*it)->_state = STATE_IN;
D(cerr << "STATE_IN; fd == " << (*it)->fd << "; revents == " << pfd[i].revents << endl;)
                } 
                if(pfd[i].revents & POLLHUP) {
                        (*it)->_state = STATE_DISCONNECT;
D(cerr << "STATE_DISCONNECT; fd == " << (*it)->fd << "; revents == " << pfd[i].revents << endl;)
                } 
        }
        free((void *)pfd);
        return p;
}

void Socket::listen(int in) {
        long sock_flags;

        sock_flags = fcntl(fd, F_GETFL);
        sock_flags |= O_NONBLOCK;
        fcntl(fd, F_SETFL, sock_flags);
	::listen(fd, in);
}

Socket *Socket::accept(void) {
        int _fd;
        struct sockaddr_in addr;
        socklen_t addr_len = 0;
        long sock_flags;
        Socket *s;

        memset(&addr, 0, sizeof(struct sockaddr_in));
        addr_len = sizeof(struct sockaddr_in);
        _fd = ::accept(fd, (struct sockaddr *)&addr, &addr_len);
        if(_fd < 0)
                return NULL;
        sock_flags = fcntl(_fd, F_GETFL);
        sock_flags |= O_NONBLOCK;
        fcntl(_fd, F_SETFL, sock_flags);
D(cerr << "accept == " << _fd << "; addr_len == " << addr_len << endl;)
        s = new Socket(_fd, inet_ntoa(addr.sin_addr), ntohs(addr.sin_port), type);
        return s;
}

/*
bool Socket::connect(void) {
        struct ip_mreqn multiaddr;
        struct sockaddr_in ma;

        rtp_sock = socket(PF_INET, SOCK_DGRAM, 0);
        multiaddr.imr_address.s_addr = htonl(INADDR_ANY);
        memcpy(&multiaddr.imr_multiaddr.s_addr, multicast_ip, 4);
        multiaddr.imr_ifindex = 0;
        if(setsockopt(rtp_sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &multiaddr, sizeof(struct sockaddr_in)) != 0)
                return false;
        rtp_sock = socket(PF_INET, SOCK_DGRAM, 0);
        ma.sin_family = AF_INET;
        ma.sin_addr.s_addr = INADDR_ANY;
        ma.sin_port = htons(RTP_MCAST_SERVER_PORT);
        if(bind(rtp_sock, (const struct sockaddr*)&ma, sizeof(struct sockaddr_in)) != 0)
                return false;
        ma.sin_family = AF_INET;
        memcpy(&ma.sin_addr.s_addr, multicast_ip, 4);
        ma.sin_port = htons(RTP_MCAST_CLIENT_PORT);
        if(connect(rtp_sock, (const struct sockaddr*)&ma, sizeof(struct sockaddr_in)) != 0)
                return false;
        return true;
}
*/

bool Socket::recv(string &rez) {
        rez = "???";
        char buf[1501];
        int l = 0;
        buf[0] = '\0';

        if((l = ::recv(fd, buf, 1500, MSG_NOSIGNAL | MSG_DONTWAIT)) > 0) {
                buf[l] = '\0';
                rez = buf;
D(cerr << "read - ok!"<< endl;)
                return true;
        } else if(l == 0) {
                rez = "";
                return true;
        }
        rez = "";
        return false;
}

bool Socket::send(void *data, int len) {
        return (::send(fd, data, len, MSG_NOSIGNAL | MSG_DONTWAIT) > 0);
}

bool Socket::send(const string *data) {
        if(::send(fd, data->c_str(), data->length(), 0) > 0)
                return true;
        return false;
}

bool Socket::send(const string data) {
        if(::send(fd, data.c_str(), data.length(), MSG_NOSIGNAL | MSG_DONTWAIT) > 0)
                return true;
        return false;
}

uint16_t in_chksum(const uint16_t *addr, register uint32_t len) {
        int32_t nleft = len;
        const uint16_t *w = addr;
        uint16_t answer;
        int32_t sum = 0;

        while(nleft > 1)  {
                sum += *w++;
                nleft -= 2;
        }
        if(nleft == 1)
                sum += htons(*(uint8_t  *)w << 8);
        sum = (sum >> 16) + (sum & 0xffff);     /* add hi 16 to low 16 */
        sum += (sum >> 16);                     /* add carry */
        answer = ~sum;                          /* truncate to 16 bits */
        return answer;
}

bool Socket::send2v(void **v_ptr, int *v_len) {
        struct iovec iov[2];
        int iovcnt = sizeof(iov) / sizeof(struct iovec);
        iov[0].iov_base = v_ptr[0];
        iov[0].iov_len = v_len[0];
        iov[1].iov_base = v_ptr[1];
        iov[1].iov_len = v_len[1];
        if(::writev(fd, iov, iovcnt))
                return true;
        return false;
}

bool Socket::send3v(void **v_ptr, int *v_len) {
        struct iovec iov[3];
        int iovcnt = sizeof(iov) / sizeof(struct iovec);
        iov[0].iov_base = v_ptr[0];
        iov[0].iov_len = v_len[0];
        iov[1].iov_base = v_ptr[1];
        iov[1].iov_len = v_len[1];
        iov[2].iov_base = v_ptr[2];
        iov[2].iov_len = v_len[2];
        if(::writev(fd, iov, iovcnt))
                return true;
        return false;
}

---