Protocol refactoring.

This commit is contained in:
Rolf Ahrenberg 2007-09-14 15:44:25 +00:00
parent df3658365f
commit da50e032fd
8 changed files with 352 additions and 211 deletions

View File

@ -1,7 +1,7 @@
# #
# Makefile for a Video Disk Recorder plugin # Makefile for a Video Disk Recorder plugin
# #
# $Id: Makefile,v 1.1 2007/09/12 17:28:59 rahrenbe Exp $ # $Id: Makefile,v 1.2 2007/09/14 15:44:25 rahrenbe Exp $
# Debugging on/off # Debugging on/off
IPTV_DEBUG = 1 IPTV_DEBUG = 1
@ -57,7 +57,7 @@ endif
### The object files (add further files here): ### The object files (add further files here):
OBJS = $(PLUGIN).o device.o streamer.o OBJS = $(PLUGIN).o device.o streamer.o protocoludp.o
### The main target: ### The main target:

View File

@ -3,7 +3,7 @@
* *
* See the README file for copyright information and how to reach the author. * See the README file for copyright information and how to reach the author.
* *
* $Id: device.c,v 1.6 2007/09/13 18:14:41 rahrenbe Exp $ * $Id: device.c,v 1.7 2007/09/14 15:44:25 rahrenbe Exp $
*/ */
#include "common.h" #include "common.h"
@ -19,19 +19,23 @@ cIptvDevice::cIptvDevice(unsigned int Index)
: deviceIndex(Index), : deviceIndex(Index),
isPacketDelivered(false), isPacketDelivered(false),
isOpenDvr(false), isOpenDvr(false),
pIptvStreamer(NULL) mutex()
{ {
debug("cIptvDevice::cIptvDevice(%d)\n", deviceIndex); debug("cIptvDevice::cIptvDevice(%d)\n", deviceIndex);
tsBuffer = new cRingBufferLinear(MEGABYTE(8), TS_SIZE, false, "IPTV"); tsBuffer = new cRingBufferLinear(MEGABYTE(8), TS_SIZE, false, "IPTV");
tsBuffer->SetTimeouts(100, 100); tsBuffer->SetTimeouts(100, 100);
pUdpProtocol = new cIptvProtocolUdp();
//pRtspProtocol = new cIptvProtocolRtsp();
//pHttpProtocol = new cIptvProtocolHttp();
pIptvStreamer = new cIptvStreamer(tsBuffer, &mutex); pIptvStreamer = new cIptvStreamer(tsBuffer, &mutex);
StartSectionHandler();
} }
cIptvDevice::~cIptvDevice() cIptvDevice::~cIptvDevice()
{ {
debug("cIptvDevice::~cIptvDevice(%d)\n", deviceIndex); debug("cIptvDevice::~cIptvDevice(%d)\n", deviceIndex);
if (pIptvStreamer) delete pIptvStreamer;
delete pIptvStreamer; delete pUdpProtocol;
delete tsBuffer; delete tsBuffer;
} }
@ -64,21 +68,21 @@ cIptvDevice *cIptvDevice::Get(unsigned int DeviceIndex)
return NULL; return NULL;
} }
cString cIptvDevice::GetChannelSettings(const char *Param, int *IpPort, int *Protocol) cString cIptvDevice::GetChannelSettings(const char *Param, int *IpPort, cIptvProtocolIf* *Protocol)
{ {
unsigned int a, b, c, d; unsigned int a, b, c, d;
debug("cIptvDevice::GetChannelSettings(%d)\n", deviceIndex); debug("cIptvDevice::GetChannelSettings(%d)\n", deviceIndex);
if (sscanf(Param, "IPTV-UDP-%u.%u.%u.%u-%u", &a, &b, &c, &d, IpPort) == 5) { if (sscanf(Param, "IPTV-UDP-%u.%u.%u.%u-%u", &a, &b, &c, &d, IpPort) == 5) {
*Protocol = PROTOCOL_UDP; *Protocol = pUdpProtocol;
return cString::sprintf("%u.%u.%u.%u", a, b, c, d); return cString::sprintf("%u.%u.%u.%u", a, b, c, d);
} }
else if (sscanf(Param, "IPTV-RTSP-%u.%u.%u.%u-%u", &a, &b, &c, &d, IpPort) == 5) { else if (sscanf(Param, "IPTV-RTSP-%u.%u.%u.%u-%u", &a, &b, &c, &d, IpPort) == 5) {
*Protocol = PROTOCOL_RTSP; *Protocol = NULL; // pRtspProtocol;
return cString::sprintf("%u.%u.%u.%u", a, b, c, d); return cString::sprintf("%u.%u.%u.%u", a, b, c, d);
} }
else if (sscanf(Param, "IPTV-HTTP-%u.%u.%u.%u-%u", &a, &b, &c, &d, IpPort) == 5) { else if (sscanf(Param, "IPTV-HTTP-%u.%u.%u.%u-%u", &a, &b, &c, &d, IpPort) == 5) {
*Protocol = PROTOCOL_HTTP; *Protocol = NULL; // pHttpProtocol;
return cString::sprintf("%u.%u.%u.%u", a, b, c, d); return cString::sprintf("%u.%u.%u.%u", a, b, c, d);
} }
return NULL; return NULL;
@ -117,13 +121,14 @@ bool cIptvDevice::ProvidesChannel(const cChannel *Channel, int Priority, bool *N
bool cIptvDevice::SetChannelDevice(const cChannel *Channel, bool LiveView) bool cIptvDevice::SetChannelDevice(const cChannel *Channel, bool LiveView)
{ {
int port, protocol; int port;
cString addr; cString addr;
cIptvProtocolIf *protocol;
debug("cIptvDevice::SetChannelDevice(%d)\n", deviceIndex); debug("cIptvDevice::SetChannelDevice(%d)\n", deviceIndex);
addr = GetChannelSettings(Channel->Param(), &port, &protocol); addr = GetChannelSettings(Channel->Param(), &port, &protocol);
if (addr) if (addr)
pIptvStreamer->SetStream(addr, port, (protocol == PROTOCOL_UDP) ? "udp" : (protocol == PROTOCOL_RTSP) ? "rtsp" : "http"); pIptvStreamer->Set(addr, port, protocol);
return true; return true;
} }
@ -140,7 +145,7 @@ bool cIptvDevice::OpenDvr(void)
isPacketDelivered = false; isPacketDelivered = false;
tsBuffer->Clear(); tsBuffer->Clear();
mutex.Unlock(); mutex.Unlock();
pIptvStreamer->OpenStream(); pIptvStreamer->Open();
isOpenDvr = true; isOpenDvr = true;
return true; return true;
} }
@ -148,7 +153,7 @@ bool cIptvDevice::OpenDvr(void)
void cIptvDevice::CloseDvr(void) void cIptvDevice::CloseDvr(void)
{ {
debug("cIptvDevice::CloseDvr(%d)\n", deviceIndex); debug("cIptvDevice::CloseDvr(%d)\n", deviceIndex);
pIptvStreamer->CloseStream(); pIptvStreamer->Close();
isOpenDvr = false; isOpenDvr = false;
} }

View File

@ -3,13 +3,16 @@
* *
* See the README file for copyright information and how to reach the author. * See the README file for copyright information and how to reach the author.
* *
* $Id: device.h,v 1.3 2007/09/13 18:14:41 rahrenbe Exp $ * $Id: device.h,v 1.4 2007/09/14 15:44:25 rahrenbe Exp $
*/ */
#ifndef __IPTV_DEVICE_H #ifndef __IPTV_DEVICE_H
#define __IPTV_DEVICE_H #define __IPTV_DEVICE_H
#include <vdr/device.h> #include <vdr/device.h>
#include "protocoludp.h"
//#include "protocolrtsp.h"
//#include "protocolhttp.h"
#include "streamer.h" #include "streamer.h"
class cIptvDevice : public cDevice { class cIptvDevice : public cDevice {
@ -22,15 +25,13 @@ public:
// private parts // private parts
private: private:
enum tProtocol {
PROTOCOL_UDP,
PROTOCOL_RTSP,
PROTOCOL_HTTP
};
unsigned int deviceIndex; unsigned int deviceIndex;
bool isPacketDelivered; bool isPacketDelivered;
bool isOpenDvr; bool isOpenDvr;
cRingBufferLinear *tsBuffer; cRingBufferLinear *tsBuffer;
cIptvProtocolUdp *pUdpProtocol;
//cIptvProtocolRtsp *pRtspProtocol;
//cIptvProtocolHttp *pHttpProtocol;
cIptvStreamer *pIptvStreamer; cIptvStreamer *pIptvStreamer;
cMutex mutex; cMutex mutex;
@ -41,7 +42,7 @@ public:
// for channel parsing // for channel parsing
private: private:
cString GetChannelSettings(const char *Param, int *IpPort, int *Protocol); cString GetChannelSettings(const char *Param, int *IpPort, cIptvProtocolIf* *Protocol);
bool ProvidesIptv(const char *Param) const; bool ProvidesIptv(const char *Param) const;
// for channel selection // for channel selection

26
protocolif.h Normal file
View File

@ -0,0 +1,26 @@
/*
* protocolif.h: IPTV plugin for the Video Disk Recorder
*
* See the README file for copyright information and how to reach the author.
*
* $Id: protocolif.h,v 1.1 2007/09/14 15:44:25 rahrenbe Exp $
*/
#ifndef __IPTV_PROTOCOLIF_H
#define __IPTV_PROTOCOLIF_H
class cIptvProtocolIf {
public:
cIptvProtocolIf() {}
virtual ~cIptvProtocolIf() {}
virtual int Read(unsigned char *Buffer, int Len) = 0;
virtual bool Set(const char* Address, const int Port) = 0;
virtual bool Open(void) = 0;
virtual bool Close(void) = 0;
private:
cIptvProtocolIf(const cIptvProtocolIf&);
cIptvProtocolIf& operator=(const cIptvProtocolIf&);
};
#endif // __IPTV_PROTOCOLIF_H

206
protocoludp.c Normal file
View File

@ -0,0 +1,206 @@
/*
* streamer.c: IPTV plugin for the Video Disk Recorder
*
* See the README file for copyright information and how to reach the author.
*
* $Id: protocoludp.c,v 1.1 2007/09/14 15:44:25 rahrenbe Exp $
*/
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <fcntl.h>
#include <unistd.h>
#include "common.h"
#include "protocoludp.h"
cIptvProtocolUdp::cIptvProtocolUdp()
: streamPort(1234),
socketDesc(-1),
mcastActive(false)
{
debug("cIptvProtocolUdp::cIptvProtocolUdp()\n");
streamAddr = strdup("");
}
cIptvProtocolUdp::~cIptvProtocolUdp()
{
debug("cIptvProtocolUdp::~cIptvProtocolUdp()\n");
// Drop the multicast group and close the socket
Close();
// Free allocated memory
free(streamAddr);
}
bool cIptvProtocolUdp::OpenSocket(const int Port)
{
debug("cIptvProtocolUdp::OpenSocket()\n");
// If socket is there already and it is bound to a different port, it must
// be closed first
if (Port != streamPort) {
debug("cIptvProtocolUdp::OpenSocket(): Socket tear-down\n");
CloseSocket();
}
// Bind to the socket if it is not active already
if (socketDesc < 0) {
int yes = 1;
// Create socket
socketDesc = socket(PF_INET, SOCK_DGRAM, 0);
if (socketDesc < 0) {
char tmp[64];
error("ERROR: socket(): %s", strerror_r(errno, tmp, sizeof(tmp)));
return false;
}
// Make it use non-blocking I/O to avoid stuck read calls
if (fcntl(socketDesc, F_SETFL, O_NONBLOCK)) {
char tmp[64];
error("ERROR: fcntl(): %s", strerror_r(errno, tmp, sizeof(tmp)));
CloseSocket();
return false;
}
// Allow multiple sockets to use the same PORT number
if (setsockopt(socketDesc, SOL_SOCKET, SO_REUSEADDR, &yes,
sizeof(yes)) < 0) {
char tmp[64];
error("ERROR: setsockopt(): %s", strerror_r(errno, tmp, sizeof(tmp)));
CloseSocket();
return false;
}
// Bind socket
memset(&sockAddr, '\0', sizeof(sockAddr));
sockAddr.sin_family = AF_INET;
sockAddr.sin_port = htons(Port);
sockAddr.sin_addr.s_addr = htonl(INADDR_ANY);
int err = bind(socketDesc, (struct sockaddr *)&sockAddr, sizeof(sockAddr));
if (err < 0) {
char tmp[64];
error("ERROR: bind(): %s", strerror_r(errno, tmp, sizeof(tmp)));
CloseSocket();
return false;
}
// Update stream port
streamPort = Port;
}
return true;
}
void cIptvProtocolUdp::CloseSocket(void)
{
debug("cIptvProtocolUdp::CloseSocket()\n");
// Check if socket exists
if (socketDesc >= 0) {
close(socketDesc);
socketDesc = -1;
}
}
bool cIptvProtocolUdp::JoinMulticast(void)
{
debug("cIptvProtocolUdp::JoinMulticast()\n");
// Check that stream address is valid
if (!mcastActive && !isempty(streamAddr)) {
// Ensure that socket is valid
OpenSocket(streamPort);
// Join a new multicast group
struct ip_mreq mreq;
mreq.imr_multiaddr.s_addr = inet_addr(streamAddr);
mreq.imr_interface.s_addr = htonl(INADDR_ANY);
int err = setsockopt(socketDesc, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq,
sizeof(mreq));
if (err < 0) {
char tmp[64];
error("ERROR: setsockopt(): %s", strerror_r(errno, tmp, sizeof(tmp)));
return false;
}
// Update multicasting flag
mcastActive = true;
}
return true;
}
bool cIptvProtocolUdp::DropMulticast(void)
{
debug("cIptvProtocolUdp::DropMulticast()\n");
// Check that stream address is valid
if (mcastActive && !isempty(streamAddr)) {
// Ensure that socket is valid
OpenSocket(streamPort);
// Drop the multicast group
struct ip_mreq mreq;
mreq.imr_multiaddr.s_addr = inet_addr(streamAddr);
mreq.imr_interface.s_addr = htonl(INADDR_ANY);
int err = setsockopt(socketDesc, IPPROTO_IP, IP_DROP_MEMBERSHIP, &mreq,
sizeof(mreq));
if (err < 0) {
char tmp[64];
error("ERROR: setsockopt(): %s", strerror_r(errno, tmp, sizeof(tmp)));
return false;
}
// Update multicasting flag
mcastActive = false;
}
return true;
}
int cIptvProtocolUdp::Read(unsigned char *Buffer, int Len)
{
//debug("cIptvProtocolUdp::Read()\n");
socklen_t addrlen = sizeof(sockAddr);
// Wait for data
struct timeval tv;
tv.tv_sec = 0;
tv.tv_usec = 500000;
// Use select
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(socketDesc, &rfds);
int retval = select(socketDesc + 1, &rfds, NULL, NULL, &tv);
// Check if error
if (retval < 0) {
char tmp[64];
error("ERROR: select(): %s", strerror_r(errno, tmp, sizeof(tmp)));
return -1;
}
// Check if data available
else if (retval) {
// Read data from socket
return recvfrom(socketDesc, Buffer, Len, MSG_DONTWAIT,
(struct sockaddr *)&sockAddr, &addrlen);
}
return 0;
}
bool cIptvProtocolUdp::Open(void)
{
debug("cIptvProtocolUdp::Open(): streamAddr = %s\n", streamAddr);
// Join a new multicast group
JoinMulticast();
return true;
}
bool cIptvProtocolUdp::Close(void)
{
debug("cIptvProtocolUdp::Close(): streamAddr = %s\n", streamAddr);
// Drop the multicast group
DropMulticast();
// Close the socket
CloseSocket();
return true;
}
bool cIptvProtocolUdp::Set(const char* Address, const int Port)
{
debug("cIptvProtocolUdp::Set(): %s:%d\n", Address, Port);
if (!isempty(Address)) {
// Drop the multicast group
DropMulticast();
// Update stream address and port
streamAddr = strcpyrealloc(streamAddr, Address);
streamPort = Port;
// Join a new multicast group
JoinMulticast();
}
return true;
}

39
protocoludp.h Normal file
View File

@ -0,0 +1,39 @@
/*
* protocoludp.h: IPTV plugin for the Video Disk Recorder
*
* See the README file for copyright information and how to reach the author.
*
* $Id: protocoludp.h,v 1.1 2007/09/14 15:44:25 rahrenbe Exp $
*/
#ifndef __IPTV_PROTOCOLUDP_H
#define __IPTV_PROTOCOLUDP_H
#include <arpa/inet.h>
#include "protocolif.h"
class cIptvProtocolUdp : public cIptvProtocolIf {
private:
char* streamAddr;
int streamPort;
int socketDesc;
struct sockaddr_in sockAddr;
bool mcastActive;
private:
bool OpenSocket(const int Port);
void CloseSocket(void);
bool JoinMulticast(void);
bool DropMulticast(void);
public:
cIptvProtocolUdp();
virtual ~cIptvProtocolUdp();
virtual int Read(unsigned char *Buffer, int Len);
virtual bool Set(const char* Address, const int Port);
virtual bool Open(void);
virtual bool Close(void);
};
#endif // __IPTV_PROTOCOLUDP_H

View File

@ -3,7 +3,7 @@
* *
* See the README file for copyright information and how to reach the author. * See the README file for copyright information and how to reach the author.
* *
* $Id: streamer.c,v 1.10 2007/09/13 16:58:22 rahrenbe Exp $ * $Id: streamer.c,v 1.11 2007/09/14 15:44:25 rahrenbe Exp $
*/ */
#include <sys/types.h> #include <sys/types.h>
@ -19,68 +19,45 @@
#include "common.h" #include "common.h"
#include "streamer.h" #include "streamer.h"
cIptvStreamer::cIptvStreamer(cRingBufferLinear* Buffer, cMutex* Mutex) cIptvStreamer::cIptvStreamer(cRingBufferLinear* RingBuffer, cMutex* Mutex)
: cThread("IPTV streamer"), : cThread("IPTV streamer"),
streamPort(1234), ringBuffer(RingBuffer),
socketDesc(-1),
pRingBuffer(Buffer),
bufferSize(TS_SIZE * 7),
mutex(Mutex), mutex(Mutex),
mcastActive(false) readBufferLen(TS_SIZE * 7),
protocol(NULL)
{ {
debug("cIptvStreamer::cIptvStreamer()\n"); debug("cIptvStreamer::cIptvStreamer()\n");
streamAddr = strdup("");
// Create the socket
OpenSocket(streamPort);
// Allocate receive buffer // Allocate receive buffer
pReceiveBuffer = MALLOC(unsigned char, bufferSize); readBuffer = MALLOC(unsigned char, readBufferLen);
if (!pReceiveBuffer) if (!readBuffer)
error("ERROR: MALLOC(pReceiveBuffer) failed"); error("ERROR: MALLOC(readBuffer) failed");
} }
cIptvStreamer::~cIptvStreamer() cIptvStreamer::~cIptvStreamer()
{ {
debug("cIptvStreamer::~cIptvStreamer()\n"); debug("cIptvStreamer::~cIptvStreamer()\n");
// Drop the multicast group // Close the protocol
DropMulticast(); Close();
// Close the stream and socket
CloseStream();
CloseSocket();
// Free allocated memory // Free allocated memory
free(streamAddr); free(readBuffer);
free(pReceiveBuffer);
} }
void cIptvStreamer::Action(void) void cIptvStreamer::Action(void)
{ {
debug("cIptvStreamer::Action(): Entering\n"); debug("cIptvStreamer::Action(): Entering\n");
// Create files necessary for selecting I/O from socket
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(socketDesc, &rfds);
// Do the thread loop // Do the thread loop
while (Running()) { while (Running()) {
if (pRingBuffer && mutex && pReceiveBuffer && (socketDesc >= 0)) { if (ringBuffer && mutex && readBuffer && protocol) {
struct timeval tv; int length = protocol->Read(readBuffer, readBufferLen);
socklen_t addrlen = sizeof(sockAddr); if (length >= 0) {
// Wait for data
tv.tv_sec = 0;
tv.tv_usec = 500000;
int retval = select(socketDesc + 1, &rfds, NULL, NULL, &tv);
if (retval < 0) {
char tmp[64];
error("ERROR: select(): %s", strerror_r(errno, tmp, sizeof(tmp)));
} else if (retval) {
// Read data from socket
int length = recvfrom(socketDesc, pReceiveBuffer, bufferSize, MSG_DONTWAIT,
(struct sockaddr *)&sockAddr, &addrlen);
mutex->Lock(); mutex->Lock();
int p = pRingBuffer->Put(pReceiveBuffer, length); int p = ringBuffer->Put(readBuffer, length);
if (p != length && Running()) if (p != length && Running())
pRingBuffer->ReportOverflow(length - p); ringBuffer->ReportOverflow(length - p);
mutex->Unlock(); mutex->Unlock();
} else }
debug("cIptvStreamer::Action(): Timeout\n"); else
cCondWait::SleepMs(3); // reduce cpu load
} }
else else
cCondWait::SleepMs(100); // avoid busy loop cCondWait::SleepMs(100); // avoid busy loop
@ -88,148 +65,44 @@ void cIptvStreamer::Action(void)
debug("cIptvStreamer::Action(): Exiting\n"); debug("cIptvStreamer::Action(): Exiting\n");
} }
bool cIptvStreamer::OpenSocket(const int port) bool cIptvStreamer::Open(void)
{ {
debug("cIptvStreamer::OpenSocket()\n"); debug("cIptvStreamer::Open()\n");
// If socket is there already and it is bound to a different port, it must // Open the protocol
// be closed first if (protocol)
if (port != streamPort) { protocol->Open();
debug("cIptvStreamer::OpenSocket(): Socket tear-down\n");
CloseSocket();
}
// Bind to the socket if it is not active already
if (socketDesc < 0) {
int yes = 1;
// Create socket
socketDesc = socket(PF_INET, SOCK_DGRAM, 0);
if (socketDesc < 0) {
char tmp[64];
error("ERROR: socket(): %s", strerror_r(errno, tmp, sizeof(tmp)));
return false;
}
// Make it use non-blocking I/O to avoid stuck read calls
if (fcntl(socketDesc, F_SETFL, O_NONBLOCK)) {
char tmp[64];
error("ERROR: fcntl(): %s", strerror_r(errno, tmp, sizeof(tmp)));
CloseSocket();
return false;
}
// Allow multiple sockets to use the same PORT number
if (setsockopt(socketDesc, SOL_SOCKET, SO_REUSEADDR, &yes,
sizeof(yes)) < 0) {
char tmp[64];
error("ERROR: setsockopt(): %s", strerror_r(errno, tmp, sizeof(tmp)));
CloseSocket();
return false;
}
// Bind socket
memset(&sockAddr, '\0', sizeof(sockAddr));
sockAddr.sin_family = AF_INET;
sockAddr.sin_port = htons(port);
sockAddr.sin_addr.s_addr = htonl(INADDR_ANY);
int err = bind(socketDesc, (struct sockaddr *)&sockAddr, sizeof(sockAddr));
if (err < 0) {
char tmp[64];
error("ERROR: bind(): %s", strerror_r(errno, tmp, sizeof(tmp)));
CloseSocket();
return false;
}
// Update stream port
streamPort = port;
}
return true;
}
void cIptvStreamer::CloseSocket(void)
{
debug("cIptvStreamer::CloseSocket()\n");
// Check if socket exists
if (socketDesc >= 0) {
close(socketDesc);
socketDesc = -1;
}
}
bool cIptvStreamer::JoinMulticast(void)
{
debug("cIptvStreamer::JoinMulticast()\n");
// Check that stream address is valid
if (!mcastActive && !isempty(streamAddr)) {
// Ensure that socket is valid
OpenSocket(streamPort);
// Join a new multicast group
struct ip_mreq mreq;
mreq.imr_multiaddr.s_addr = inet_addr(streamAddr);
mreq.imr_interface.s_addr = htonl(INADDR_ANY);
int err = setsockopt(socketDesc, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq,
sizeof(mreq));
if (err < 0) {
char tmp[64];
error("ERROR: setsockopt(): %s", strerror_r(errno, tmp, sizeof(tmp)));
return false;
}
// Update multicasting flag
mcastActive = true;
}
return true;
}
bool cIptvStreamer::DropMulticast(void)
{
debug("cIptvStreamer::DropMulticast()\n");
// Check that stream address is valid
if (mcastActive && !isempty(streamAddr)) {
// Ensure that socket is valid
OpenSocket(streamPort);
// Drop the multicast group
struct ip_mreq mreq;
mreq.imr_multiaddr.s_addr = inet_addr(streamAddr);
mreq.imr_interface.s_addr = htonl(INADDR_ANY);
int err = setsockopt(socketDesc, IPPROTO_IP, IP_DROP_MEMBERSHIP, &mreq,
sizeof(mreq));
if (err < 0) {
char tmp[64];
error("ERROR: setsockopt(): %s", strerror_r(errno, tmp, sizeof(tmp)));
return false;
}
// Update multicasting flag
mcastActive = false;
}
return true;
}
bool cIptvStreamer::OpenStream(void)
{
debug("cIptvStreamer::OpenStream(): streamAddr = %s\n", streamAddr);
// Join a new multicast group
JoinMulticast();
// Start thread // Start thread
Start(); Start();
return true; return true;
} }
bool cIptvStreamer::CloseStream(void) bool cIptvStreamer::Close(void)
{ {
debug("cIptvStreamer::CloseStream(): streamAddr = %s\n", streamAddr); debug("cIptvStreamer::Close()\n");
// Stop thread // Stop thread
if (Running()) if (Running())
Cancel(3); Cancel(3);
// Drop the multicast group // Close the protocol
DropMulticast(); if (protocol)
protocol->Close();
return true; return true;
} }
bool cIptvStreamer::SetStream(const char* address, const int port, const char* protocol) bool cIptvStreamer::Set(const char* Address, const int Port, cIptvProtocolIf* Protocol)
{ {
debug("cIptvStreamer::SetStream(): %s://%s:%d\n", protocol, address, port); debug("cIptvStreamer::Set(): %s:%d\n", Address, Port);
if (!isempty(address)) { if (!isempty(Address)) {
// Drop the multicast group // Update protocol; Close the existing one if changed
DropMulticast(); if (protocol != Protocol) {
// Update stream address and port if (protocol)
streamAddr = strcpyrealloc(streamAddr, address); protocol->Close();
streamPort = port; protocol = Protocol;
// Join a new multicast group if (protocol)
JoinMulticast(); protocol->Open();
}
// Set protocol address and port
if (protocol)
protocol->Set(Address, Port);
} }
return true; return true;
} }

View File

@ -3,7 +3,7 @@
* *
* See the README file for copyright information and how to reach the author. * See the README file for copyright information and how to reach the author.
* *
* $Id: streamer.h,v 1.5 2007/09/13 16:58:22 rahrenbe Exp $ * $Id: streamer.h,v 1.6 2007/09/14 15:44:25 rahrenbe Exp $
*/ */
#ifndef __IPTV_STREAMER_H #ifndef __IPTV_STREAMER_H
@ -14,32 +14,23 @@
#include <vdr/thread.h> #include <vdr/thread.h>
#include <vdr/ringbuffer.h> #include <vdr/ringbuffer.h>
#include "protocolif.h"
class cIptvStreamer : public cThread { class cIptvStreamer : public cThread {
private: private:
char* streamAddr; cRingBufferLinear* ringBuffer;
int streamPort;
int socketDesc;
struct sockaddr_in sockAddr;
cRingBufferLinear* pRingBuffer;
unsigned char* pReceiveBuffer;
unsigned int bufferSize;
cMutex* mutex; cMutex* mutex;
bool mcastActive; unsigned char* readBuffer;
unsigned int readBufferLen;
private: cIptvProtocolIf* protocol;
bool OpenSocket(const int port);
void CloseSocket(void);
bool JoinMulticast(void);
bool DropMulticast(void);
public: public:
cIptvStreamer(); cIptvStreamer(cRingBufferLinear* RingBuffer, cMutex* Mutex);
cIptvStreamer(cRingBufferLinear* Buffer, cMutex* Mutex);
virtual ~cIptvStreamer(); virtual ~cIptvStreamer();
virtual void Action(void); virtual void Action(void);
bool SetStream(const char* address, const int port, const char* protocol); bool Set(const char* Address, const int Port, cIptvProtocolIf* Protocol);
bool OpenStream(void); bool Open(void);
bool CloseStream(void); bool Close(void);
}; };
#endif // __IPTV_STREAMER_H #endif // __IPTV_STREAMER_H