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

1 files changed, 57 insertions, 0 deletions

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