Basic support for HTTP streaming of recordings

CONTRIBUTORS

@@ -208,3 +208,6 @@ Methodus
 
 Uwe
   for reporting a compiler error in client/device.c with VDR < 1.7.22
+
+Chris Tallon
+  for his kind permission to use VOMP's recplayer for replaying recordings

HISTORY

@@ -1,6 +1,7 @@
 VDR Plugin 'streamdev' Revision History
 ---------------------------------------
  
+- Basic support for HTTP streaming of recordings
 - Close writer when streamer is finished
 - Don't abort VTP connection if filter stream is broken
 - Restructured cStreamdevStreamer: Moved inbound buffer into actual subclass.

server/Makefile

@@ -22,7 +22,7 @@ SERVEROBJS = $(PLUGIN).o \
 	componentVTP.o connectionVTP.o \
 	componentHTTP.o connectionHTTP.o menuHTTP.o \
 	componentIGMP.o connectionIGMP.o \
-	streamer.o livestreamer.o livefilter.o recplayer.o \
+	streamer.o livestreamer.o livefilter.o recstreamer.o recplayer.o \
 	menu.o suspend.o setup.o
 	
 ### The main target:

server/connectionHTTP.c

@@ -5,6 +5,8 @@
 #include <ctype.h>
 #include <time.h>
 #include <stdarg.h>
+#include <vdr/thread.h>
+#include <vdr/recording.h>
  
 #include "server/connectionHTTP.h"
 #include "server/menuHTTP.h"
@@ -14,9 +16,10 @@
 cConnectionHTTP::cConnectionHTTP(void): 
 		cServerConnection("HTTP"),
 		m_Status(hsRequest),
-		m_LiveStreamer(NULL),
-		m_Channel(NULL),
+		m_Streamer(NULL),
 		m_StreamType((eStreamType)StreamdevServerSetup.HTTPStreamType),
+		m_Channel(NULL),
+		m_Recording(NULL),
 		m_ChannelList(NULL)
 {
 	Dprintf("constructor hsRequest\n");
@@ -26,7 +29,8 @@ cConnectionHTTP::cConnectionHTTP(void):
 
 cConnectionHTTP::~cConnectionHTTP() 
 {
-	delete m_LiveStreamer;
+	delete m_Streamer;
+	delete m_Recording;
 }
 
 bool cConnectionHTTP::CanAuthenticate(void)
@@ -168,9 +172,10 @@ bool cConnectionHTTP::ProcessRequest(void)
 				device = GetDevice(m_Channel, StreamdevServerSetup.HTTPPriority);
 			if (device != NULL) {
 				device->SwitchChannel(m_Channel, false);
-				m_LiveStreamer = new cStreamdevLiveStreamer(StreamdevServerSetup.HTTPPriority, this);
-				if (m_LiveStreamer->SetChannel(m_Channel, m_StreamType, m_Apid[0] ? m_Apid : NULL, m_Dpid[0] ? m_Dpid : NULL)) {
-					m_LiveStreamer->SetDevice(device);
+				cStreamdevLiveStreamer* liveStreamer = new cStreamdevLiveStreamer(StreamdevServerSetup.HTTPPriority, this);
+				m_Streamer = liveStreamer;
+				if (liveStreamer->SetChannel(m_Channel, m_StreamType, m_Apid[0] ? m_Apid : NULL, m_Dpid[0] ? m_Dpid : NULL)) {
+					liveStreamer->SetDevice(device);
 					if (!SetDSCP())
 						LOG_ERROR_STR("unable to set DSCP sockopt");
 					if (m_StreamType == stEXT) {
@@ -183,10 +188,26 @@ bool cConnectionHTTP::ProcessRequest(void)
 						return HttpResponse(200, false, "video/mpeg");
 					}
 				}
-				DELETENULL(m_LiveStreamer);
+				DELETENULL(m_Streamer);
 			}
 			return HttpResponse(503, true);
 		}
+		else if (m_Recording != NULL) {
+			Dprintf("GET recording\n");
+			cStreamdevRecStreamer* recStreamer = new cStreamdevRecStreamer(m_Recording, this);
+			m_Streamer = recStreamer;
+			int64_t from, to;
+			uint64_t total = recStreamer->GetLength();
+			if (ParseRange(from, to)) {
+				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);
+				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);
+			}
+			else
+				return HttpResponse(200, false, "video/mpeg", "Accept-Ranges: bytes");
+		}
 		else {
 			return HttpResponse(404, true);
 		}
@@ -198,8 +219,9 @@ bool cConnectionHTTP::ProcessRequest(void)
 		else if (m_Channel != NULL) {
 			if (ProvidesChannel(m_Channel, StreamdevServerSetup.HTTPPriority)) {
 				if (m_StreamType == stEXT) {
-					m_LiveStreamer = new cStreamdevLiveStreamer(StreamdevServerSetup.HTTPPriority, this);
-					m_LiveStreamer->SetChannel(m_Channel, m_StreamType, m_Apid[0] ? m_Apid : NULL, m_Dpid[0] ? m_Dpid : NULL);
+					cStreamdevLiveStreamer *liveStreamer = new cStreamdevLiveStreamer(StreamdevServerSetup.HTTPPriority, this);
+					liveStreamer->SetChannel(m_Channel, m_StreamType, m_Apid[0] ? m_Apid : NULL, m_Dpid[0] ? m_Dpid : NULL);
+					m_Streamer = liveStreamer;
 					return Respond("HTTP/1.0 200 OK");
 				} else if (m_StreamType == stES && (m_Apid[0] || m_Dpid[0] || ISRADIO(m_Channel))) {
 					return HttpResponse(200, true, "audio/mpeg", "icy-name: %s", m_Channel->Name());
@@ -211,6 +233,22 @@ bool cConnectionHTTP::ProcessRequest(void)
 			}
 			return HttpResponse(503, true);
 		}
+		else if (m_Recording != NULL) {
+			Dprintf("HEAD recording\n");
+			cStreamdevRecStreamer *recStreamer = new cStreamdevRecStreamer(m_Recording, this);
+			m_Streamer = recStreamer;
+			int64_t from, to;
+			uint64_t total = recStreamer->GetLength();
+			if (ParseRange(from, to)) {
+				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);
+				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);
+			}
+			else
+				return HttpResponse(200, true, "video/mpeg", "Accept-Ranges: bytes");
+		}
 		else {
 			return HttpResponse(404, true);
 		}
@@ -244,9 +282,11 @@ bool cConnectionHTTP::HttpResponse(int Code, bool Last, const char* ContentType,
 	switch (Code)
 	{
 		case 200: rc = Respond("HTTP/1.1 200 OK"); break;
+		case 206: rc = Respond("HTTP/1.1 206 Partial Content"); break;
 		case 400: rc = Respond("HTTP/1.1 400 Bad Request"); break;
 		case 401: rc = Respond("HTTP/1.1 401 Authorization Required"); break;
 		case 404: rc = Respond("HTTP/1.1 404 Not Found"); break;
+		case 416: rc = Respond("HTTP/1.1 416 Requested range not satisfiable"); break;
 		case 503: rc = Respond("HTTP/1.1 503 Service Unavailable"); break;
 		default:  rc = Respond("HTTP/1.1 500 Internal Server Error");
 	}
@@ -279,6 +319,40 @@ bool cConnectionHTTP::HttpResponse(int Code, bool Last, const char* ContentType,
 	return rc && Respond("");
 }
 
+bool cConnectionHTTP::ParseRange(int64_t &From, int64_t &To) const
+{
+	const static std::string RANGE("HTTP_RANGE");
+	From = To = 0L;
+	tStrStrMap::const_iterator it = Headers().find(RANGE);
+	if (it != Headers().end()) {
+		size_t b = it->second.find("bytes=");
+		if (b != std::string::npos) {
+			char* e = NULL;
+			const char* r = it->second.c_str() + b + sizeof("bytes=") - 1;
+			if (strchr(r, ',') != NULL)
+				esyslog("streamdev-server cConnectionHTTP::GetRange: Multi-ranges not supported");
+			From = strtol(r, &e, 10);
+			if (r != e) {
+				if (From < 0L) {
+					To = -1L;
+					return *e == 0 || *e == ',';
+				}
+				else if (*e == '-') {
+					r = e + 1;
+					if (*r == 0 || *e == ',') {
+						To = -1L;
+						return true;
+					}
+					To = strtol(r, &e, 10);
+					return r != e && To >= From &&
+							(*e == 0 || *e == ',');
+				}
+			}
+		}
+	}
+	return false;
+}
+
 void cConnectionHTTP::Flushed(void) 
 {
 	if (m_Status != hsBody)
@@ -296,9 +370,9 @@ void cConnectionHTTP::Flushed(void)
 		}
 		return;
 	}
-	else if (m_Channel != NULL) {
+	else if (m_Streamer != NULL) {
 		Dprintf("streamer start\n");
-		m_LiveStreamer->Start(this);
+		m_Streamer->Start(this);
 		m_Status = hsFinished;
 	}
 	else {
@@ -401,6 +475,13 @@ bool cConnectionHTTP::ProcessURI(const std::string& PathInfo)
 	if ((m_ChannelList = ChannelListFromString(PathInfo.substr(1, file_pos), filespec.c_str(), fileext.c_str())) != NULL) {
 		Dprintf("Channel list requested\n");
 		return true;
+	} else if (strcmp(fileext.c_str(), ".rec") == 0) {
+		cThreadLock RecordingsLock(&Recordings);
+		cRecording* rec = Recordings.Get(atoi(filespec.c_str()) - 1);
+		Dprintf("Recording %s%s found\n", rec ? rec->Name() : filespec.c_str(), rec ? "" : " not");
+		if (rec)
+			m_Recording = new cRecording(rec->FileName());
+		return m_Recording != NULL;
 	} else if ((m_Channel = ChannelFromString(filespec.c_str(), &m_Apid[0], &m_Dpid[0])) != NULL) {
 		Dprintf("Channel found. Apid/Dpid is %d/%d\n", m_Apid[0], m_Dpid[0]);
 		return true;
@@ -411,5 +492,5 @@ bool cConnectionHTTP::ProcessURI(const std::string& PathInfo)
 cString cConnectionHTTP::ToText() const
 {
 	cString str = cServerConnection::ToText();
-	return m_LiveStreamer ? cString::sprintf("%s\t%s", *str, *m_LiveStreamer->ToText()) : str;
+	return m_Streamer ? cString::sprintf("%s\t%s", *str, *m_Streamer->ToText()) : str;
 }

server/connectionHTTP.h

@@ -7,12 +7,12 @@
 
 #include "connection.h"
 #include "server/livestreamer.h"
+#include "server/recstreamer.h"
 
 #include <map>
 #include <tools/select.h>
 
 class cChannel;
-class cStreamdevLiveStreamer;
 class cChannelList;
 
 class cConnectionHTTP: public cServerConnection {
@@ -27,12 +27,14 @@ private:
 	std::string                       m_Authorization;
 	eHTTPStatus                       m_Status;
 	tStrStrMap                        m_Params;
+	cStreamdevStreamer               *m_Streamer;
+	eStreamType                       m_StreamType;
 	// job: transfer
-	cStreamdevLiveStreamer           *m_LiveStreamer;
 	const cChannel                   *m_Channel;
 	int                               m_Apid[2];
 	int                               m_Dpid[2];
-	eStreamType                       m_StreamType;
+	// job: replay
+	cRecording                       *m_Recording;
 	// job: listing
 	cChannelList                     *m_ChannelList;
 
@@ -40,6 +42,13 @@ private:
 	bool ProcessURI(const std::string &PathInfo);
 	bool HttpResponse(int Code, bool Last, const char* ContentType = NULL, const char* Headers = "", ...);
 			//__attribute__ ((format (printf, 5, 6)));
+	/**
+	 * Extract byte range from HTTP Range header. Returns false if no valid
+	 * range is found. The contents of From and To are undefined in this
+	 * case. From may be negative in which case To is undefined.
+	 * TODO: support for multiple ranges.
+	 */
+	bool ParseRange(int64_t &From, int64_t &To) const;
 protected:
 	bool ProcessRequest(void);
 
@@ -47,8 +56,8 @@ public:
 	cConnectionHTTP(void);
 	virtual ~cConnectionHTTP();
 
-	virtual void Attach(void) { if (m_LiveStreamer != NULL) m_LiveStreamer->Attach(); }
-	virtual void Detach(void) { if (m_LiveStreamer != NULL) m_LiveStreamer->Detach(); }
+	virtual void Attach(void) { if (m_Streamer != NULL) m_Streamer->Attach(); }
+	virtual void Detach(void) { if (m_Streamer != NULL) m_Streamer->Detach(); }
 
 	virtual cString ToText() const;
 
@@ -62,7 +71,7 @@ public:
 
 inline bool cConnectionHTTP::Abort(void) const
 {
-	return !IsOpen() || (m_LiveStreamer && m_LiveStreamer->Abort());
+	return !IsOpen() || (m_Streamer && m_Streamer->Abort());
 }
 
 #endif // VDR_STREAMDEV_SERVERS_CONNECTIONVTP_H

server/livestreamer.h

@@ -49,7 +49,7 @@ public:
 	bool SetChannel(const cChannel *Channel, eStreamType StreamType, const int* Apid = NULL, const int* Dpid = NULL);
 	void SetPriority(int Priority);
 	void GetSignal(int *DevNum, int *Strength, int *Quality) const;
-	cString ToText() const;
+	virtual cString ToText() const;
 	
 	void Receive(uchar *Data, int Length);
 	virtual bool IsReceiving(void) const;

server/recstreamer.c

@@ -0,0 +1,67 @@
+#include "remux/ts2ps.h"
+#include "remux/ts2pes.h"
+#include "remux/ts2es.h"
+#include "remux/extern.h"
+
+#include <vdr/ringbuffer.h>
+#include "server/recstreamer.h"
+#include "server/connection.h"
+#include "common.h"
+
+using namespace Streamdev;
+
+// --- cStreamdevRecStreamer -------------------------------------------------
+
+cStreamdevRecStreamer::cStreamdevRecStreamer(cRecording *Rec, const cServerConnection *Connection):
+		cStreamdevStreamer("streamdev-recstreaming", Connection),
+		m_RecPlayer(Rec),
+		m_From(0L)
+{
+	Dprintf("New rec streamer\n");
+	m_To = (int64_t) m_RecPlayer.getLengthBytes() - 1;
+}
+
+cStreamdevRecStreamer::~cStreamdevRecStreamer() 
+{
+	Dprintf("Desctructing rec streamer\n");
+	Stop();
+}
+
+int64_t cStreamdevRecStreamer::SetRange(int64_t &From, int64_t &To)
+{
+	int64_t l = (int64_t) m_RecPlayer.getLengthBytes();
+	if (From < 0L) {
+		From += l;
+		if (From < 0L)
+			From = 0L;
+		To = l - 1;
+	}
+	else {
+		if (To < 0L)
+			To += l;
+		else if (To >= l)
+			To = l - 1;
+		if (From > To) {
+			// invalid range - return whole content
+			From = 0L;
+			To = l - 1;
+		}
+	}
+	m_From = From;
+	m_To = To;
+	return m_To - m_From + 1;
+}
+
+uchar* cStreamdevRecStreamer::GetFromReceiver(int &Count)
+{
+	if (m_From <= m_To) {
+		Count = (int) m_RecPlayer.getBlock(m_Buffer, m_From, sizeof(m_Buffer));
+		return m_Buffer;
+	}
+	return NULL;
+}
+
+cString cStreamdevRecStreamer::ToText() const
+{
+	return "REPLAY";
+}

server/recstreamer.h

@@ -0,0 +1,32 @@
+#ifndef VDR_STREAMDEV_RECSTREAMER_H
+#define VDR_STREAMDEV_RECSTREAMER_H
+
+#include "server/streamer.h"
+#include "server/recplayer.h"
+
+#define RECBUFSIZE (174 * TS_SIZE)
+
+// --- cStreamdevRecStreamer -------------------------------------------------
+
+class cStreamdevRecStreamer: public cStreamdevStreamer {
+private:
+	//Streamdev::cTSRemux    *m_Remux;
+	RecPlayer               m_RecPlayer;
+	int64_t                 m_From;
+	int64_t                 m_To;
+	uchar                   m_Buffer[RECBUFSIZE];
+
+protected:
+	virtual uchar* GetFromReceiver(int &Count);
+	virtual void DelFromReceiver(int Count) { m_From += Count; };
+
+public:
+	virtual bool IsReceiving(void) const { return m_From <= m_To; };
+	inline uint64_t GetLength() { return m_RecPlayer.getLengthBytes(); }
+	int64_t SetRange(int64_t &From, int64_t &To);
+	virtual cString ToText() const;
+	cStreamdevRecStreamer(cRecording *Recording, const cServerConnection *Connection);
+	virtual ~cStreamdevRecStreamer();
+};
+
+#endif // VDR_STREAMDEV_RECSTREAMER_H

server/streamer.h

@@ -92,6 +92,8 @@ public:
 
 	virtual void Detach(void) {}
 	virtual void Attach(void) {}
+
+	virtual cString ToText() const { return ""; };
 };
 
 inline bool cStreamdevStreamer::Abort(void)