node.hh

#ifndef __SDR_NODE_HH__
#define __SDR_NODE_HH__

#include <set>
#include <list>
#include <iostream>
#include <complex>
#include <stdint.h>
#include <pthread.h>

#include "buffer.hh"
#include "queue.hh"
#include "exception.hh"

namespace sdr {

class Config
{
public:
  typedef enum {
    Type_UNDEFINED = 0,
    Type_u8,   
    Type_s8,   
    Type_u16,  
    Type_s16,  
    Type_f32,  
    Type_f64,  
    Type_cu8,  
    Type_cs8,  
    Type_cu16, 
    Type_cs16, 
    Type_cf32, 
    Type_cf64  
  } Type;

public:
  Config();
  Config(Type type, double sampleRate, size_t bufferSize, size_t numBuffers);
  Config(const Config &other);

  const Config &operator= (const Config &other);
  bool operator== (const Config &other) const;

  inline bool hasType() const { return Type_UNDEFINED != _type; }
  inline Type type() const { return _type; }
  inline void setType(Type type) { _type = type; }
  inline bool hasSampleRate() const { return 0 != _sampleRate; }
  inline double sampleRate() const { return _sampleRate; }
  inline void setSampleRate(double rate) { _sampleRate = rate; }
  inline bool hasBufferSize() const { return 0 != _bufferSize; }
  inline size_t bufferSize() const { return _bufferSize; }
  inline void setBufferSize(size_t size) { _bufferSize = size; }
  inline bool hasNumBuffers() const { return 0 != _numBuffers; }
  inline size_t numBuffers() const { return _numBuffers; }
  inline void setNumBuffers(size_t N) { _numBuffers = N; }

  template <typename T> static inline Type typeId();

protected:
  Type   _type;
  double _sampleRate;
  size_t _bufferSize;
  size_t _numBuffers;
};

template <>
inline Config::Type Config::typeId<uint8_t>() { return Type_u8; }
template <>
inline Config::Type Config::typeId<int8_t>() { return Type_s8; }
template <>
inline Config::Type Config::typeId<uint16_t>() { return Type_u16; }
template <>
inline Config::Type Config::typeId<int16_t>() { return Type_s16; }
template <>
inline Config::Type Config::typeId<float>() { return Type_f32; }
template <>
inline Config::Type Config::typeId<double>() { return Type_f64; }
template <>
inline Config::Type Config::typeId< std::complex<uint8_t> >() { return Type_cu8; }
template <>
inline Config::Type Config::typeId< std::complex<int8_t> >() { return Type_cs8; }
template <>
inline Config::Type Config::typeId< std::complex<uint16_t> >() { return Type_cu16; }
template <>
inline Config::Type Config::typeId< std::complex<int16_t> >() { return Type_cs16; }
template <>
inline Config::Type Config::typeId< std::complex<float> >() { return Type_cf32; }
template <>
inline Config::Type Config::typeId< std::complex<double> >() { return Type_cf64; }


inline const char *typeName(Config::Type type) {
  switch (type) {
  case Config::Type_UNDEFINED: return "UNDEFINED";
  case Config::Type_u8: return "uint8";
  case Config::Type_s8: return "int8";
  case Config::Type_u16: return "uint16";
  case Config::Type_s16: return "int16";
  case Config::Type_f32: return "float";
  case Config::Type_f64: return "double";
  case Config::Type_cu8: return "complex uint8";
  case Config::Type_cs8: return "complex int8";
  case Config::Type_cu16: return "complex uint16";
  case Config::Type_cs16: return "complex int16";
  case Config::Type_cf32: return "complex float";
  case Config::Type_cf64: return "complex double";
  }
  return "unknown";
}

inline std::ostream &operator<<(std::ostream &stream, Config::Type type) {
  stream << typeName(type) << " (" << (int)type << ")";
  return stream;
}


class SinkBase
{
public:
  SinkBase();
  virtual ~SinkBase();

  virtual void handleBuffer(const RawBuffer &buffer, bool allow_overwrite) = 0;
  virtual void config(const Config &src_cfg) = 0;
};



template <class Scalar>
class Sink: public SinkBase
{
public:
  Sink() : SinkBase() { }
  virtual ~Sink() { }

  virtual void process(const Buffer<Scalar> &buffer, bool allow_overwrite) = 0;

  virtual void handleBuffer(const RawBuffer &buffer, bool allow_overwrite) {
    this->process(Buffer<Scalar>(buffer), allow_overwrite);
  }
};



class Source
{
public:
  Source();
  virtual ~Source();

  virtual void send(const RawBuffer &buffer, bool allow_overwrite=false);

  void connect(SinkBase *sink, bool direct=false);
  void disconnect(SinkBase *sink);

  virtual void setConfig(const Config &config);

  virtual double sampleRate() const;
  virtual Config::Type type() const;

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

protected:
  void signalEOS();

  void propagateConfig(const Config &config);

protected:
  Config _config;
  std::map<SinkBase *, bool> _sinks;
  std::list<DelegateInterface *> _eos;
};



class BlockingSource: public Source {
public:
  BlockingSource(bool parallel=false, bool connect_idle=true, bool stop_queue_on_eos=false);
  virtual ~BlockingSource();

  virtual void next() = 0;

  inline bool isActive() const { return _is_active; }

  virtual void start();
  virtual void stop();

protected:
  void _parallel_main();
  void _nonvirt_idle_cb();

protected:
  bool _is_active;
  bool _is_parallel;
  pthread_t _thread;

private:
  static void *_pthread_main_wrapper(void *);
};


class Proxy: public SinkBase, public Source
{
public:
  Proxy();
  virtual ~Proxy();

  virtual void config(const Config &src_cfg);
  virtual void handleBuffer(const RawBuffer &buffer, bool allow_overwrite);
};


}

#endif // __SDR_NODE_HH__