Implemented remuxing of recordings (closes #1892)

9 files changed, 138 insertions, 67 deletions

diff --git a/server/connectionHTTP.c b/server/connectionHTTP.c
index bcb4b58..517fb2e 100644
--- a/server/connectionHTTP.c
+++ b/server/connectionHTTP.c
@@ -35,6 +35,7 @@ cConnectionHTTP::cConnectionHTTP(void):
 
 cConnectionHTTP::~cConnectionHTTP() 
 {
+	SetStreamer(NULL);
 	delete m_RecPlayer;
 	delete m_MenuList;
 }
@@ -164,6 +165,14 @@ bool cConnectionHTTP::ProcessRequest(void)
 		}
 	}
 
+	tStrStrMap::const_iterator it;
+	it = m_Params.find("apid");
+	if (it != m_Params.end())
+		m_Apid[0] = atoi(it->second.c_str());
+	it = m_Params.find("dpid");
+	if (it != m_Params.end())
+		m_Dpid[0] = atoi(it->second.c_str());
+
 	tStrStrMap::const_iterator it_method = Headers().find(REQUEST_METHOD);
 	tStrStrMap::const_iterator it_pathinfo = Headers().find(PATH_INFO);
 	if (it_method == Headers().end() || it_pathinfo == Headers().end()) {
@@ -196,30 +205,46 @@ bool cConnectionHTTP::ProcessRequest(void)
 		}
 		else if (m_RecPlayer != NULL) {
 			Dprintf("GET recording\n");
+			bool isPes = m_RecPlayer->getCurrentRecording()->IsPesRecording();
+			// no remuxing for old PES recordings
+			if (isPes && m_StreamType != stPES)
+				return HttpResponse(503, true);
+
 			int64_t from, to;
 			bool hasRange = ParseRange(from, to);
 
 			cStreamdevRecStreamer* recStreamer;
 			if (from == 0 && hasRange && m_ReplayFakeRange) {
-				recStreamer = new cStreamdevRecStreamer(m_RecPlayer, this);
+				recStreamer = new cStreamdevRecStreamer(this, m_RecPlayer, m_StreamType, (int64_t) 0L, m_Apid[0] ? m_Apid : NULL, m_Dpid[0] ? m_Dpid : NULL);
 				from += m_ReplayPos;
 				if (to >= 0)
 					to += m_ReplayPos;
 			}
 			else
-				recStreamer = new cStreamdevRecStreamer(m_RecPlayer, this, m_ReplayPos);
+				recStreamer = new cStreamdevRecStreamer(this, m_RecPlayer, m_StreamType, m_ReplayPos, m_Apid[0] ? m_Apid : NULL, m_Dpid[0] ? m_Dpid : NULL);
 			SetStreamer(recStreamer);
+
+			if (m_StreamType == stEXT)
+				return Respond("HTTP/1.0 200 OK");
+			else if (m_StreamType == stES && (m_Apid[0] || m_Dpid[0]))
+				return HttpResponse(200, false, "audio/mpeg");
+
+			const char* contentType = (isPes || m_RecPlayer->getPatPmtData()->Vpid()) ? "video/mpeg" : "audio/mpeg";
+			// range not supported when remuxing
+			if (m_StreamType != stTS && !isPes)
+				return HttpResponse(200, false, contentType);
+
 			uint64_t total = recStreamer->GetLength();
 			if (hasRange) {
 				int64_t length = recStreamer->SetRange(from, to);
 				Dprintf("range response: %lld-%lld/%lld, len %lld\n", (long long)from, (long long)to, (long long)total, (long long)length);
 				if (length < 0L)
-					return HttpResponse(416, true, "video/mpeg", "Accept-Ranges: bytes\r\nContent-Range: bytes */%llu", (unsigned long long) total);
+					return HttpResponse(416, true, contentType, "Accept-Ranges: bytes\r\nContent-Range: bytes */%llu", (unsigned long long) total);
 				else
-					return HttpResponse(206, false, "video/mpeg", "Accept-Ranges: bytes\r\nContent-Range: bytes %lld-%lld/%llu\r\nContent-Length: %lld", (long long) from, (long long) to, (unsigned long long) total, (long long) length);
+					return HttpResponse(206, false, contentType, "Accept-Ranges: bytes\r\nContent-Range: bytes %lld-%lld/%llu\r\nContent-Length: %lld", (long long) from, (long long) to, (unsigned long long) total, (long long) length);
 			}
 			else
-				return HttpResponse(200, false, "video/mpeg", "Accept-Ranges: bytes");
+				return HttpResponse(200, false, contentType, "Accept-Ranges: bytes");
 		}
 		else {
 			return HttpResponse(404, true);
@@ -247,29 +272,45 @@ bool cConnectionHTTP::ProcessRequest(void)
 		}
 		else if (m_RecPlayer != NULL) {
 			Dprintf("HEAD recording\n");
+			bool isPes = m_RecPlayer->getCurrentRecording()->IsPesRecording();
+			// no remuxing for old PES recordings
+			if (isPes && m_StreamType != stPES)
+				return HttpResponse(503, true);
+
 			int64_t from, to;
 			bool hasRange = ParseRange(from, to);
 			
 			cStreamdevRecStreamer* recStreamer;
 			if (from == 0 && hasRange && m_ReplayFakeRange) {
-				recStreamer = new cStreamdevRecStreamer(m_RecPlayer, this, m_ReplayPos);
+				recStreamer = new cStreamdevRecStreamer(this, m_RecPlayer, m_StreamType, m_ReplayPos, m_Apid[0] ? m_Apid : NULL, m_Dpid[0] ? m_Dpid : NULL);
 				from += m_ReplayPos;
 				if (to >= 0)
 					to += m_ReplayPos;
 			}
 			else
-				recStreamer = new cStreamdevRecStreamer(m_RecPlayer, this, m_ReplayPos);
+				recStreamer = new cStreamdevRecStreamer(this, m_RecPlayer, m_StreamType, m_ReplayPos, m_Apid[0] ? m_Apid : NULL, m_Dpid[0] ? m_Dpid : NULL);
 			SetStreamer(recStreamer);
+
+			if (m_StreamType == stEXT)
+				return Respond("HTTP/1.0 200 OK");
+			else if (m_StreamType == stES && (m_Apid[0] || m_Dpid[0]))
+				return HttpResponse(200, true, "audio/mpeg");
+
+			const char* contentType = (isPes || m_RecPlayer->getPatPmtData()->Vpid()) ? "video/mpeg" : "audio/mpeg";
+			// range not supported when remuxing
+			if (m_StreamType != stTS && !isPes)
+				return HttpResponse(200, false, contentType);
+
 			uint64_t total = recStreamer->GetLength();
 			if (hasRange) {
 				int64_t length = recStreamer->SetRange(from, to);
 				if (length < 0L)
-					return HttpResponse(416, true, "video/mpeg", "Accept-Ranges: bytes\r\nContent-Range: bytes */%llu", (unsigned long long) total);
+					return HttpResponse(416, true, contentType, "Accept-Ranges: bytes\r\nContent-Range: bytes */%llu", (unsigned long long) total);
 				else
-					return HttpResponse(206, true, "video/mpeg", "Accept-Ranges: bytes\r\nContent-Range: bytes %lld-%lld/%llu\r\nContent-Length: %lld", (long long) from, (long long) to, (unsigned long long) total, (long long) length);
+					return HttpResponse(206, true, contentType, "Accept-Ranges: bytes\r\nContent-Range: bytes %lld-%lld/%llu\r\nContent-Length: %lld", (long long) from, (long long) to, (unsigned long long) total, (long long) length);
 			}
 			else
-				return HttpResponse(200, true, "video/mpeg", "Accept-Ranges: bytes");
+				return HttpResponse(200, true, contentType, "Accept-Ranges: bytes");
 		}
 		else {
 			return HttpResponse(404, true);
diff --git a/server/livestreamer.c b/server/livestreamer.c
index 8dce11d..41befe4 100644
--- a/server/livestreamer.c
+++ b/server/livestreamer.c
@@ -9,7 +9,6 @@
 #include "remux/extern.h"
 
 #include <vdr/transfer.h>
-#include <vdr/ringbuffer.h>
 
 #include "server/livestreamer.h"
 #include "server/setup.h"
@@ -339,23 +338,21 @@ cStreamdevLiveStreamer::cStreamdevLiveStreamer(const cServerConnection *Connecti
 		cStreamdevStreamer("streamdev-livestreaming", Connection),
 		m_Priority(Priority),
 		m_NumPids(0),
-		m_StreamType(StreamType),
 		m_Channel(Channel),
 		m_Device(NULL),
 		m_Receiver(NULL),
 		m_PatFilter(NULL),
-		m_Remux(NULL),
 		m_SwitchLive(false)
 {
 		m_ReceiveBuffer = new cStreamdevBuffer(LIVEBUFSIZE, TS_SIZE *2, true, "streamdev-livestreamer"),
 		m_ReceiveBuffer->SetTimeouts(0, 100);
 		if (Priority == IDLEPRIORITY) {
-			SetChannel(Apid, Dpid);
+			SetChannel(StreamType, Apid, Dpid);
 		}
 		else {
 			m_Device = SwitchDevice(Channel, Priority);
 			if (m_Device)
-				SetChannel(Apid, Dpid);
+				SetChannel(StreamType, Apid, Dpid);
 			memcpy(m_Caids,Channel->Caids(),sizeof(m_Caids));
 		}
 }
@@ -366,7 +363,6 @@ cStreamdevLiveStreamer::~cStreamdevLiveStreamer()
 	Stop();
 	DELETENULL(m_PatFilter);
 	DELETENULL(m_Receiver);
-	delete m_Remux;
 	delete m_ReceiveBuffer;
 }
 
@@ -488,7 +484,7 @@ void cStreamdevLiveStreamer::StartReceiver(bool Force)
 		DELETENULL(m_Receiver);
 }
 
-bool cStreamdevLiveStreamer::SetChannel(const int* Apid, const int *Dpid) 
+bool cStreamdevLiveStreamer::SetChannel(eStreamType StreamType, const int* Apid, const int *Dpid) 
 {
 	Dprintf("Initializing Remuxer for full channel transfer\n");
 	//printf("ca pid: %d\n", Channel->Ca());
@@ -496,7 +492,7 @@ bool cStreamdevLiveStreamer::SetChannel(const int* Apid, const int *Dpid)
 	const int *Apids = Apid ? Apid : m_Channel->Apids();
 	const int *Dpids = Dpid ? Dpid : m_Channel->Dpids();
 
-	switch (m_StreamType) {
+	switch (StreamType) {
 	case stES: 
 		{
 			int pid = ISRADIO(m_Channel) ? m_Channel->Apid(0) : m_Channel->Vpid();
@@ -504,22 +500,22 @@ bool cStreamdevLiveStreamer::SetChannel(const int* Apid, const int *Dpid)
 				pid = Apid[0];
 			else if (Dpid && Dpid[0])
 				pid = Dpid[0];
-			m_Remux = new cTS2ESRemux(pid);
+			SetRemux(new cTS2ESRemux(pid));
 			return SetPids(pid);
 		}
 
 	case stPES: 
-		m_Remux = new cTS2PESRemux(m_Channel->Vpid(), Apids, Dpids, m_Channel->Spids());
+		SetRemux(new cTS2PESRemux(m_Channel->Vpid(), Apids, Dpids, m_Channel->Spids()));
 		return SetPids(m_Channel->Vpid(), Apids, Dpids, m_Channel->Spids());
 
 #ifdef STREAMDEV_PS
 	case stPS:  
-		m_Remux = new cTS2PSRemux(m_Channel->Vpid(), Apids, Dpids, m_Channel->Spids());
+		SetRemux(new cTS2PSRemux(m_Channel->Vpid(), Apids, Dpids, m_Channel->Spids()));
 		return SetPids(m_Channel->Vpid(), Apids, Dpids, m_Channel->Spids());
 #endif
 
 	case stEXT:
-		m_Remux = new cExternRemux(Connection(), m_Channel, Apids, Dpids);
+		SetRemux(new cExternRemux(Connection(), m_Channel, Apids, Dpids));
 		// fall through
 	case stTS:
 		// This should never happen, but ...
@@ -574,36 +570,14 @@ int cStreamdevLiveStreamer::Put(const uchar *Data, int Count)
 		int siCount;
 		uchar *siData = m_PatFilter->Get(siCount);
 		if (siData) {
-			if (m_Remux)
-				siCount = m_Remux->Put(siData, siCount);
-			else
-				siCount = cStreamdevStreamer::Put(siData, siCount);
+			siCount = cStreamdevStreamer::Put(siData, siCount);
 			if (siCount)
 				m_PatFilter->Del(siCount);
 		}
 	}
-	if (m_Remux)
-		return m_Remux->Put(Data, Count);
-	else
-		return cStreamdevStreamer::Put(Data, Count);
-}
-
-uchar *cStreamdevLiveStreamer::Get(int &Count)
-{
-	if (m_Remux)
-		return m_Remux->Get(Count);
-	else
-		return cStreamdevStreamer::Get(Count);
+	return cStreamdevStreamer::Put(Data, Count);
 }
 
-void cStreamdevLiveStreamer::Del(int Count)
-{
-	if (m_Remux)
-		m_Remux->Del(Count);
-	else
-		cStreamdevStreamer::Del(Count);
-}
-	
 void cStreamdevLiveStreamer::Attach(void) 
 { 
 	Dprintf("cStreamdevLiveStreamer::Attach()\n");
diff --git a/server/livestreamer.h b/server/livestreamer.h
index 055de4d..b525c7b 100644
--- a/server/livestreamer.h
+++ b/server/livestreamer.h
@@ -11,9 +11,6 @@
 
 #define LIVEBUFSIZE (20000 * TS_SIZE)
 
-namespace Streamdev {
-	class cTSRemux;
-}
 class cStreamdevPatFilter;
 class cStreamdevLiveReceiver;
 
@@ -29,13 +26,11 @@ private:
 	int                     m_Pids[MAXRECEIVEPIDS + 1];
 	int                     m_NumPids;
 	int                     m_Caids[MAXCAIDS + 1];
-	eStreamType             m_StreamType;
 	const cChannel         *m_Channel;
 	cDevice                *m_Device;
 	cStreamdevLiveReceiver *m_Receiver;
 	cStreamdevBuffer       *m_ReceiveBuffer;
 	cStreamdevPatFilter    *m_PatFilter;
-	Streamdev::cTSRemux    *m_Remux;
 	bool                    m_SwitchLive;
 
 	void StartReceiver(bool Force = false);
@@ -48,7 +43,7 @@ private:
 	/* Find a suitable device and tune it to the requested channel. */
 	cDevice *SwitchDevice(const cChannel *Channel, int Priority);
 
-	bool SetChannel(const int* Apid = NULL, const int* Dpid = NULL);
+	bool SetChannel(eStreamType StreamType, const int* Apid = NULL, const int* Dpid = NULL);
 
 protected:
 	virtual uchar* GetFromReceiver(int &Count) { return m_ReceiveBuffer->Get(Count); }
@@ -72,9 +67,6 @@ public:
 	void Receive(uchar *Data, int Length);
 	virtual bool IsReceiving(void) const;
 
-	virtual uchar *Get(int &Count);
-	virtual void Del(int Count);
-
 	virtual void Attach(void);
 	virtual void Detach(void);
 
diff --git a/server/recplayer.c b/server/recplayer.c
index 3885b4c..1b60ef5 100644
--- a/server/recplayer.c
+++ b/server/recplayer.c
@@ -42,6 +42,12 @@ RecPlayer::RecPlayer(const char* FileName)
   if (!indexFile) esyslog("ERROR: Streamdev: Failed to create indexfile!");
 
   scan();
+
+  parser = new cPatPmtParser();
+  unsigned char buffer[2 * TS_SIZE];
+  unsigned long l = getBlock(buffer, 0UL, sizeof(buffer));
+  if (!l || !parser->ParsePatPmt(buffer, (int) l))
+    esyslog("ERROR: Streamdev: Failed to parse PAT/PMT");
 }
 
 void RecPlayer::scan()
@@ -87,6 +93,7 @@ RecPlayer::~RecPlayer()
   if (file) fclose(file);
   delete indexFile;
   delete recording;
+  delete parser;
 }
 
 int RecPlayer::openFile(int index)
diff --git a/server/recplayer.h b/server/recplayer.h
index e56c2cd..60100e8 100644
--- a/server/recplayer.h
+++ b/server/recplayer.h
@@ -23,6 +23,7 @@
 
 #include <stdio.h>
 #include <vdr/recording.h>
+#include <vdr/remux.h>
 
 #include "server/streamer.h"
 
@@ -44,6 +45,7 @@ class RecPlayer
     int openFile(int index);
     uint64_t getLastPosition();
     cRecording* getCurrentRecording();
+    const cPatPmtParser* getPatPmtData() { return parser; }
     void scan();
     uint64_t positionFromResume(int ResumeID);
     uint64_t positionFromMark(int MarkIndex);
@@ -56,6 +58,7 @@ class RecPlayer
   private:
     cRecording* recording;
     cIndexFile* indexFile;
+    cPatPmtParser* parser;
     FILE* file;
     int fileOpen;
     Segment* segments[1000];
diff --git a/server/recstreamer.c b/server/recstreamer.c
index bd01b96..1ee0b25 100644
--- a/server/recstreamer.c
+++ b/server/recstreamer.c
@@ -12,7 +12,7 @@ using namespace Streamdev;
 
 // --- cStreamdevRecStreamer -------------------------------------------------
 
-cStreamdevRecStreamer::cStreamdevRecStreamer(RecPlayer *RecPlayer, const cServerConnection *Connection, int64_t StartOffset):
+cStreamdevRecStreamer::cStreamdevRecStreamer(const cServerConnection *Connection, RecPlayer *RecPlayer, eStreamType StreamType, int64_t StartOffset, const int *Apid, const int *Dpid):
 		cStreamdevStreamer("streamdev-recstreaming", Connection),
 		m_RecPlayer(RecPlayer),
 		m_StartOffset(StartOffset),
@@ -20,6 +20,36 @@ cStreamdevRecStreamer::cStreamdevRecStreamer(RecPlayer *RecPlayer, const cServer
 {
 	Dprintf("New rec streamer\n");
 	m_To = (int64_t) m_RecPlayer->getLengthBytes() - StartOffset - 1;
+
+	const cPatPmtParser *parser = RecPlayer->getPatPmtData();
+	const int *Apids = Apid ? Apid : parser->Apids();
+	const int *Dpids = Dpid ? Dpid : parser->Dpids();
+	switch (StreamType) {
+	case stES:
+		{
+			int pid = parser->Vpid();
+			if (Apid && Apid[0])
+				pid = Apid[0];
+			else if (Dpid && Dpid[0])
+				pid = Dpid[0];
+			SetRemux(new cTS2ESRemux(pid));
+		}
+		break;
+	case stPES:
+		if (!m_RecPlayer->getCurrentRecording()->IsPesRecording())
+			SetRemux(new cTS2PESRemux(parser->Vpid(), Apids, Dpids, parser->Spids()));
+		break;
+#ifdef STREAMDEV_PS
+	case stPS:
+		SetRemux(new cTS2PSRemux(parser->Vpid(), Apids, Dpids, parser->Spids()));
+		break;
+#endif
+	case stEXT:
+		SetRemux(new cExternRemux(Connection, parser, Apids, Dpids));
+                break;
+	default:
+		break;
+	}
 }
 
 cStreamdevRecStreamer::~cStreamdevRecStreamer() 
diff --git a/server/recstreamer.h b/server/recstreamer.h
index ee4b120..68f0c17 100644
--- a/server/recstreamer.h
+++ b/server/recstreamer.h
@@ -1,6 +1,7 @@
 #ifndef VDR_STREAMDEV_RECSTREAMER_H
 #define VDR_STREAMDEV_RECSTREAMER_H
 
+#include "common.h"
 #include "server/streamer.h"
 #include "server/recplayer.h"
 
@@ -26,7 +27,7 @@ public:
 	uint64_t GetLength() { return m_RecPlayer->getLengthBytes() - m_StartOffset; }
 	int64_t SetRange(int64_t &From, int64_t &To);
 	virtual cString ToText() const;
-	cStreamdevRecStreamer(RecPlayer *RecPlayer, const cServerConnection *Connection, int64_t StartOffset = 0L);
+	cStreamdevRecStreamer(const cServerConnection *Connection, RecPlayer *RecPlayer, eStreamType StreamType, int64_t StartOffset = 0L, const int *Apids = NULL, const int *Dpids = NULL);
 	virtual ~cStreamdevRecStreamer();
 };
 
diff --git a/server/streamer.c b/server/streamer.c
index 16411ed..d1a9b3e 100644
--- a/server/streamer.c
+++ b/server/streamer.c
@@ -8,7 +8,6 @@
 #include <unistd.h>
 
 #include "server/streamer.h"
-#include "server/suspend.h"
 #include "tools/socket.h"
 #include "tools/select.h"
 #include "common.h"
@@ -108,21 +107,37 @@ void cStreamdevWriter::Action(void)
 	Dprintf("Max. Transmit Blocksize was: %d\n", max);
 }
 
+// --- cRemuxDummy ------------------------------------------------------------
+
+class cRemuxDummy: public Streamdev::cTSRemux {
+private:
+	cStreamdevBuffer	m_Buffer;
+public:
+	cRemuxDummy();
+	virtual int Put(const uchar *Data, int Count) { return m_Buffer.Put(Data, Count); }
+	virtual uchar *Get(int& Count) { return m_Buffer.Get(Count); }
+	virtual void Del(int Count) { return m_Buffer.Del(Count); }
+};
+
+cRemuxDummy::cRemuxDummy(): m_Buffer(WRITERBUFSIZE, TS_SIZE * 2)
+{
+	m_Buffer.SetTimeouts(100, 100);
+}
+
 // --- cStreamdevStreamer -----------------------------------------------------
 
 cStreamdevStreamer::cStreamdevStreamer(const char *Name, const cServerConnection *Connection):
 		cThread(Name),
 		m_Connection(Connection),
-		m_Writer(NULL),
-		m_SendBuffer(new cStreamdevBuffer(WRITERBUFSIZE, TS_SIZE * 2))
+		m_Remux(new cRemuxDummy()),
+		m_Writer(NULL)
 {
-	m_SendBuffer->SetTimeouts(100, 100);
 }
 
 cStreamdevStreamer::~cStreamdevStreamer() 
 {
 	Dprintf("Desctructing streamer\n");
-	delete m_SendBuffer;
+	delete m_Remux;
 }
 
 void cStreamdevStreamer::Start(cTBSocket *Socket) 
@@ -163,3 +178,7 @@ void cStreamdevStreamer::Action(void)
 	}
 }
 
+int cStreamdevStreamer::Put(const uchar *Data, int Count) {
+	return m_Remux->Put(Data, Count);
+}
+
diff --git a/server/streamer.h b/server/streamer.h
index d9b2998..af69cbe 100644
--- a/server/streamer.h
+++ b/server/streamer.h
@@ -9,6 +9,8 @@
 #include <vdr/ringbuffer.h>
 #include <vdr/tools.h>
 
+#include "remux/tsremux.h"
+
 class cTBSocket;
 class cStreamdevStreamer;
 class cServerConnection;
@@ -65,16 +67,18 @@ public:
 class cStreamdevStreamer: public cThread {
 private:
 	const cServerConnection *m_Connection;
-	cStreamdevWriter  *m_Writer;
-	cStreamdevBuffer  *m_SendBuffer;
+	Streamdev::cTSRemux     *m_Remux;
+	cStreamdevWriter        *m_Writer;
+	cStreamdevBuffer        *m_SendBuffer;
 
 protected:
 	virtual uchar* GetFromReceiver(int &Count) = 0;
 	virtual void DelFromReceiver(int Count) = 0;
-	virtual int Put(const uchar *Data, int Count) { return m_SendBuffer->PutTS(Data, Count); }
+	virtual int Put(const uchar *Data, int Count);
 	virtual void Action(void);
 
 	bool IsRunning(void) const { return m_Writer; }
+	void SetRemux(Streamdev::cTSRemux *Remux) { delete m_Remux; m_Remux = Remux; }
 
 public:
 	cStreamdevStreamer(const char *Name, const cServerConnection *Connection = NULL);
@@ -87,8 +91,8 @@ public:
 	virtual bool IsReceiving(void) const = 0;
 	bool Abort(void);
 
-	virtual uchar *Get(int &Count) { return m_SendBuffer->Get(Count); }
-	virtual void Del(int Count) { m_SendBuffer->Del(Count); }
+	uchar *Get(int &Count) { return m_Remux->Get(Count); }
+	void Del(int Count) { m_Remux->Del(Count); } 
 
 	virtual void Detach(void) {}
 	virtual void Attach(void) {}