Fix issues

.github/workflows/main.yml

@@ -19,3 +19,18 @@ jobs:
         run: python -m pip install platformio
       - name: Build firmware
         run: pio ci --lib="." --board ${{matrix.pio-env}}  "examples/${{matrix.example}}"
+
+  test:
+    name: Unit Tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.x'
+          architecture: 'x64'
+      - name: Install PlatformIO
+        run: python -m pip install platformio
+      - name: Run unit tests
+        run: pio test -e native

README.md

@@ -23,7 +23,137 @@ More examples in Wiki/[Examples](https://github.com/luisllamasbinaburo/Arduino-R
 ### Observable, observers and operators legend
 More info about the Observables, Observers, and Operators available in the [Wiki](https://github.com/luisllamasbinaburo/Arduino-ReactiveArduino/wiki)
 
-![alt text](https://github.com/luisllamasbinaburo/Arduino-ReactiveArduino/blob/master/ReactiveArduino%20Legend.png "Legend")
+```mermaid
+flowchart LR
+    subgraph OBS[Observables]
+        direction TB
+        ManualDefer
+        Range
+        RangeDefer
+        Array
+        ArrayDefer
+        Property
+        AnalogInput
+        DigitalInput
+        TimerMillis
+        TimerMicros
+        IntervalMillis
+        IntervalMicros
+        SerialChar
+        SerialByte
+        SerialString
+        SerialInteger
+        SerialFloat
+        SerialDouble
+    end
+
+    subgraph OPR[Operators]
+        direction TB
+        subgraph OPO[Operators]
+            Where
+            Distinct
+            First
+            Last
+            Take
+            TakeAt
+            TakeFirst
+            TakeLast
+            TakeUntil
+            TakeWhile
+            Skip
+            SkipUntil
+            SkipWhile
+            Batch
+            TimeoutMillis
+            TimeoutMicros
+            ForEach
+            If
+            Loop
+            Repeat
+            Reset
+            NoReset
+        end
+        subgraph TRN[Transformations]
+            Select
+            Cast
+            Map
+            Reduce
+            Limit
+            LimitLower
+            LimitUpper
+            Scale
+            ElapsedMicros
+            ElapsedMillis
+            Micros
+            Millis
+            Frequency
+            Threshold
+            Toggle
+            AdcToVoltage
+            Split
+            Join
+            Buffer
+            StringBuffer
+            ToBool
+            ToInt
+            ToFloat
+            ParseInt
+            ParseFloat
+        end
+        subgraph FLT[Filters]
+            OnRising
+            OnFalling
+            Median3
+            Median5
+            MovingAverage
+            MovingRMS
+            LowPass
+            HighPass
+            PassBand
+            StopBand
+            WindowMillis
+            WindowMicros
+            DebounceMillis
+            DebounceMicros
+            IsLessOrEqual
+            IsLess
+            IsGreaterOrEqual
+            IsGreater
+            IsNotEqual
+            IsEqual
+            IsZero
+            IsNotZero
+        end
+        subgraph AGG[Aggregates]
+            Count
+            Countdown
+            Sum
+            Min
+            Max
+            Average
+            Any
+            RMS
+            All
+            None
+        end
+    end
+
+    subgraph OBV[Observers]
+        direction TB
+        Do
+        Finally
+        DoAndFinally
+        DoNothing
+        Property
+        Array
+        CircularBuffer
+        DigitalOutput
+        AnalogOutput
+        Serial
+    end
+
+    OBS --> OPR --> OBV
+```
 
 ### Creating Observables
 Observables are generally generated through factory methods provided by the Reactive class.
@@ -51,10 +181,19 @@ Hot observables emits the sequence when an observer subscribes to it. For exampl
 ```c++
 FromArray(values, valuesLength)
 ```
-Cold observable does not emits any item when a observer subscribes to it. You have to explicitly call the `Next()` method whenever you want. For example, `FromArrayDefer(...)`
+Cold observable does not emits any item when a observer subscribes to it. You have to explicitly call the `Next()` method whenever you want. For example, `FromArrayDefer(...)` or `ManualDefer(...)`.
 ```c++
 FromArrayDefer(values, valuesLength)
 ```
+
+With `ManualDefer(...)` you also have to explicitly call `Next()` to emit, and `Complete()` to finish the sequence.
+```c++
+auto obs = ManualDefer<int>();
+obs >> ToSerial<int>();
+// ...later in code
+obs.Next();
+obs.Complete();
+```
 ### Dynamic memory considerations
 On many occasions we generate operators directly when we chain them, for example in the `Setup()`. However, creating an operator allocates dynamic memory. Therefore, you should avoid creating them in `Loop()`, or you could run out of memory.
 If you need to reuse (typically, call some operator method later in your code) set it as a global variable, and chain as normal.

platformio.ini

@@ -0,0 +1,19 @@
+; PlatformIO Unit Testing configuration for the ReactiveArduino library.
+;
+; Run the tests with:
+;     pio test -e native
+;
+; The `native` environment compiles and runs the tests on the host machine.
+; Arduino hardware functions (millis, micros, pinMode, digitalWrite, analogWrite,
+; analogRead, digitalRead, String, Serial) are stubbed in test/WProgram.h so the
+; header-only library can be exercised without a board.
+
+[platformio]
+default_envs = native
+
+[env:native]
+platform = native
+test_framework = unity
+build_flags =
+    -Isrc
+    -Itest

src/Aggregates/AggregateAll.h

@@ -33,7 +33,7 @@ AggregateAll<T>::AggregateAll(ReactivePredicate<T> condition)
 template <typename T>
 void AggregateAll<T>::OnNext(T value)
 {
-	if (_state && _condition(value)) _state = false;
+	if (!_condition(value)) _state = false;
 
 	this->_childObservers.OnNext(_state);
 }

src/Aggregates/AggregateAverage.h

@@ -33,7 +33,7 @@ void AggregateAverage<T>::OnNext(T value)
 {
 	_sum += value;
 	_count++;
-	this->_childObservers.OnNext(_sum / _count);
+	this->_childObservers.OnNext(_sum / static_cast<T>(_count));
 }
 
 #endif

src/Aggregates/AggregateCount.h

@@ -19,7 +19,7 @@ class AggregateCount : public Operator<T, int>
 	void OnNext(T value) override;
 
 private:
-	int _count = false;
+	int _count = 0;
 };
 
 template <typename T>

src/Aggregates/AggregateCountdown.h

@@ -19,7 +19,8 @@ class AggregateCountdown : public Operator<T, int>
 	void OnNext(T value) override;
 
 private:
-	int _count = false;
+	int _count = 0;
+	bool _completed = false;
 };
 
 template <typename T>
@@ -31,11 +32,14 @@ AggregateCountdown<T>::AggregateCountdown(int count)
 template <typename T>
 void AggregateCountdown<T>::OnNext(T value)
 {
+	if (_completed) return;
+
 	_count--;
 	this->_childObservers.OnNext(_count);
 
 	if (_count <= 0)
 	{
+		_completed = true;
 		this->_childObservers.OnComplete();
 	}
 }

src/Aggregates/AggregateNone.h

@@ -33,7 +33,7 @@ AggregateNone<T>::AggregateNone(ReactivePredicate<T> condition)
 template <typename T>
 void AggregateNone<T>::OnNext(T value)
 {
-	if (!_condition(value)) _state = false;
+	if (_condition(value)) _state = false;
 
 	this->_childObservers.OnNext(_state);
 }

src/Aggregates/AggregateRMS.h

@@ -33,7 +33,7 @@ void AggregateRMS<T>::OnNext(T value)
 {
 	_sumSqr += value * value;
 	_count++;
-	this->_childObservers.OnNext(sqrt(_sumSqr / _count));
+	this->_childObservers.OnNext(static_cast<T>(sqrt(static_cast<double>(_sumSqr) / static_cast<double>(_count))));
 }
 
 #endif

src/Filters/FilterWindowMicros.h

@@ -32,9 +32,7 @@ void FilterWindowMicros<T>::OnNext(T value)
 	}
 
 	if (_started && static_cast<unsigned long>(micros() - _lastTrigger) <= _interval)
-	{
 		this->_childObservers.OnNext(value);
-	}
 }
 
 #endif

src/Filters/FilterWindowMillis.h

@@ -39,9 +39,7 @@ void FilterWindowMillis<T>::OnNext(T value)
 	}
 
 	if (_started && static_cast<unsigned long>(millis() - _lastTrigger) <= _interval)
-	{
 		this->_childObservers.OnNext(value);
-	}
 }
 
 #endif

src/Observables/ObservableIntervalMicros.h

@@ -33,7 +33,6 @@ class ObservableIntervalMicros : public Observable<unsigned long>
 
 private:
 	bool _isActive;
-	bool _isExpired;
 	unsigned long _startTime;
 	unsigned long _delay;
 	unsigned long _offset;
@@ -121,7 +120,9 @@ unsigned long ObservableIntervalMicros<T>::GetElapsedTime()
 template <typename T>
 unsigned long ObservableIntervalMicros<T>::GetRemainingTime()
 {
-	return _interval - micros() + _startTime;
+	unsigned long elapsed = micros() - _startTime;
+	if (elapsed >= _interval) return 0;
+	return _interval - elapsed;
 }
 
 template <typename T>

src/Observables/ObservableIntervalMillis.h

@@ -33,7 +33,6 @@ class ObservableIntervalMillis : public Observable<unsigned long>
 
 private:
 	bool _isActive;
-	bool _isExpired;
 	unsigned long _startTime;
 	unsigned long _delay;
 	unsigned long _offset;
@@ -121,7 +120,9 @@ unsigned long ObservableIntervalMillis<T>::GetElapsedTime()
 template <typename T>
 unsigned long ObservableIntervalMillis<T>::GetRemainingTime()
 {
-	return _interval - millis() + _startTime;
+	unsigned long elapsed = millis() - _startTime;
+	if (elapsed >= _interval) return 0;
+	return _interval - elapsed;
 }
 
 template <typename T>

src/Observables/ObservableRange.h

@@ -54,8 +54,16 @@ void ObservableRange<T>::UnSubscribe(IObserver<T> &observer)
 template<typename T>
 void ObservableRange<T>::Run()
 {
-	for (auto i = _start; i <= _end; i += _step)
-		this->_childObservers.OnNext(i);
+	if (_step > 0)
+	{
+		for (auto i = _start; i <= _end; i += _step)
+			this->_childObservers.OnNext(i);
+	}
+	else if (_step < 0)
+	{
+		for (auto i = _start; i >= _end; i += _step)
+			this->_childObservers.OnNext(i);
+	}
 
 	this->_childObservers.OnComplete();
 }

src/Observables/ObservableRangeDefer.h

@@ -53,13 +53,14 @@ void ObservableRangeDefer<T>::UnSubscribe(IObserver<T> &observer)
 template <typename T>
 void ObservableRangeDefer<T>::Next()
 {
-	if (_value > _end) return;
+	if (_step > 0 && _value > _end) return;
+	if (_step < 0 && _value < _end) return;
 
 	T value = _value;
 	this->_childObservers.OnNext(value);
 	_value += _step;
 
-	if (_value > _end)
+	if ((_step > 0 && _value > _end) || (_step < 0 && _value < _end))
 		this->_childObservers.OnComplete();
 }
 

src/Observables/ObservableSerialByte.h

@@ -15,8 +15,8 @@ class ObservableSerial<byte> : public Observable<byte>
 {
 public:
 	ObservableSerial();
-	void Subscribe(IObserver<byte> &observer);
-	void UnSubscribe(IObserver<byte> &observer);
+	void Subscribe(IObserver<byte> &observer) override;
+	void UnSubscribe(IObserver<byte> &observer) override;
 	void Receive();
 
 private:

src/Observables/ObservableSerialDouble.h

@@ -23,7 +23,7 @@ class ObservableSerial<double> : public Observable<double>
 private:
 	char _separator;
 
-	float _data = 0;
+	double _data = 0;
 	int _dataReal = 0;
 	int _dataDecimal = 0;
 	int _dataPow = 1;

src/Observables/ObservableSerialString.h

@@ -47,9 +47,7 @@ inline void ObservableSerial<String>::Receive()
 	{
 		const char newChar = Serial.read();
 		if (newChar != _separator)
-		{
 			_buffer.concat(newChar);
-		}
 		else
 		{
 			_childObservers.OnNext(_buffer);