queue.hh

#ifndef __SDR_QUEUE_HH__
#define __SDR_QUEUE_HH__

#include <list>
#include <map>
#include "buffer.hh"
#include <pthread.h>
#include <iostream>


namespace sdr {

// Forward decl.
class SinkBase;

class DelegateInterface {
public:
  virtual void operator() () = 0;
  virtual void *instance() = 0;
};

template <class T>
class Delegate: public DelegateInterface
{
public:
  Delegate(T *instance, void (T::*func)(void)) : _instance(instance), _function(func) { }
  virtual ~Delegate() {}
  virtual void operator() () { (_instance->*_function)(); }
  virtual void *instance() { return _instance; }

protected:
  T *_instance;
  void (T::*_function)(void);
};


class Queue
{
public:
  class Message {
  public:
    Message(const RawBuffer &buffer, SinkBase *sink, bool allow_overwrite)
      : _buffer(buffer), _sink(sink), _allow_overwrite(allow_overwrite) { }
    Message(const Message &other)
      : _buffer(other._buffer), _sink(other._sink), _allow_overwrite(other._allow_overwrite) { }
    const Message &operator= (const Message &other) {
      _buffer = other._buffer;
      _sink   = other._sink;
      _allow_overwrite = other._allow_overwrite;
      return *this;
    }
    inline const RawBuffer &buffer() const { return _buffer; }
    inline RawBuffer &buffer() { return _buffer; }
    inline SinkBase *sink() const { return _sink; }
    inline bool allowOverwrite() const { return _allow_overwrite; }

  protected:
    RawBuffer _buffer;
    SinkBase *_sink;
    bool _allow_overwrite;
  };

protected:
  Queue();

public:
  virtual ~Queue();

  static Queue &get();

  void send(const RawBuffer &buffer, SinkBase *sink, bool allow_overwrite=false);

  void start();

  void stop();
  void wait();

  bool isStopped() const;
  bool isRunning() const;

  template <class T>
  void addIdle(T *instance, void (T::*function)(void)) {
    _idle.push_back(new Delegate<T>(instance, function));
  }

  template <class T>
  void remIdle(T *instance) {
    std::list<DelegateInterface *>::iterator item = _idle.begin();
    while (item != _idle.end()) {
      if ( (*item)->instance() == ((void *)instance)) {
        item = _idle.erase(item);
      } else {
        item++;
      }
    }
  }

  template <class T>
  void addStart(T *instance, void (T::*function)(void)) {
    _onStart.push_back(new Delegate<T>(instance, function));
  }

  template <class T>
  void remStart(T *instance) {
    std::list<DelegateInterface *>::iterator item = _onStart.begin();
    while (item != _onStart.end()) {
      if ( (*item)->instance() == ((void *)instance)) {
        item = _onStart.erase(item);
      } else {
        item++;
      }
    }
  }

  template <class T>
  void addStop(T *instance, void (T::*function)(void)) {
    _onStop.push_back(new Delegate<T>(instance, function));
  }

  template <class T>
  void remStop(T *instance) {
    std::list<DelegateInterface *>::iterator item = _onStop.begin();
    while (item != _onStop.end()) {
      if ( (*item)->instance() == ((void *)instance)) {
        item = _onStop.erase(item);
      } else {
        item++;
      }
    }
  }

protected:
  void _main();
  void _signalIdle();
  void _signalStart();
  void _signalStop();

protected:
  bool _running;
  pthread_t _thread;
  pthread_mutex_t _queue_lock;
  pthread_cond_t  _queue_cond;

  std::list<Message> _queue;
  std::list<DelegateInterface *> _idle;
  std::list<DelegateInterface *> _onStart;
  std::list<DelegateInterface *> _onStop;

private:
  static Queue *_instance;
  static void *__thread_start(void *ptr);
};


}
#endif // __SDR_QUEUE_HH__