udp.cpp

linux下实现视频播放的播放器

/*
 *  Copyright (C) 2005-2007  gulikoza
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

/* $Id$ */

#include "iptv.h"

#define USES_BASECLASS
#include "video.h"

//#define DEBUG
#include "log.h"

#define MODULE "udpSource"

#ifdef WIN32
#define WINSOCK_STRERROR_SIZE 20
static const char *winsock_strerror(char *buf)
{
    snprintf(buf, WINSOCK_STRERROR_SIZE, "Winsock error %d", WSAGetLastError());
    buf[WINSOCK_STRERROR_SIZE - 1] = '\0';
    return buf;
}

char strerror_buf[WINSOCK_STRERROR_SIZE];
#define close(x) closesocket(x)
#endif

#ifndef IP_ADD_MEMBERSHIP
#define IP_ADD_MEMBERSHIP	5
struct ip_mreq {
    struct in_addr imr_multiaddr;
    struct in_addr imr_interface;
};
#endif

bool udpSource::Open(CStr& s)
{
    CStr address = s;
    struct ip_mreq mreq;
    unsigned short port;
    int i_opt;

    const char * c = address.replace(':', '\0');

    if(!c) {
	ERROR_MSG("No port specified!");
	return false;
    }

    port = atoi(c);
    if(!port) {
	ERROR_MSG("Invalid port specified!");
	return false;
    }

    ERROR_MSG("Connecting to %s, port %d", address.c_str(), port);

#ifdef WIN32
#define strerror(x) winsock_strerror(strerror_buf)
    WSADATA Data;

    /* init winsock */
    if(!WSAStartup(MAKEWORD(2, 0), &Data)) {
	/* Confirm that the WinSock DLL supports 2.0.*/
	if(LOBYTE(Data.wVersion) != 2 || HIBYTE(Data.wVersion) != 0) {
	    /* We could not find a suitable WinSock DLL. */
	    WSACleanup();
	    ERROR_MSG("Error initializing WinSock2!");
	    return false;
	}
    }
#endif

    /* create what looks like an ordinary UDP socket */
    if((desc = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
	ERROR_MSG("Error opening socket: %s", strerror(desc));
	return false;
    }

    /* set up destination address */
    SDL_memset(&addr, 0, sizeof(addr));
    addr.sin_family=AF_INET;
    addr.sin_addr.s_addr=htonl(INADDR_ANY); 	/* N.B.: differs from sender */
    addr.sin_port=htons(port);

    /* use SO_REUSEADDR */
    i_opt = true;
    if(setsockopt(desc, SOL_SOCKET, SO_REUSEADDR, (char *) &i_opt, sizeof(i_opt)) == -1) {
        ERROR_MSG("Cannot configure socket (SO_REUSEADDR: %s)", strerror(desc));
    }    

    /* bind to receive address */
    if(bind(desc, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
	ERROR_MSG("Error on bind: %s", strerror(desc));
	return false;
    }

    /* use setsockopt() to request that the kernel join a multicast group */
    mreq.imr_multiaddr.s_addr=inet_addr(address.c_str());
    if(strlen(iface) == 0)
	mreq.imr_interface.s_addr=htonl(INADDR_ANY);
    else
	mreq.imr_interface.s_addr=inet_addr(iface);
    if(setsockopt(desc, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char *) &mreq, sizeof(mreq)) < 0) {
	ERROR_MSG("Error on IP_ADD_MEMBERSHIP: %s", strerror(desc));
	return false;
    }

    /* Increase the receive buffer size to 1/2MB (8Mb/s during 1/2s) to avoid
     * packet loss caused by scheduling problems */
    i_opt = 0x80000;
    if(setsockopt(desc, SOL_SOCKET, SO_RCVBUF, (char *) &i_opt, sizeof(i_opt)) == -1) {
        ERROR_MSG("Cannot configure socket (SO_RCVBUF: %s)", strerror(desc));
    }

    i_opt = 0x80000;
    if(setsockopt(desc, SOL_SOCKET, SO_SNDBUF, (char *) &i_opt, sizeof(i_opt)) == -1) {
        ERROR_MSG("Cannot configure socket (SO_SNDBUF: %s)", strerror(desc));
    }

    ERROR_MSG("Connection OK, starting receive thread..");
    ThreadCreate();

    return true;
}

int udpSource::Run()
{
    // Connect the splitter
    Splitter * splitter;

    try {
	splitter = new mpeg_demux(false);
    } catch (const char * c) {
	fprintf(stderr, "[%d] EXCEPTION: %s\n", SDL_GetTicks(), c);
	TerminateThread(false);
	return 0;
    }

    // UDP stream is locked to source clock
    timer->unlocked = false;

    // Should be enough for 1 packet
    char buffer[1500];

    FILE * output = NULL;

    if(!filename.empty()) {
	output = fopen(filename.c_str(), "r");
	while(output) {
	    fclose(output);
	    const char * c = filename.replace('.', '\0');
	    if(c) {
		snprintf(buffer, 255, "%s - %d.%s", filename.c_str(), SDL_GetTicks(), c);
	    } else {
		snprintf(buffer, 255, "%s - %d", filename.c_str(), SDL_GetTicks());
	    }
	    filename = buffer;
	    output = fopen(filename.c_str(), "r");
	}
	output = fopen(filename.c_str(), "wb");
	if(!output) {
	    ERROR_MSG("Error opening file %s!", filename.c_str());
	} else {
	    ERROR_MSG("Recording to %s", filename.c_str());
	}
    }

    if(!splitter) {
	ERROR_MSG("Error connecting splitter, teminating thread!");
	return -1;
    }

    fd_set rfds;
    struct timeval tv;

    // Copy fd to local variable
    int fd = desc;

    ERROR_MSG("(ThreadID: %u) init complete", SDL_ThreadID());

    while(ThreadRun()) {

#ifdef WIN32
	int addrlen = sizeof(addr);
#else
	socklen_t addrlen = sizeof(addr);
#endif
	FD_ZERO(&rfds);
	FD_SET(fd, &rfds);
	tv.tv_sec = 0;
	tv.tv_usec = 250000;

	int n = select(fd+1, &rfds, NULL, NULL, &tv);
	if(n > 0) {
	    if((n = recvfrom(fd, buffer, 1500, 0, (struct sockaddr *) &addr, &addrlen)) < 0) {
		ERROR_MSG("Error on receive: %s, terminating thread!", strerror(desc));
		break;
	    }
	} else {
	    LOG_MSG("Timeout or error waiting for data!");
	    continue;
	}

	DEBUG_MSG("Read %d bytes", n);

	// ProcessData will block when decoding buffer is full,
	// but not if the decoder is not running
	splitter->ProcessData((unsigned char*)buffer, n);

	if(output) {
	    // Strip RTP headers from the recording
	    if((buffer[0] != 0x47) && (buffer[12] == 0x47))
		fwrite(buffer+12, sizeof(char), n-12, output);
	    else
		fwrite(buffer, sizeof(char), n, output);
	}
    }

    if(output)
	fclose(output);

    SAFE_DELETE(splitter);
    ERROR_MSG("(ThreadID: %u) finish complete", SDL_ThreadID());

    return 0;
}