1 files changed, 163 insertions, 0 deletions
diff --git a/mpeglib/lib/decoder/commandPipe.cpp b/mpeglib/lib/decoder/commandPipe.cpp
new file mode 100644
index 00000000..05991ca5
--- /dev/null
+++ b/mpeglib/lib/decoder/commandPipe.cpp
@@ -0,0 +1,163 @@
+/*
+  thread safe fifo queue for commands
+  Copyright (C) 2000  Martin Vogt
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of the GNU Library General Public License as published by
+  the Free Software Foundation.
+
+  For more information look at the file COPYRIGHT in this package
+
+ */
+
+
+#include "commandPipe.h"
+
+#define _COMMAND_ARRAY_SIZE   100
+
+CommandPipe::CommandPipe() {
+  abs_thread_cond_init(&spaceCond);
+  abs_thread_cond_init(&emptyCond);
+  abs_thread_cond_init(&dataCond);
+  abs_thread_mutex_init(&pipeMut);
+
+  entries=0;
+  readPos=0;
+  writePos=0;
+
+  
+  int i;
+  
+  commandArray=new Command*[_COMMAND_ARRAY_SIZE];
+
+
+  for(i=0;i<_COMMAND_ARRAY_SIZE;i++) {
+    commandArray[i]=new Command(_COMMAND_NONE,0);
+  }
+}
+
+
+CommandPipe::~CommandPipe() {
+  abs_thread_cond_destroy(&spaceCond);
+  abs_thread_cond_destroy(&emptyCond);
+  abs_thread_cond_destroy(&dataCond);
+  abs_thread_mutex_destroy(&pipeMut);
+
+
+  int i;
+
+  for(i=0;i<_COMMAND_ARRAY_SIZE;i++) {
+    delete commandArray[i];
+  }
+  delete [] commandArray;
+}
+
+void CommandPipe::sendCommand(Command& cmd) {
+  sendCommand(cmd,true);
+}
+
+void CommandPipe::sendCommandNoWait(Command& cmd) {
+  sendCommand(cmd,false);
+}
+
+
+void CommandPipe::sendCommand(Command& cmd,int lWait) {
+  lockCommandPipe();
+  if (entries==_COMMAND_ARRAY_SIZE) {
+    waitForSpace();
+  }
+  cmd.copyTo(commandArray[writePos]);
+  writePos++;
+  if (writePos==_COMMAND_ARRAY_SIZE) {
+    writePos=0;
+  }
+  entries++;
+  // low water signal
+  if (entries==1) {
+    signalData();
+  }
+  unlockCommandPipe(); 
+  if (lWait) {
+    waitForEmptyQueue();
+  }
+}
+
+
+int CommandPipe::hasCommand(Command* dest) {
+  lockCommandPipe();
+  if (entries==0) {
+    unlockCommandPipe();
+    return false;
+  }
+
+  commandArray[readPos]->copyTo(dest);
+  readPos++;
+  if (readPos==_COMMAND_ARRAY_SIZE) {
+    readPos=0;
+  }  
+  entries--;
+  // low water
+  if (entries==0) {
+    signalEmpty();
+    unlockCommandPipe();
+    return true;
+  }
+  // if we are on highwater, signal space
+  if (entries==_COMMAND_ARRAY_SIZE-1) {
+    signalSpace();
+  }
+  unlockCommandPipe();
+  return true;
+}
+
+
+void CommandPipe::waitForEmptyQueue() {
+  lockCommandPipe();
+  while (entries > 0) {
+    waitForEmpty();
+  }
+  unlockCommandPipe();
+}
+
+void CommandPipe::waitForCommand() {
+  lockCommandPipe();
+  while (entries == 0) {
+    waitForData();
+  }
+  unlockCommandPipe();
+}
+
+
+void CommandPipe::lockCommandPipe() {
+  abs_thread_mutex_lock(&pipeMut);
+}
+
+
+void CommandPipe::unlockCommandPipe() {
+  abs_thread_mutex_unlock(&pipeMut);
+}
+
+void CommandPipe::waitForSpace() {
+  abs_thread_cond_wait(&spaceCond,&pipeMut);
+}
+
+void CommandPipe::waitForEmpty() {
+  abs_thread_cond_wait(&emptyCond,&pipeMut);
+}
+ 
+void CommandPipe::waitForData() {
+  abs_thread_cond_wait(&dataCond,&pipeMut);
+}
+
+
+void CommandPipe::signalSpace() {
+  abs_thread_cond_signal(&spaceCond);
+}
+
+void CommandPipe::signalEmpty() {
+  abs_thread_cond_signal(&emptyCond);
+}
+
+void CommandPipe::signalData() {
+  abs_thread_cond_signal(&dataCond);
+}