Snapshot 2007-05-09

This commit is contained in:
Frank Schmirler
2010-12-02 08:59:14 +01:00
parent e6249bf957
commit 5a270cc3ab
22 changed files with 1040 additions and 260 deletions

View File

@@ -1,5 +1,5 @@
/*
* $Id: device.c,v 1.8 2007/01/15 12:15:12 schmirl Exp $
* $Id: device.c,v 1.13 2007/05/07 12:18:18 schmirl Exp $
*/
#include "client/device.h"
@@ -42,6 +42,8 @@ cStreamdevDevice::cStreamdevDevice(void) {
#endif
m_Device = this;
m_Pids = 0;
m_DvrClosed = true;
if (StreamdevClientSetup.SyncEPG)
ClientSocket.SynchronizeEPG();
@@ -59,13 +61,22 @@ cStreamdevDevice::~cStreamdevDevice() {
bool cStreamdevDevice::ProvidesSource(int Source) const {
Dprintf("ProvidesSource, Source=%d\n", Source);
return false;
return true;
}
bool cStreamdevDevice::ProvidesTransponder(const cChannel *Channel) const
{
Dprintf("ProvidesTransponder\n");
return false;
return true;
}
bool cStreamdevDevice::IsTunedToTransponder(const cChannel *Channel)
{
bool res = false;
if (ClientSocket.DataSocket(siLive) != NULL
&& TRANSPONDER(Channel, m_Channel))
res = true;
return res;
}
bool cStreamdevDevice::ProvidesChannel(const cChannel *Channel, int Priority,
@@ -123,37 +134,118 @@ bool cStreamdevDevice::SetChannelDevice(const cChannel *Channel,
bool cStreamdevDevice::SetPid(cPidHandle *Handle, int Type, bool On) {
Dprintf("SetPid, Pid=%d, Type=%d, On=%d, used=%d\n", Handle->pid, Type, On,
Handle->used);
if (Handle->pid && (On || !Handle->used))
return ClientSocket.SetPid(Handle->pid, On);
return true;
LOCK_THREAD;
if (On && !m_TSBuffer) {
Dprintf("SetPid: no data connection -> OpenDvr()");
OpenDvrInt();
}
bool res = true;
if (Handle->pid && (On || !Handle->used)) {
res = ClientSocket.SetPid(Handle->pid, On);
m_Pids += (!res) ? 0 : On ? 1 : -1;
if (m_Pids < 0)
m_Pids = 0;
if(m_Pids < 1 && m_DvrClosed) {
Dprintf("SetPid: 0 pids left -> CloseDvr()");
CloseDvrInt();
}
}
return res;
}
bool cStreamdevDevice::OpenDvrInt(void) {
Dprintf("OpenDvrInt\n");
LOCK_THREAD;
CloseDvrInt();
if (m_TSBuffer) {
Dprintf("cStreamdevDevice::OpenDvrInt(): DVR connection already open\n");
return true;
}
Dprintf("cStreamdevDevice::OpenDvrInt(): Connecting ...\n");
if (ClientSocket.CreateDataConnection(siLive)) {
m_TSBuffer = new cTSBuffer(*ClientSocket.DataSocket(siLive), MEGABYTE(2), CardIndex() + 1);
return true;
}
esyslog("cStreamdevDevice::OpenDvrInt(): DVR connection FAILED");
return false;
}
bool cStreamdevDevice::OpenDvr(void) {
Dprintf("OpenDvr\n");
CloseDvr();
if (ClientSocket.CreateDataConnection(siLive)) {
//m_Assembler = new cStreamdevAssembler(ClientSocket.DataSocket(siLive));
//m_TSBuffer = new cTSBuffer(m_Assembler->ReadPipe(), MEGABYTE(2), CardIndex() + 1);
m_TSBuffer = new cTSBuffer(*ClientSocket.DataSocket(siLive), MEGABYTE(2), CardIndex() + 1);
Dprintf("waiting\n");
//m_Assembler->WaitForFill();
Dprintf("resuming\n");
return true;
LOCK_THREAD;
m_DvrClosed = false;
return OpenDvrInt();
}
void cStreamdevDevice::CloseDvrInt(void) {
Dprintf("CloseDvrInt\n");
LOCK_THREAD;
if (ClientSocket.CheckConnection()) {
if (!m_DvrClosed) {
Dprintf("cStreamdevDevice::CloseDvrInt(): m_DvrClosed=false -> not closing yet\n");
return;
}
if (m_Pids > 0) {
Dprintf("cStreamdevDevice::CloseDvrInt(): %d active pids -> not closing yet\n", m_Pids);
return;
}
} else {
Dprintf("cStreamdevDevice::CloseDvrInt(): Control connection gone !\n");
}
return false;
Dprintf("cStreamdevDevice::CloseDvrInt(): Closing DVR connection\n");
DELETENULL(m_TSBuffer);
ClientSocket.CloseDvr();
}
void cStreamdevDevice::CloseDvr(void) {
Dprintf("CloseDvr\n");
LOCK_THREAD;
//DELETENULL(m_Assembler);
DELETENULL(m_TSBuffer);
ClientSocket.CloseDvr();
m_DvrClosed = true;
CloseDvrInt();
}
bool cStreamdevDevice::GetTSPacket(uchar *&Data) {
if (m_TSBuffer) {
Data = m_TSBuffer->Get();
#if 1 // TODO: this should be fixed in vdr cTSBuffer
// simple disconnect detection
static int m_TSFails = 0;
if (!Data) {
LOCK_THREAD;
if(!ClientSocket.DataSocket(siLive)) {
return false; // triggers CloseDvr() + OpenDvr() in cDevice
}
cPoller Poller(*ClientSocket.DataSocket(siLive));
errno = 0;
if (Poller.Poll() && !errno) {
char tmp[1];
if (recv(*ClientSocket.DataSocket(siLive), tmp, 1, MSG_PEEK) == 0 && !errno) {
esyslog("cStreamDevice::GetTSPacket: GetChecked: NOTHING (%d)", m_TSFails);
m_TSFails++;
if (m_TSFails > 10) {
isyslog("cStreamdevDevice::GetTSPacket(): disconnected");
m_Pids = 0;
CloseDvrInt();
m_TSFails = 0;
return false;
}
return true;
}
}
m_TSFails = 0;
}
#endif
return true;
}
return false;
@@ -162,11 +254,24 @@ bool cStreamdevDevice::GetTSPacket(uchar *&Data) {
#if VDRVERSNUM >= 10300
int cStreamdevDevice::OpenFilter(u_short Pid, u_char Tid, u_char Mask) {
Dprintf("OpenFilter\n");
if (StreamdevClientSetup.StreamFilters
&& ClientSocket.SetFilter(Pid, Tid, Mask, true)) {
return m_Filters->OpenFilter(Pid, Tid, Mask);
} else
if (!StreamdevClientSetup.StreamFilters)
return -1;
if (!ClientSocket.DataSocket(siLiveFilter)) {
if (ClientSocket.CreateDataConnection(siLiveFilter)) {
m_Filters->SetConnection(*ClientSocket.DataSocket(siLiveFilter));
} else {
isyslog("cStreamdevDevice::OpenFilter: connect failed: %m");
return -1;
}
}
if (ClientSocket.SetFilter(Pid, Tid, Mask, true))
return m_Filters->OpenFilter(Pid, Tid, Mask);
return -1;
}
#endif
@@ -177,11 +282,17 @@ bool cStreamdevDevice::Init(void) {
}
bool cStreamdevDevice::ReInit(void) {
if(m_Device) {
m_Device->Lock();
m_Device->m_Filters->SetConnection(-1);
m_Device->m_Pids = 0;
}
ClientSocket.Quit();
ClientSocket.Reset();
if (m_Device != NULL) {
DELETENULL(m_Device->m_TSBuffer);
//DELETENULL(m_Device->m_TSBuffer);
DELETENULL(m_Device->m_Assembler);
m_Device->Unlock();
}
return StreamdevClientSetup.StartClient ? Init() : true;
}

View File

@@ -1,5 +1,5 @@
/*
* $Id: device.h,v 1.3 2005/02/08 15:21:19 lordjaxom Exp $
* $Id: device.h,v 1.5 2007/04/24 10:43:40 schmirl Exp $
*/
#ifndef VDR_STREAMDEV_DEVICE_H
@@ -25,9 +25,14 @@ private:
#if VDRVERSNUM >= 10307
cStreamdevFilters *m_Filters;
#endif
int m_Pids;
bool m_DvrClosed;
static cStreamdevDevice *m_Device;
bool OpenDvrInt(void);
void CloseDvrInt(void);
protected:
virtual bool SetChannelDevice(const cChannel *Channel, bool LiveView);
virtual bool HasLock(int TimeoutMs)
@@ -51,9 +56,10 @@ public:
virtual ~cStreamdevDevice();
virtual bool ProvidesSource(int Source) const;
virtual bool ProvidesTransponder(const cChannel *Channel) const;
virtual bool ProvidesTransponder(const cChannel *Channel) const;
virtual bool ProvidesChannel(const cChannel *Channel, int Priority = -1,
bool *NeedsDetachReceivers = NULL) const;
virtual bool IsTunedToTransponder(const cChannel *Channel);
static bool Init(void);
static bool ReInit(void);

View File

@@ -1,5 +1,5 @@
/*
* $Id: filter.c,v 1.3 2005/11/06 16:43:58 lordjaxom Exp $
* $Id: filter.c,v 1.11 2007/04/24 11:23:16 schmirl Exp $
*/
#include "client/filter.h"
@@ -7,113 +7,288 @@
#include "tools/select.h"
#include "common.h"
#include <vdr/ringbuffer.h>
#include <vdr/device.h>
#if VDRVERSNUM >= 10300
// --- cStreamdevFilter ------------------------------------------------------
class cStreamdevFilter: public cListObject {
private:
uchar m_Buffer[4096];
int m_Used;
int m_Pipe[2];
u_short m_Pid;
u_char m_Tid;
u_char m_Mask;
public:
cStreamdevFilter(u_short Pid, u_char Tid, u_char Mask);
virtual ~cStreamdevFilter();
bool Matches(u_short Pid, u_char Tid);
bool PutSection(const uchar *Data, int Length, bool Pusi);
int ReadPipe(void) const { return m_Pipe[0]; }
bool IsClosed(void);
void Reset(void);
u_short Pid(void) const { return m_Pid; }
u_char Tid(void) const { return m_Tid; }
u_char Mask(void) const { return m_Mask; }
};
inline bool cStreamdevFilter::Matches(u_short Pid, u_char Tid) {
return m_Pid == Pid && m_Tid == (Tid & m_Mask);
}
cStreamdevFilter::cStreamdevFilter(u_short Pid, u_char Tid, u_char Mask) {
m_Used = 0;
m_Pid = Pid;
m_Tid = Tid;
m_Mask = Mask;
m_Pipe[0] = m_Pipe[1] = -1;
if (pipe(m_Pipe) != 0 || fcntl(m_Pipe[0], F_SETFL, O_NONBLOCK) != 0) {
esyslog("streamev-client: coudln't open section filter pipe: %m");
m_Pipe[0] = m_Pipe[1] = -1;
#ifdef SOCK_SEQPACKET
// SOCK_SEQPACKET (since kernel 2.6.4)
if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, m_Pipe) != 0) {
esyslog("streamdev-client: socketpair(SOCK_SEQPACKET) failed: %m, trying SOCK_DGRAM");
}
#endif
if (m_Pipe[0] < 0 && socketpair(AF_UNIX, SOCK_DGRAM, 0, m_Pipe) != 0) {
esyslog("streamdev-client: couldn't open section filter socket: %m");
}
else if(fcntl(m_Pipe[0], F_SETFL, O_NONBLOCK) != 0 ||
fcntl(m_Pipe[1], F_SETFL, O_NONBLOCK) != 0) {
esyslog("streamdev-client: couldn't set section filter socket to non-blocking mode: %m");
}
}
cStreamdevFilter::~cStreamdevFilter() {
Dprintf("~cStreamdevFilter %p\n", this);
if (m_Pipe[0] >= 0)
close(m_Pipe[0]);
// ownership of handle m_Pipe[0] has been transferred to VDR section handler
//if (m_Pipe[0] >= 0)
// close(m_Pipe[0]);
if (m_Pipe[1] >= 0)
close(m_Pipe[1]);
}
bool cStreamdevFilter::PutSection(const uchar *Data, int Length) {
bool cStreamdevFilter::PutSection(const uchar *Data, int Length, bool Pusi) {
if (!m_Used && !Pusi) /* wait for payload unit start indicator */
return true;
if (m_Used && Pusi) /* reset at payload unit start */
Reset();
if (m_Used + Length >= (int)sizeof(m_Buffer)) {
esyslog("ERROR: Streamdev: Section handler buffer overflow (%d bytes lost)",
Length);
m_Used = 0;
Reset();
return true;
}
memcpy(m_Buffer + m_Used, Data, Length);
m_Used += Length;
if (m_Used > 3) {
int length = (((m_Buffer[1] & 0x0F) << 8) | m_Buffer[2]) + 3;
if (m_Used == length) {
if (write(m_Pipe[1], m_Buffer, length) < 0)
return false;
m_Used = 0;
if (write(m_Pipe[1], m_Buffer, length) < 0) {
if(errno == EAGAIN || errno == EWOULDBLOCK)
dsyslog("cStreamdevFilter::PutSection socket overflow, "
"Pid %4d Tid %3d", m_Pid, m_Tid);
else
return false;
}
}
if (m_Used > length) {
dsyslog("cStreamdevFilter::PutSection: m_Used > length ! Pid %2d, Tid%2d "
"(len %3d, got %d/%d)", m_Pid, m_Tid, Length, m_Used, length);
if(Length < TS_SIZE-5) {
// TS packet not full -> this must be last TS packet of section data -> safe to reset now
Reset();
}
}
}
return true;
}
void cStreamdevFilter::Reset(void) {
if(m_Used)
dsyslog("cStreamdevFilter::Reset skipping %d bytes", m_Used);
m_Used = 0;
}
bool cStreamdevFilter::IsClosed(void) {
char m_Buffer[3] = {0,0,0}; /* tid 0, 0 bytes */
// Test if pipe/socket has been closed by writing empty section
if (write(m_Pipe[1], m_Buffer, 3) < 0 &&
errno != EAGAIN &&
errno != EWOULDBLOCK) {
if (errno != ECONNREFUSED &&
errno != ECONNRESET &&
errno != EPIPE)
esyslog("cStreamdevFilter::IsClosed failed: %m");
return true;
}
return false;
}
// --- cStreamdevFilters -----------------------------------------------------
cStreamdevFilters::cStreamdevFilters(void):
cThread("streamdev-client: sections assembler") {
m_Active = false;
m_RingBuffer = new cRingBufferLinear(MEGABYTE(1), TS_SIZE * 2, true);
Start();
m_TSBuffer = NULL;
}
cStreamdevFilters::~cStreamdevFilters() {
if (m_Active) {
m_Active = false;
Cancel(3);
}
delete m_RingBuffer;
SetConnection(-1);
}
int cStreamdevFilters::OpenFilter(u_short Pid, u_char Tid, u_char Mask) {
CarbageCollect();
cStreamdevFilter *f = new cStreamdevFilter(Pid, Tid, Mask);
int fh = f->ReadPipe();
Lock();
Add(f);
return f->ReadPipe();
Unlock();
return fh;
}
cStreamdevFilter *cStreamdevFilters::Matches(u_short Pid, u_char Tid) {
for (cStreamdevFilter *f = First(); f; f = Next(f)) {
if (f->Matches(Pid, Tid))
return f;
void cStreamdevFilters::CarbageCollect(void) {
LOCK_THREAD;
for (cStreamdevFilter *fi = First(); fi;) {
if (fi->IsClosed()) {
if (errno == ECONNREFUSED ||
errno == ECONNRESET ||
errno == EPIPE) {
ClientSocket.SetFilter(fi->Pid(), fi->Tid(), fi->Mask(), false);
Dprintf("cStreamdevFilters::CarbageCollector: filter closed: Pid %4d, Tid %3d, Mask %2x (%d filters left)",
(int)fi->Pid(), (int)fi->Tid(), fi->Mask(), Count()-1);
cStreamdevFilter *next = Prev(fi);
Del(fi);
fi = next ? Next(next) : First();
} else {
esyslog("cStreamdevFilters::CarbageCollector() error: "
"Pid %4d, Tid %3d, Mask %2x (%d filters left) failed",
(int)fi->Pid(), (int)fi->Tid(), fi->Mask(), Count()-1);
LOG_ERROR;
fi = Next(fi);
}
} else {
fi = Next(fi);
}
}
return NULL;
}
void cStreamdevFilters::Put(const uchar *Data) {
int p = m_RingBuffer->Put(Data, TS_SIZE);
if (p != TS_SIZE)
m_RingBuffer->ReportOverflow(TS_SIZE - p);
bool cStreamdevFilters::ReActivateFilters(void)
{
LOCK_THREAD;
bool res = true;
CarbageCollect();
for (cStreamdevFilter *fi = First(); fi; fi = Next(fi)) {
res = ClientSocket.SetFilter(fi->Pid(), fi->Tid(), fi->Mask(), true) && res;
Dprintf("ReActivateFilters(%d, %d, %d) -> %s", fi->Pid(), fi->Tid(), fi->Mask(), res ? "Ok" :"FAIL");
}
return res;
}
void cStreamdevFilters::SetConnection(int Handle) {
Cancel(2);
DELETENULL(m_TSBuffer);
if (Handle >= 0) {
m_TSBuffer = new cTSBuffer(Handle, MEGABYTE(1), 1);
ReActivateFilters();
Start();
}
}
void cStreamdevFilters::Action(void) {
m_Active = true;
while (m_Active) {
int recvd;
const uchar *block = m_RingBuffer->Get(recvd);
int fails = 0;
if (block && recvd > 0) {
cStreamdevFilter *f;
while (Running()) {
const uchar *block = m_TSBuffer->Get();
if (block) {
u_short pid = (((u_short)block[1] & PID_MASK_HI) << 8) | block[2];
u_char tid = block[3];
if ((f = Matches(pid, tid)) != NULL) {
int len = block[4];
if (!f->PutSection(block + 5, len)) {
if (errno != EPIPE) {
esyslog("streamdev-client: couldn't send section packet: %m");
bool Pusi = block[1] & 0x40;
int len = block[4];
#if 0
if (block[1] == 0xff &&
block[2] == 0xff &&
block[3] == 0xff &&
block[4] == 0x7f)
isyslog("*********** TRANSPONDER -> %s **********", block+5);
#endif
LOCK_THREAD;
cStreamdevFilter *f = First();
while (f) {
cStreamdevFilter *next = Next(f);
if (f->Matches(pid, tid)) {
if (f->PutSection(block + 5, len, Pusi))
break;
if (errno != ECONNREFUSED &&
errno != ECONNRESET &&
errno != EPIPE) {
Dprintf("FATAL ERROR: %m\n");
esyslog("streamdev-client: couldn't send section packet: %m");
}
ClientSocket.SetFilter(f->Pid(), f->Tid(), f->Mask(), false);
Del(f);
// Filter was closed.
// - need to check remaining filters for another match
}
f = next;
}
m_RingBuffer->Del(TS_SIZE);
} else
usleep(1);
} else {
#if 1 // TODO: this should be fixed in vdr cTSBuffer
// Check disconnection
int fd = *ClientSocket.DataSocket(siLiveFilter);
if(fd < 0)
break;
cPoller Poller(fd);
if (Poller.Poll()) {
char tmp[1];
errno = 0;
Dprintf("cStreamdevFilters::Action(): checking connection");
if (recv(fd, tmp, 1, MSG_PEEK) == 0 && errno != EAGAIN) {
++fails;
if (fails >= 10) {
esyslog("cStreamdevFilters::Action(): stream disconnected ?");
ClientSocket.CloseDataConnection(siLiveFilter);
break;
}
} else {
fails = 0;
}
} else {
fails = 0;
}
cCondWait::SleepMs(10);
#endif
}
}
DELETENULL(m_TSBuffer);
dsyslog("StreamdevFilters::Action() ended");
}
#endif // VDRVERSNUM >= 10300

View File

@@ -1,5 +1,5 @@
/*
* $Id: filter.h,v 1.1.1.1 2004/12/30 22:44:04 lordjaxom Exp $
* $Id: filter.h,v 1.4 2007/04/24 11:23:16 schmirl Exp $
*/
#ifndef VDR_STREAMDEV_FILTER_H
@@ -12,52 +12,25 @@
#include <vdr/tools.h>
#include <vdr/thread.h>
class cRingBufferFrame;
class cRingBufferLinear;
class cStreamdevFilter: public cListObject {
private:
uchar m_Buffer[4096];
int m_Used;
int m_Pipe[2];
u_short m_Pid;
u_char m_Tid;
u_char m_Mask;
cRingBufferFrame *m_RingBuffer;
public:
cStreamdevFilter(u_short Pid, u_char Tid, u_char Mask);
virtual ~cStreamdevFilter();
bool Matches(u_short Pid, u_char Tid);
bool PutSection(const uchar *Data, int Length);
int ReadPipe(void) const { return m_Pipe[0]; }
u_short Pid(void) const { return m_Pid; }
u_char Tid(void) const { return m_Tid; }
u_char Mask(void) const { return m_Mask; }
};
inline bool cStreamdevFilter::Matches(u_short Pid, u_char Tid) {
return m_Pid == Pid && m_Tid == (Tid & m_Mask);
}
class cTSBuffer;
class cStreamdevFilter;
class cStreamdevFilters: public cList<cStreamdevFilter>, public cThread {
private:
bool m_Active;
cRingBufferLinear *m_RingBuffer;
cTSBuffer *m_TSBuffer;
protected:
virtual void Action(void);
void CarbageCollect(void);
bool ReActivateFilters(void);
public:
cStreamdevFilters(void);
virtual ~cStreamdevFilters();
void SetConnection(int Handle);
int OpenFilter(u_short Pid, u_char Tid, u_char Mask);
cStreamdevFilter *Matches(u_short Pid, u_char Tid);
void Put(const uchar *Data);
};
# endif // VDRVERSNUM >= 10300

View File

@@ -1,5 +1,5 @@
/*
* $Id: socket.c,v 1.7 2007/01/15 11:45:48 schmirl Exp $
* $Id: socket.c,v 1.8 2007/04/24 10:57:34 schmirl Exp $
*/
#include <tools/select.h>
@@ -215,6 +215,24 @@ bool cClientSocket::CreateDataConnection(eSocketId Id) {
return true;
}
bool cClientSocket::CloseDataConnection(eSocketId Id) {
//if (!CheckConnection()) return false;
CMD_LOCK;
if(Id == siLive || Id == siLiveFilter)
if (m_DataSockets[Id] != NULL) {
std::string command = (std::string)"ABRT " + (const char*)itoa(Id);
if (!Command(command, 220)) {
if (errno == 0)
esyslog("ERROR: Streamdev: Couldn't cleanly close data connection");
//return false;
}
DELETENULL(m_DataSockets[Id]);
}
return true;
}
bool cClientSocket::SetChannelDevice(const cChannel *Channel) {
if (!CheckConnection()) return false;

View File

@@ -1,5 +1,5 @@
/*
* $Id: socket.h,v 1.3 2005/02/08 17:22:35 lordjaxom Exp $
* $Id: socket.h,v 1.4 2007/04/24 10:57:34 schmirl Exp $
*/
#ifndef VDR_STREAMDEV_CLIENT_CONNECTION_H
@@ -47,6 +47,7 @@ public:
bool CheckConnection(void);
bool ProvidesChannel(const cChannel *Channel, int Priority);
bool CreateDataConnection(eSocketId Id);
bool CloseDataConnection(eSocketId Id);
bool SetChannelDevice(const cChannel *Channel);
bool SetPid(int Pid, bool On);
#if VDRVERSNUM >= 10300