A big rework of the source/sink system.

Author: aentinger

examples/Demo_Shared/Consumer.inot renamed to examples/Demo_Shared_Counter/Consumer.inot

@@ -1,11 +1,13 @@
-void setup() {
+void setup()
+{
   Serial.begin(115200);
   while (!Serial) { }
 
   Producer.start();
   Consumer.start();
 }
 
-void loop() {
+void loop()
+{
   rtos::ThisThread::yield();
 }

examples/Demo_Shared/Demo_Shared.ino renamed to examples/Demo_Shared_Counter/Demo_Shared_Counter.ino

@@ -0,0 +1,11 @@
+SINK(counter, int);
+
+void setup()
+{
+
+}
+
+void loop()
+{
+  Serial.println(counter.read());
+}

examples/Demo_Shared/Producer.inot renamed to examples/Demo_Shared_Counter/Producer.inot

@@ -0,0 +1,16 @@
+/*
+ * This examples demonstrates the SOURCE/SINK abstraction.
+ * Each thread may have any number of SOURCES and SINKS that can be connected
+ * together using the "connectTo" method.
+ */
+
+void setup()
+{
+  Producer.counter.connectTo(Consumer.counter);
+  Producer.start();
+  Consumer.start();
+}
+
+void loop() {
+  rtos::ThisThread::yield();
+}

examples/Demo_Shared/SharedVariables.h renamed to examples/Demo_Shared_Counter/SharedVariables.h

@@ -0,0 +1,13 @@
+SOURCE(counter, int);
+
+void setup()
+{
+
+}
+
+void loop()
+{
+  static int i = 0;
+  counter.write(i);
+  i++;
+}

examples/Demo_Source_Sink_Counter/Consumer.inot

@@ -3,7 +3,7 @@
  * An 'bool' SINK with a size of '0'. This kind of SINK has no buffer so the reading thread
  * will block until the writing thread has written something, or vice versa.
  */
-SINK(led, bool, 0)
+SINK(led, bool);
 
 void setup()
 {

examples/Demo_Source_Sink_Counter/Demo_Source_Sink_Counter.ino

@@ -1,5 +1,5 @@
 /* The output SOURCE, it sends 'bool' values. */
-SOURCE(led, bool)
+SOURCE(led, bool);
 
 void setup()
 {
@@ -8,8 +8,8 @@ void setup()
 
 void loop()
 {
-  led.send(true);
+  led.write(true);
   delay(100);
-  led.send(false);
+  led.write(false);
   delay(100);
 }

examples/Demo_Source_Sink_Counter/Producer.inot

@@ -39,9 +39,9 @@ public: \
   Source<type> name; \
 private:
 
-#define SINK(name, type, size) \
+#define SINK(name, type) \
 public: \
-  Sink<type> name{size}; \
+  SinkBlocking<type> name; \
 private:
 // we need to call the Sink<T>(int size) non-default constructor using size as parameter.
 // This is done by writing

examples/Demo_Source_Sink/SharedVariables.h renamed to examples/Demo_Source_Sink_Counter/SharedVariables.h

@@ -29,121 +29,79 @@
  * CLASS DECLARATION
  **************************************************************************************/
 
-template<class T>
-class Sink
+template<typename T>
+class SinkBase
 {
-  private:
-    rtos::Mutex dataMutex;
-    rtos::ConditionVariable dataAvailable;
-    rtos::ConditionVariable slotAvailable;
-    T latest;
-    Sink *next;
-    const int size;
-    int first, last;
-    bool full;
-    T *queue;
-
-  public:
-    Sink(int s) :
-      dataAvailable(dataMutex),
-      slotAvailable(dataMutex),
-      size(s),
-      queue((size > 0) ? new T[size] : nullptr),
-      first(0), last(0), full(false)
-    {};
-
-    ~Sink() {
-      if (queue != nullptr) { delete queue; }
-    }
-
-
-  //protected: TODO
-    void connectTo(Sink &sink)
-    {
-      if (next == nullptr) {
-        next = &sink;
-      } else {
-        next->connectTo(sink);
-      }
-    }
-
-    T read()
-    {
-      // Non-blocking shared variable
-      if (size == -1) {
-        dataMutex.lock();
-        T res = latest;
-        dataMutex.unlock();
-        return res;
-      }
-
-      // Blocking shared variable
-      if (size == 0) {
-        dataMutex.lock();
-        while (!full) {
-          dataAvailable.wait();
-        }
-        T res = latest;
-        full = false;
-        slotAvailable.notify_all();
-        dataMutex.unlock();
-        return res;
-      }
-
-      // Blocking queue
-      dataMutex.lock();
-      while (first == last && !full) {
-        dataAvailable.wait();
-      }
-      T res = queue[first++];
-      first %= size;
-      if (full) {
-        full = false;
-        slotAvailable.notify_one();
-      }
-      dataMutex.unlock();
-      return res;
-    }
-
-  //protected: TODO
-    void inject(const T &value)
-    {
-      dataMutex.lock();
-
-      // Non-blocking shared variable
-      if (size == -1) {
-        latest = value;
-      }
-
-      // Blocking shared variable
-      else if (size == 0) {
-        while (full) {
-          slotAvailable.wait();
-        }
-        latest = value;
-        full = true;
-        dataAvailable.notify_one();
-        slotAvailable.wait();
-      }
-
-      // Blocking queue
-      else {
-        while (full) {
-          slotAvailable.wait();
-        }
-        if (first == last) {
-          dataAvailable.notify_one();
-        }
-        queue[last++] = value;
-        last %= size;
-        if (first == last) {
-          full = true;
-        }
-      }
-      dataMutex.unlock();
-
-      if (next) next->inject(value);
-    }
+public:
+
+  virtual ~SinkBase() { }
+
+  virtual T read() = 0;
+  virtual void inject(T const & value) = 0;
+};
+
+template<typename T>
+class SinkNonBlocking : public SinkBase<T>
+{
+  /* TODO - Do we really need this? */
 };
 
+template<typename T>
+class SinkBlocking : public SinkBase<T>
+{
+public:
+
+           SinkBlocking();
+  virtual ~SinkBlocking() { }
+
+  virtual T read() override;
+  virtual void inject(T const & value) override;
+
+
+private:
+
+  T _data;
+  bool _is_data_available;
+  rtos::Mutex _mutex;
+  rtos::ConditionVariable _cond_data_available;
+  rtos::ConditionVariable _cond_slot_available;
+
+};
+
+/**************************************************************************************
+ * PUBLIC MEMBER FUNCTIONS - SinkBlocking
+ **************************************************************************************/
+
+template<typename T>
+SinkBlocking<T>::SinkBlocking()
+: _is_data_available{false}
+, _cond_data_available(_mutex)
+, _cond_slot_available(_mutex)
+{ }
+
+template<typename T>
+T SinkBlocking<T>::read()
+{
+  _mutex.lock();
+  while (!_is_data_available)
+    _cond_data_available.wait();
+  T const d = _data;
+  _is_data_available = false;
+  _cond_slot_available.notify_all();
+  _mutex.unlock();
+  return d;
+}
+
+template<typename T>
+void SinkBlocking<T>::inject(T const & value)
+{
+  _mutex.lock();
+  while (_is_data_available)
+    _cond_slot_available.wait();
+  _data = value;
+  _is_data_available = true;
+  _cond_data_available.notify_all();
+  _mutex.unlock();
+}
+
 #endif /* ARDUINO_THREADS_SINK_HPP_ */

examples/Demo_Source_Sink/Demo_Source_Sink.ino renamed to examples/Demo_Source_Sink_LED/Demo_Source_Sink_LED.ino

@@ -19,37 +19,55 @@
 #ifndef ARDUINO_THREADS_SOURCE_HPP_
 #define ARDUINO_THREADS_SOURCE_HPP_
 
+/**************************************************************************************
+ * INCLUDE
+ **************************************************************************************/
+
+#include <list>
+#include <algorithm>
+
 /**************************************************************************************
  * FORWARD DECLARATION
  **************************************************************************************/
 
 template<class T>
-class Sink;
+class SinkBase;
 
 /**************************************************************************************
  * CLASS DECLARATION
  **************************************************************************************/
 
-template<class T>
+template<typename T>
 class Source
 {
-  public:
-    Source() {};
-
-    void connectTo(Sink<T> &sink) {
-      if (destination == nullptr) {
-        destination = &sink;
-      } else {
-        destination->connectTo(sink);
-      }
-    }
-
-    void send(const T &value) {
-      if (destination) destination->inject(value);
-    }
-
-  private:
-    Sink<T> *destination;
+public:
+
+  void connectTo(SinkBase<T> & sink);
+  void write(T const & value);
+
+private:
+  std::list<SinkBase<T> *> _sink_list;
 };
 
+/**************************************************************************************
+ * PUBLIC MEMBER FUNCTIONS
+ **************************************************************************************/
+
+template<typename T>
+void Source<T>::connectTo(SinkBase<T> & sink)
+{
+  _sink_list.push_back(&sink);
+}
+
+template<typename T>
+void Source<T>::write(T const & value)
+{
+  std::for_each(std::begin(_sink_list),
+                std::end  (_sink_list),
+                [value](SinkBase<T> * sink)
+                {
+                  sink->inject(value);
+                });
+}
+
 #endif /* ARDUINO_THREADS_SOURCE_HPP_ */