Range-based for loops and observables
In this section, we will implement range-based for loops on a custom type written by us to help you understand how all the things mentioned earlier in this chapter can be put together to write programs that support modern idioms. We will implement a class that returns a series of numbers within a bound and will implement infrastructure support for the iteration of the values based on range-based for loops. First, we write the "Iterable/Iterator" (aka "Enumerable/Enumerable") version by leveraging the range-based for loops. After some tweaks, the implementation will be transformed to Observable/Observer (the key interface of Reactive Programming) patterns: The implementation of Observable/Observer pattern here is just for elucidation purpose and should not be considered as an Industrial strength implementation of these patterns.
The following iterable class is a nested class:
// Iterobservable.cpp
// we can use Range Based For loop as given below (see the main below)
// for (auto l : EnumerableRange<5, 25>()) { std::cout << l << ' '; }
// std::cout << endl;
#include <iostream>
#include <vector>
#include <iterator>
#include <algorithm>
#include <functional>
using namespace std;
template<long START, long END>
class EnumerableRange {
public:
class iterable : public std::iterator<
std::input_iterator_tag, // category
long, // value_type
long, // difference_type
const long*, // pointer type
long> // reference type
{
long current_num = START;
public:
reference operator*() const { return current_num; }
explicit iterable(long val = 0) : current_num(val) {}
iterable& operator++() {
current_num = ( END >= START) ? current_num + 1 :
current_num - 1;
return *this;
}
iterable operator++(int) {
iterable retval = *this; ++(*this); return retval;
}
bool operator==(iterable other) const
{ return current_num == other.current_num; }
bool operator!=(iterable other) const
{ return !(*this == other); }
};
The preceding code implements an inner class derived from std::iterator to take care of the requirements for a type to be enumerable through range-based for loops. We will now write two public methods, (begin() and end()), so consumers of the class can use range-based for loops:
iterable begin() { return iterable(START); }
iterable end() { return iterable(END >= START ? END + 1 :
END - 1); }
};
Now, we can write code to consume the preceding class as follows:
for (long l : EnumerableRange<5, 25>())
{ std::cout << l << ' '; }
In the previous chapter, we defined the IEnumerable<T> interface. The idea was to stick with the documentation of Reactive eXtensions. The iterable class is very similar to the IEnumerable<T> implementation in the previous chapter. As outlined in the previous chapter, the preceding class can be made push based, if we tweak the code a bit. Let us write an OBSERVER class that contains three methods. We will be using Function Wrappers available with standard library to define the methods:
struct OBSERVER {
std::function<void(const long&)> ondata;
std::function<void()> oncompleted;
std::function<void(const std::exception &)> onexception;
};
The ObservableRange class given here contains a vector<T> that stores the list of subscribers. When a new number is generated, the event will be notified to all subscribers. If we dispatch the notification call from an asynchronous method, the consumer is decoupled from the producer of the range stream. We have not implemented the IObserver/IObserver<T> interface for the following class, but we can subscribe to notifications through subscribe methods:
template<long START, long END>
class ObservableRange {
private:
//---------- Container to store observers
std::vector<
std::pair<const OBSERVER&,int>> _observers;
int _id = 0;
We will store the list of subscribers in an std::vector as an std::pair. The first value in the std::pair is the reference to the OBSERVER and the second value in the std::pair is an integer that uniquely identifies the subscriber. Consumers are supposed to unsubscribe by using the ID returned by the subscribe method:
//---- The following implementation of iterable does
//---- not allow to take address of the pointed value [ &(*it)
//---- Eg- &(*iterable.begin()) will be ill-formed
//---- Code is just for demonstrate Obervable/Observer
class iterable : public std::iterator<
std::input_iterator_tag, // category
long, // value_type
long, // difference_type
const long*, // pointer type
long> // reference type
{
long current_num = START;
public:
reference operator*() const { return current_num; }
explicit iterable(long val = 0) : current_num(val) {}
iterable& operator++() {
current_num = ( END >= START) ? current_num + 1 :
current_num - 1;
return *this;
}
iterable operator++(int) {
iterable retval = *this; ++(*this); return retval;
}
bool operator==(iterable other) const
{ return current_num == other.current_num; }
bool operator!=(iterable other) const
{ return !(*this == other); }
};
iterable begin() { return iterable(START); }
iterable end() { return iterable(END >= START ? END + 1 : END - 1); }
// generate values between the range
// This is a private method and will be invoked from the generate
// ideally speaking, we should invoke this method with std::asnyc
void generate_async()
{
auto& subscribers = _observers;
for( auto l : *this )
for (const auto& obs : subscribers) {
const OBSERVER& ob = obs.first;
ob.ondata(l);
}
}
//----- The public interface of the call include generate which triggers
//----- the generation of the sequence, subscribe/unsubscribe pair
public:
//-------- the public interface to trigger generation
//-------- of thevalues. The generate_async can be executed
//--------- via std::async to return to the caller
void generate() { generate_async(); }
//---------- subscribe method. The clients which
//----------- expects notification can register here
int subscribe(const OBSERVER& call) {
// https://en.cppreference.com/w/cpp/container/vector/emplace_back
_observers.emplace_back(call, ++_id);
return _id;
}
//------------ has just stubbed unsubscribe to keep
//------------- the listing small
void unsubscribe(const int subscription) {}
};
int main() {
//------ Call the Range based enumerable
for (long l : EnumerableRange<5, 25>())
{ std::cout << l << ' '; }
std::cout << endl;
// instantiate an instance of ObservableRange
auto j = ObservableRange<10,20>();
OBSERVER test_handler;
test_handler.ondata = [=](const long & r)
{cout << r << endl; };
//---- subscribe to the notifiactions
int cnt = j.subscribe(test_handler);
j.generate(); //trigget events to generate notifications
return 0;
}