Added locking to the queue and locked queue in encoder to empty it before stopping network stream.

3 files changed, 73 insertions, 2 deletions

diff --git a/src/encoder.cc b/src/encoder.cc
index 52a7f93..c304b5a 100644
--- a/src/encoder.cc
+++ b/src/encoder.cc
@@ -26,6 +26,8 @@
 
 #include "encoder.h"
 
+#include <time.h>
+
 Encoder::Encoder(Error* err,
                  const char *gip,
                  const int gport,
@@ -144,9 +146,22 @@ void Encoder::start() {
 
 
 void Encoder::stop(n_savestate save) {
+  struct timespec ts;
   // TODO: set save state in package header.
-  // TODO: Flush not yet sent video packages.
+
+  // Lock the queue and wait until all elements are sent on the network.
+  queue->lock();
+  while(queue->peek()) {
+    /* Remove any late buffer */
+    /* We don't care, the encoder finishes them all */
+    ts.tv_sec = 0;
+    ts.tv_nsec = 100000000L;	// 100ms
+    nanosleep(&ts, NULL);
+  }
+
   record = 0;
+
+  queue->unlock();
   if(s) {
     if(n) delete n;
     delete s;
diff --git a/src/player.h b/src/player.h
index ca2d241..0f0f7e5 100644
--- a/src/player.h
+++ b/src/player.h
@@ -32,7 +32,6 @@
 #include <stdlib.h>
 #include <semaphore.h>
 #include <pthread.h>
-#include <time.h>
 #include <SDL/SDL.h>
 //#include <avformat.h>
 
diff --git a/src/queue.h b/src/queue.h
index 5e6d0f8..a18b783 100644
--- a/src/queue.h
+++ b/src/queue.h
@@ -49,10 +49,15 @@ public:
 
   void push(T *t);
   T *pop();
+  T *peek();
   
+  void lock();
+  void unlock();
+
   void plength();
 
 private:
+  bool locked;
   int limit;
   buf_t *head;
   buf_t *tail;
@@ -67,6 +72,7 @@ private:
 template<typename T>
 Queue<T>::Queue(int glimit)
 {
+  locked = false;
   pthread_mutex_init (&mutex, NULL);
   limit = glimit;
   count = 0;
@@ -87,6 +93,9 @@ Queue<T>::~Queue()
   pthread_mutex_destroy(&mutex);
 }
 
+
+
+
 /**
  * Push element on queue.
  */
@@ -95,6 +104,11 @@ void Queue<T>::push(T *t)
 {
   pthread_mutex_lock(&mutex);
 
+  if(locked) {
+    delete t;
+    return;
+  }
+
   buf_t *b = (buf_t*)xmalloc(sizeof(*b));
   b->data = (void*)t;
 
@@ -167,6 +181,27 @@ T *Queue<T>::_pop()
 }
 
 /**
+ * Peek foremost element in queue
+ * If queue is empty, NULL is returned.
+ */
+template<typename T>
+T *Queue<T>::peek()
+{
+  pthread_mutex_lock(&mutex);
+  T *d;
+
+  assert(count >= 0);
+  
+  if(count == 0) {
+    return NULL;
+  }
+  
+  d = (T*)head->data;
+  pthread_mutex_unlock(&mutex);
+  return d;
+}
+
+/**
  * Print current length of queue
  */
 template<typename T>
@@ -178,5 +213,27 @@ void Queue<T>::plength()
   pthread_mutex_unlock(&mutex);
 }
 
+/**
+ * Lock the queue (all elements pushed from this point will be deleted.)
+ */
+template<typename T>
+void Queue<T>::lock()
+{
+  pthread_mutex_lock(&mutex);
+  locked = true;
+  pthread_mutex_unlock(&mutex);
+}
+
+/**
+ * Unlock the queue.
+ */
+template<typename T>
+void Queue<T>::unlock()
+{
+  pthread_mutex_lock(&mutex);
+  locked = false;
+  pthread_mutex_unlock(&mutex);
+}
+
 #endif