Support Sink with multiple elements

examples/Threading/Demo_Source_Sink_Counter/Consumer.inot

@@ -1,4 +1,4 @@
-SINK(counter, int);
+SINK(counter, int, 10);
 
 void setup()
 {

src/Arduino_Threads.h

@@ -44,19 +44,37 @@ public: \
   Source<type> name; \
 private:
 
-#define SINK(name, type) \
+/* We need to call the SinkBlocking<T>(size_t const size)
+ * non-default constructor using size as parameter.
+
+ * This is achieved via
+ *   SinkBlocking<type> name{size};
+ * instead of
+ *   SinkBlocking<type> name(size);
+ * otherwise the compiler will read it as a declaration
+ * of a method called "name" and we get a syntax error.
+ *
+ * This is called "C++11 uniform init" (using "{}" instead
+ * of "()" without "="... yikes!)
+ *   https://chromium.googlesource.com/chromium/src/+/master/styleguide/c++/c++-dos-and-donts.md
+ */
+
+#define SINK_2_ARG(name, type) \
+public: \
+  SinkBlocking<type> name{1}; \
+private:
+
+#define SINK_3_ARG(name, type, size) \
 public: \
-  SinkBlocking<type> name; \
+  SinkBlocking<type> name{size}; \
 private:
-// we need to call the Sink<T>(int size) non-default constructor using size as parameter.
-// This is done by writing
-//    Sink<type> name{size};
-// instead of:
-//    Sink<type> name(size);
-// otherwise the compiler will read it as a declaration of a method called "name" and we
-// get a syntax error.
-// This is called "C++11 uniform init" (using "{}" instead of "()" without "="... yikes!)
-// https://chromium.googlesource.com/chromium/src/+/master/styleguide/c++/c++-dos-and-donts.md
+
+/* Black C macro magic enabling "macro overloading"
+ * with same name macro, but multiple arguments.
+ * https://stackoverflow.com/questions/11761703/overloading-macro-on-number-of-arguments
+ */
+#define GET_SINK_MACRO(_1,_2,_3,NAME,...) NAME
+#define SINK(...) GET_SINK_MACRO(__VA_ARGS__, SINK_3_ARG, SINK_2_ARG)(__VA_ARGS__)
 
 #define SHARED(name, type) \
   Shared<type> name;

src/threading/CircularBuffer.hpp

@@ -0,0 +1,118 @@
+/*
+ * This file is part of the Arduino_ThreadsafeIO library.
+ * Copyright (c) 2021 Arduino SA.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#ifndef ARDUINO_THREADS_RINGBUFFER_HPP_
+#define ARDUINO_THREADS_RINGBUFFER_HPP_
+
+/**************************************************************************************
+ * INCLUDE
+ **************************************************************************************/
+
+#include <SharedPtr.h>
+
+/**************************************************************************************
+ * CLASS DECLARATION
+ **************************************************************************************/
+
+template <typename T>
+class CircularBuffer
+{
+public:
+
+  CircularBuffer(size_t const size);
+
+  void store(T const data);
+  T read();
+  bool isFull() const;
+  bool isEmpty() const;
+
+
+private:
+
+  mbed::SharedPtr<T> _data;
+  size_t const _size;
+  size_t _head, _tail, _num_elems;
+
+  size_t next(size_t const idx);
+};
+
+/**************************************************************************************
+ * CTOR/DTOR
+ **************************************************************************************/
+
+template <typename T>
+CircularBuffer<T>::CircularBuffer(size_t const size)
+: _data{new T[size]}
+, _size{size}
+, _head{0}
+, _tail{0}
+, _num_elems{0}
+{
+}
+
+/**************************************************************************************
+ * PUBLIC MEMBER FUNCTIONS
+ **************************************************************************************/
+
+template <typename T>
+void CircularBuffer<T>::store(T const data)
+{
+  if (!isFull())
+  {
+    _data.get()[_head] = data;
+    _head = next(_head);
+    _num_elems++;
+  }
+}
+
+template <typename T>
+T CircularBuffer<T>::read()
+{
+  if (isEmpty())
+    return T{0};
+
+  T const value = _data.get()[_tail];
+  _tail = next(_tail);
+  _num_elems--;
+
+  return value;
+}
+
+template <typename T>
+bool CircularBuffer<T>::isFull() const
+{
+  return (_num_elems == _size);
+}
+
+template <typename T>
+bool CircularBuffer<T>::isEmpty() const
+{
+  return (_num_elems == 0);
+}
+
+/**************************************************************************************
+ * PRIVATE MEMBER FUNCTIONS
+ **************************************************************************************/
+
+template <typename T>
+size_t CircularBuffer<T>::next(size_t const idx)
+{
+  return ((idx + 1) % _size);
+}
+
+#endif /* ARDUINO_THREADS_RINGBUFFER_HPP_ */

src/threading/Sink.hpp

@@ -25,6 +25,8 @@
 
 #include <mbed.h>
 
+#include "CircularBuffer.hpp"
+
 /**************************************************************************************
  * CLASS DECLARATION
  **************************************************************************************/
@@ -64,7 +66,7 @@ class SinkBlocking : public SinkBase<T>
 {
 public:
 
-           SinkBlocking();
+           SinkBlocking(size_t const size);
   virtual ~SinkBlocking() { }
 
   virtual operator T() override;
@@ -73,8 +75,7 @@ class SinkBlocking : public SinkBase<T>
 
 private:
 
-  T _data;
-  bool _is_data_available;
+  CircularBuffer<T> _data;
   rtos::Mutex _mutex;
   rtos::ConditionVariable _cond_data_available;
   rtos::ConditionVariable _cond_slot_available;
@@ -106,8 +107,8 @@ void SinkNonBlocking<T>::inject(T const & value)
  **************************************************************************************/
 
 template<typename T>
-SinkBlocking<T>::SinkBlocking()
-: _is_data_available{false}
+SinkBlocking<T>::SinkBlocking(size_t const size)
+: _data(size)
 , _cond_data_available(_mutex)
 , _cond_slot_available(_mutex)
 { }
@@ -116,10 +117,9 @@ template<typename T>
 SinkBlocking<T>::operator T()
 {
   _mutex.lock();
-  while (!_is_data_available)
+  while (_data.isEmpty())
     _cond_data_available.wait();
-  T const d = _data;
-  _is_data_available = false;
+  T const d = _data.read();
   _cond_slot_available.notify_all();
   _mutex.unlock();
   return d;
@@ -129,10 +129,9 @@ template<typename T>
 void SinkBlocking<T>::inject(T const & value)
 {
   _mutex.lock();
-  while (_is_data_available)
+  while (_data.isFull())
     _cond_slot_available.wait();
-  _data = value;
-  _is_data_available = true;
+  _data.store(value);
   _cond_data_available.notify_all();
   _mutex.unlock();
 }