Close writer when streamer is finished

HISTORY

@@ -1,6 +1,7 @@
 VDR Plugin 'streamdev' Revision History
 ---------------------------------------
  
+- Close writer when streamer is finished
 - Don't abort VTP connection if filter stream is broken
 - Restructured cStreamdevStreamer: Moved inbound buffer into actual subclass.
 - In cStreamdevStreamer dropped Activate(bool) and moved its code into Start().

server/livefilter.c

@@ -18,13 +18,17 @@
 class cStreamdevLiveFilter: public cFilter {
 private:
 	cStreamdevFilterStreamer *m_Streamer;
+	bool m_On;
 
 protected:
 	virtual void Process(u_short Pid, u_char Tid, const u_char *Data, int Length);
+	virtual void SetStatus(bool On);
 
 public:
 	cStreamdevLiveFilter(cStreamdevFilterStreamer *Streamer);
 
+	virtual bool IsAttached(void) const { return m_On; };
+
 	void Set(u_short Pid, u_char Tid, u_char Mask) {
 		cFilter::Set(Pid, Tid, Mask);
 	}
@@ -34,9 +38,16 @@ public:
 };
 
 cStreamdevLiveFilter::cStreamdevLiveFilter(cStreamdevFilterStreamer *Streamer) {
+	m_On = false;
 	m_Streamer = Streamer;
 }
 
+void cStreamdevLiveFilter::SetStatus(bool On)
+{
+	m_On = On;
+	cFilter::SetStatus(On);
+}
+
 void cStreamdevLiveFilter::Process(u_short Pid, u_char Tid, const u_char *Data, int Length) 
 {
 	uchar buffer[TS_SIZE];
@@ -113,6 +124,11 @@ void cStreamdevFilterStreamer::SetDevice(cDevice *Device)
 	Attach();
 }
 
+bool cStreamdevFilterStreamer::IsReceiving(void) const
+{
+	return m_Filter && m_Filter->IsAttached();
+}
+
 bool cStreamdevFilterStreamer::SetFilter(u_short Pid, u_char Tid, u_char Mask, bool On) 
 {	
 	Dprintf("cStreamdevFilterStreamer::SetFilter(%u,0x%x,0x%x,%s)\n", Pid, Tid, Mask, On?"On":"Off");

server/livefilter.h

@@ -26,6 +26,8 @@ public:
 
 	void SetDevice(cDevice *Device);
 	bool SetFilter(u_short Pid, u_char Tid, u_char Mask, bool On);
+
+	virtual bool IsReceiving(void) const;
 	void Receive(uchar *Data);
 	
 	virtual void Attach(void);

server/livestreamer.c

@@ -449,13 +449,19 @@ cString cStreamdevLiveStreamer::ToText() const
         return cString("");
 }
 
+bool cStreamdevLiveStreamer::IsReceiving(void) const
+{
+	cThreadLock ThreadLock(m_Device);
+	return m_Receiver && m_Receiver->IsAttached();
+}
+
 void cStreamdevLiveStreamer::StartReceiver(void)
 {
 	if (m_NumPids > 0) {
 		Dprintf("Creating Receiver to respect changed pids\n");
 		cReceiver *current = m_Receiver;
-		m_Receiver = new cStreamdevLiveReceiver(this, m_Channel, m_Priority, m_Pids);
 		cThreadLock ThreadLock(m_Device);
+		m_Receiver = new cStreamdevLiveReceiver(this, m_Channel, m_Priority, m_Pids);
 		if (IsRunning())
 			Attach();
 		delete current;

server/livestreamer.h

@@ -52,6 +52,8 @@ public:
 	cString ToText() const;
 	
 	void Receive(uchar *Data, int Length);
+	virtual bool IsReceiving(void) const;
+
 	virtual uchar *Get(int &Count);
 	virtual void Del(int Count);
 

server/streamer.c

@@ -54,6 +54,11 @@ void cStreamdevWriter::Action(void)
 		if (block == NULL) {
 			block = m_Streamer->Get(count);
 			offset = 0;
+			// still no data - are we done?
+			if (block == NULL && !m_Streamer->IsReceiving() && timeout++ > 20) {
+				esyslog("streamdev-server: streamer done - writer exiting");
+				break;
+			}
 		}
 
 		if (block != NULL) {
@@ -100,6 +105,7 @@ void cStreamdevWriter::Action(void)
 			}
 		}
 	}
+	m_Socket->Close();
 	Dprintf("Max. Transmit Blocksize was: %d\n", max);
 }
 

server/streamer.h

@@ -84,6 +84,7 @@ public:
 
 	virtual void Start(cTBSocket *Socket);
 	virtual void Stop(void);
+	virtual bool IsReceiving(void) const = 0;
 	bool Abort(void);
 
 	virtual uchar *Get(int &Count) { return m_SendBuffer->Get(Count); }