|
|
|
|
Reactive programming is a "programming paradigm oriented around data flows and the propagation of change." Relationships can be established between producers, consumers, and transformers of data, with updates occurring asynchronously, rather than by request. This aids in application scalability and better utilization of hardware resources, as threads are not blocked waiting on potentially long-running operations. The result of a database query or the return value from a web service can be delivered to the application when the result is ready, rather than the application waiting for the result to arrive. That is, it is a "push" rather than "pull" model — one reacts to data as it arrives, as opposed to pulling it element by element from a data source, where efficiency is lost as the application waits for the next element to be available.
Previous paradigms, such as asynchronous I/O or event-driven programming, also provide for non-blocking servicing of data or events, but reactive programming unifies these models such that operations as different as reading a file, obtaining mouse movements, or receiving the response of a web request can be handled using the same small set of programming concepts.
Microsoft's Reactive Extensions for .NET (Rx.NET) provides a framework for reactive programming. Microsoft has released similar frameworks for other languages, including JavaScript (RxJS), C++ (RxCpp), Ruby (Rx.rb) and Python (RxPy), and other organizations have done the same for Java (RxJava), Scala (RxScala), Clojure (RxClojure) and more.
Reactive programming centers around two concepts: an observable and an observer. An
observable is a data stream, such as a sequence of stock quotes, mouse positions, or file blocks. An
observer watches an observable by subscribing to it, and is updated as the state of the observable changes.
Three operations can be invoked on an observer, reflecting the state of the observable:
OnNext(<data value>) when the watched observable produces
a new value, OnCompleted() when the observable has finished producing data, and
OnError(<exception>) if the observable has reached an error state.
Observables can be created in several ways, including:
OnNext() on a subject invokes the corresponding method on
its observers.We can use the techniques above to create observables from OpenDDS data sample streams. Consider data generated by a fictional sensor as represented by the following IDL:
// SensorIDL\Sensor.idl
module Sensor {
#pragma DCPS_DATA_TYPE "Sensor::SensorData"
#pragma DCPS_DATA_KEY "Sensor::SensorData sensorID"
struct SensorData {
string sensorID;
unsigned long long date;
double reading;
};
};
A sensor has a name, a time at which a data reading is collected, and the reading itself. Sensor
data readings are simulated by a C++-based publisher that generates a total of 20 readings. A value
is published every two seconds with the value of the reading ranging from 0 to 9, repeating.
These samples are published on the Temperature DDS topic. The
time of the sample is the current time represented as seconds since midnight on January
1, 1970 UTC. This publisher will be used for all of the OpenDDS examples in this article.
// SensorPublisher\SensorPublisher.cpp
Sensor::SensorDataDataWriter_var sdw =
Sensor::SensorDataDataWriter::_narrow(dw);
...
Sensor::SensorData sample;
sample.sensorID = "Temp7";
time_t sampleDate;
for (int i = 0; i < 20; i++) {
// seconds since midnight 1/1/1970 UTC
time(&sampleDate);
sample.date = sampleDate;
sample.reading = i%10;
if (sdw->write(sample, DDS::HANDLE_NIL) != DDS::RETCODE_OK)
throw DDSException("sample write() failed");
// sleep for 2 seconds
std::this_thread::sleep_for(std::chrono::milliseconds(2000));
}
To create a Rx.NET observable from an OpenDDS data stream, it is convenient to first make the OpenDDS data samples accessible to .NET before creating the observable. We do this in a manner similar to what has been shown in previous Middleware News Briefs by publishing a .NET event whenever a data sample arrives.
As OpenDDS is written in unmanaged C++, we must create a .NET adapter, class SensorEventPublisher, that can be
used from a .NET language such as C#. It is declared as a managed .NET class as follows,
with its implementation in the unmanaged, standard C++ class SensorEventPublisherImpl.
// SensorSubscriberLibEvent\SensorSubscriberLibEvent.cpp
public ref class SensorEventPublisher {
SensorEventPublisherImpl *_impl;
public:
event System::EventHandler<SensorEventLib::SensorEventArgs^>
^SensorEvent;
SensorEventPublisher(array<System::String^>^ args);
...
void Publish(System::String^ sensorID,
System::DateTime date, double reading);
};
SensorEvent is declared as a .NET event which publishes events of type SensorEventArgs via
the Publish() method defined as:
// SensorSubscriberLibEvent\SensorSubscriberLibEvent.cpp
void SensorEventPublisher::Publish(System::String^ sensorID,
System::DateTime date, double reading) {
SensorEvent(this,
gcnew SensorEventLib::SensorEventArgs(sensorID, date,
reading));
}
The sensor event argument structure is defined in its own library, so it can be accessed both by
SensorSubscriberLib and any .NET projects that wish to reference it. It inherits from
EventArgs, the .NET base class for classes which contain event data.
// SensorEventLib\SensorEventLib.cs
public class SensorEventArgs : EventArgs {
public readonly string SensorID;
public readonly DateTime Date;
public readonly double Reading;
public SensorEventArgs(string sensorID, DateTime date,
double reading) {
SensorID = sensorID;
Date = date;
Reading = reading;
}
public override string ToString()
{
// Date is in UTC, so convert to local time for display
return "[" + SensorID + ", " + Date.ToLocalTime() + ", " +
Reading + "]";
}
}
For convenience, a ToString() method is provided for pretty printing, converting the time of the
sample from UTC to the local time.
In the OpenDDS subscriber, a listener is associated with a data reader for the "Temperature" topic, and
is instantiated with a reference to a SensorEventPublisher.
// SensorSubscriberLibEvent\SensorSubscriberLibEvent.cpp
DDS::DataReaderListener_var listener(
new DataReaderListenerImpl(sensorEventPublisher));
DDS::DataReader_var dr =
sub->create_datareader(topic, dr_qos, listener,
OpenDDS::DCPS::DEFAULT_STATUS_MASK);
Data samples received within the on_data_available() method of the listener are converted to
.NET types and passed as parameters to Publish().
// SensorSubscriberLibEvent\SensorSubscriberLibEvent.cpp
DataReaderListenerImpl(
SensorEventPublisher ^sensorEventPublisher) :
_sensorEventPublisher(sensorEventPublisher) {}
...
void DataReaderListenerImpl::on_data_available(
DDS::DataReader_ptr reader) {
try {
DDS::SampleInfo info;
Sensor::SensorDataDataReader_var dr =
Sensor::SensorDataDataReader::_narrow(reader);
if (dr) {
Sensor::SensorData sample;
DDS::ReturnCode_t error =
dr->take_next_sample(sample, info);
if ((error == DDS::RETCODE_OK) && info.valid_data) {
_sensorEventPublisher->Publish(
gcnew System::String(sample.sensorID.in()),
System::DateTime(1970, 1, 1, 0, 0, 0,
System::DateTimeKind::Utc) +
System::TimeSpan::FromSeconds(sample.date),
sample.reading
);
}
}
}
catch (System::Exception ^e) {
System::Console::WriteLine(e);
}
}
The time of the data sample is converted into a TimeSpan, and that is added to
the epoch date to recover the absolute time of the data sample.
Now that the data sample is published as .NET event data, one of the built-in adapters of Rx.NET can be used
to convert the event to an observable. One such pattern, Observable.FromEventPattern<>,
converts the elements produced by an event to an observable, where, in our case, the SensorEvent
created by SensorEventPublisher is used as the event source.
We use the
Select() projection operator to transform the observable sequence by applying
a function to each element that is produced by an observable. Here, we use it to transform the element from what is returned by
FromEventPattern<> into the event argument structure itself.
// ReactiveDDSCSEvent\ReactiveDDSCSEvent.cs
var sevp = new SensorEventPublisher(args);
var obs = Observable.FromEventPattern<SensorEventArgs>(
eh => sevp.SensorEvent += eh,
eh => sevp.SensorEvent -= eh)
.Select(e => e.EventArgs);
The variable obs, of type IObservable<SensorEventArgs>, can now be subscribed to. One variant
of the Subscribe() method provides for only an OnNext() notification handler to be specified — here,
we only wish to see the samples printed on the screen.
using (obs.Subscribe(Console.WriteLine)) {
sevp.Run();
Console.WriteLine("Press ENTER to stop");
Console.ReadLine();
sevp.Stop();
}
If the OnCompleted() or OnError() notifications also need to be handled, handler functions can be specified
as the second and third parameters to Subscribe(), respectively. Subscribe() returns an object
representing the subscription, and this object implements the IDisposable interface. Disposing of the subscription unsubscribes
the observer from the observable, which is done automatically by the using clause when the code block exits.
The test script Test\testReactiveDDSCSEvent.pl starts the DCPSInfoRepo, the publisher (SensorPublisher),
the C#-based subscriber (ReactiveDDSCSEvent), and waits for them to terminate. The output of the test looks like the following:
[Temp7, 10/28/2014 10:29:20 AM, 1] [Temp7, 10/28/2014 10:29:22 AM, 2] [Temp7, 10/28/2014 10:29:24 AM, 3] ...
If one is interested only in the samples produced by an observable, the method above will work well. Unfortunately, a single Windows event doesn't have the means to represent completion or error states that the adapter can directly map. For this, we must publish through a subject. To take advantage of DDS features, we will use a subject whose elements are subjects that the data samples are pushed through.
We first rename the SensorEventPublisher class to SensorEventPublisherSubject for clarity,
and change it to take a subject of subjects as an argument:
Subject<Subject<SensorEventArgs^>^>^
Two namespaces are used to improve code readability as well.
// SensorSubscriberLibSubject\SensorSubscriberLibSubject.cpp
using namespace System::Reactive::Subjects;
using namespace SensorEventLib;
public ref class SensorEventPublisherSubject {
SensorEventPublisherSubjectImpl *_impl;
Subject<Subject<SensorEventArgs^>^>^ _obsGroup;
public:
SensorEventPublisherSubject(array<System::String^>^ args,
Subject<Subject<SensorEventArgs^>^>^ obsGroup);
...
};
The subject of subjects — an observable group — is passed to the data reader listener.
// SensorSubscriberLibSubject\SensorSubscriberLibSubject.cpp
DDS::DataReaderListener_var listener(
new DataReaderListenerImpl(obsGroup));
DDS::DataReader_var dr =
sub->create_datareader(topic, dr_qos, listener,
OpenDDS::DCPS::DEFAULT_STATUS_MASK);
The DataReaderListenerImpl stores both the observable group, as well as creates a map (a .NET Dictionary<>)
of observables, which will be described below. As they are .NET types that are stored in an unmanaged
C++ class, they must be wrapped with gcroot<>.
// SensorSubscriberLibSubject\SensorSubscriberLibSubject.cpp
class DataReaderListenerImpl : public virtual
OpenDDS::DCPS::LocalObject<DDS::DataReaderListener> {
gcroot<Subject<Subject<SensorEventArgs^>^>^> _obsGroup;
gcroot<Dictionary<long, Subject<SensorEventArgs^>^>^> _obsMap;
public:
DataReaderListenerImpl(
Subject<Subject<SensorEventArgs^>^>^ obsGroup) :
_obsGroup(obsGroup),
_obsMap(gcnew Dictionary<long,
Subject<SensorEventArgs^>^>()) {}
The on_data_available() method begins in a similar manner as before,
plus the observable group and map are set to local variables to "unwrap" them from
their gcroot<> holders, which simplifies their use syntactically as it
removes a layer of indirection.
// SensorSubscriberLibSubject\SensorSubscriberLibSubject.cpp
void DataReaderListenerImpl::on_data_available(
DDS::DataReader_ptr reader) {
try {
DDS::SampleInfo info;
Subject<Subject<SensorEventArgs^>^>^ obsGroup = _obsGroup;
Dictionary<long, Subject<SensorEventArgs^>^>^
obsMap = _obsMap;
Sensor::SensorDataDataReader_var dr =
Sensor::SensorDataDataReader::_narrow(reader);
if (dr) {
Sensor::SensorData sample;
DDS::ReturnCode_t error =
dr->take_next_sample(sample, info);
if (error == DDS::RETCODE_OK) {
In the IDL that defined the OpenDDS sample type SensorData, a key was established to
divide samples into instances by the sensor that generated them:
// SensorIDL\Sensor.idl #pragma DCPS_DATA_KEY "Sensor::SensorData sensorID"
Our data happens to publish only with a sample ID of Temp7, so only one instance is present in the sample data,
but in the general case, instances will appear arbitrarily in the sample stream, and must be handled as they arise.
As each new instance is detected, a new subject is created which will be used to push its samples. This subject is
added to the observable group, so subscribers can become aware that a new instance has been created. It is also
stored in an observable map, keyed by the instance handle, so it can be easily retrieved when additional samples on
the instance are to be pushed.
// if instance is not found in the map,
// create a new observable for it and push it out
Subject<SensorEventArgs^> ^obs = nullptr;
if (!obsMap->ContainsKey(info.instance_handle)) {
obs = gcnew Subject<SensorEventArgs^>();
obsGroup->OnNext(obs);
obsMap->Add(info.instance_handle, obs);
}
else
obs = obsMap[info.instance_handle];
Once the observable corresponding to the instance has been obtained (either newly created, or retrieved from the map),
it is updated with the state of the data. If the data is valid, then it is pushed via a call to OnNext(),
but if the data isn't valid, and the instance has been disposed and will no longer publish again, then OnCompleted()
will be invoked on the observable.
// update the observable
if (info.valid_data) {
// data is valid - call OnNext with the data
obs->OnNext(gcnew SensorEventLib::SensorEventArgs(
gcnew System::String(sample.sensorID),
System::DateTime(1970, 1, 1, 0, 0, 0,
System::DateTimeKind::Utc) +
System::TimeSpan::FromSeconds(sample.date),
sample.reading));
}
else if (info.instance_state ==
DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE) {
// no more data - call OnCompleted on the observable
obs->OnCompleted();
}
Lastly, if a sample could not be taken, we can consider the listener to have failed, and OnError() is called on
all observables in the map, as all observables from this point forward are considered to have failed as the listener would no longer
be able to update them.
}
else {
// can't take the sample - assume we're dead and
// set an error state on all observables
System::Exception ^ex = gcnew
System::InvalidOperationException(
"take_next_sample() failed");
for each(auto k in obsMap->Keys)
obsMap[k]->OnError(ex);
}
}
}
catch (System::Exception ^e) {
System::Console::WriteLine(e);
}
}
The C# subscriber is somewhat different than before, due to the grouping of the observables. A subject
of subjects is created as an observable group, and passed into the instantiated SensorEventPublisherSubject.
The group is subscribed to, and each subject in the group (corresponding to each DDS instance) is also
subscribed to. Two handlers are supplied, rather than just one, as arguments to Subscribe().
The first argument to Subscribe()
is the OnNext() handler, and, as before, the sample is displayed via the call to
Console.WriteLine(). The second parameter is an OnCompleted() handler,
which, when invoked, displays a completion message, and signals an EventWaitHandle to indicate
that the application should terminate as the observable has completed. In the general case, the application
shouldn't terminate when just one observable does, but here, as the data only has one instance, only
one observable to push data samples is created, so the application can safely terminate when it reaches the completed state.
// ReactiveDDSCSSubject\ReactiveDDSCSSubject.cs
var obsGroup = new Subject<Subject<SensorEventArgs>>();
var sevp = new SensorEventPublisherSubject(args, obsGroup);
var waitHandle = new EventWaitHandle(false,
EventResetMode.AutoReset);
using (obsGroup.Subscribe(obs => obs.Subscribe(
Console.WriteLine,
() =>
{
System.Console.WriteLine("Completed");
waitHandle.Set();
})))
{
sevp.Run();
bool signaled = false;
do
{
signaled = waitHandle.WaitOne(TimeSpan.FromSeconds(1));
} while (!signaled);
sevp.Stop();
}
The output of the subject-based example, via the invocation of the Test\testReactiveDDSCSSubject.pl script,
is the same as the event-based example, save for the additional Completed message following the data samples.
We can do the same with RxCpp, by updating a group of observables. The on_data_available() method of the
data reader, given the observable group and map, first obtains the observable for the instance from the map, creating
the observable if it is not found:
// ReactiveDDSCPP\ReactiveDDSCPP.cpp
namespace rxsub = rxcpp::subjects;
...
if (dr) {
Sensor::SensorData sample;
DDS::ReturnCode_t error = dr->take_next_sample(sample, info);
if (error == DDS::RETCODE_OK) {
// if instance is not found in the map, create a new
// observable for it and push it out
auto obs = rxsub::subject<Sensor::SensorData>();
auto obsIt = _obsMap.find(info.instance_handle);
if (obsIt == _obsMap.end()) {
_obsGroup.get_subscriber().on_next(obs);
_obsMap[info.instance_handle] = obs;
}
else
obs = obsIt->second;
Then, if the data is valid, the observable is updated with the data via on_next(), otherwise,
if the instance has been disposed, mark the observable as completed with on_completed().
// update the observable
if (info.valid_data) {
// data is valid - call OnNext with the data
obs.get_subscriber().on_next(sample);
}
else if (info.instance_state ==
DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE) {
// no more data - call OnCompleted on the observable
obs.get_subscriber().on_completed();
}
Lastly, if the sample could not be taken, set an error state in all observables via on_error().
}
else {
// can't take the sample - assume we're dead and set
// an error state on all observables
std::exception_ptr ex = std::make_exception_ptr(
std::runtime_error("take_next_sample() failed"));
std::for_each(_obsMap.begin(), _obsMap.end(),
[ex](std::pair<const DDS::InstanceHandle_t,
rxsub::subject<Sensor::SensorData>> &obs){
obs.second.get_subscriber().on_error(ex);
});
}
}
Subscribing to the observable group is done in a similar way, as well. Each instance is subscribed to also by
providing two functions, one to handle OnNext() and the other to handle OnCompleted(),
with default behavior for the error state.
// ReactiveDDSCPP\ReactiveDDSCPP.cpp
rxsub::subject<rxsub::subject<Sensor::SensorData>> obsGroup;
DDS::DataReaderListener_var listener(
new DataReaderListenerImpl(obsGroup));
...
obsGroup.get_observable()
.subscribe(
[](rxsub::subject<Sensor::SensorData> &obs) {
obs.get_observable().subscribe(
[](Sensor::SensorData &sd) {
time_t date = sd.date;
std::string cdate(ctime(&date));
cdate.erase(std::remove_if(cdate.begin(), cdate.end(),
[](char c) { return c == 0x0A; }), cdate.end());
std::cout << "[" << sd.sensorID << " " << cdate << " " <<
sd.reading << "]" << std::endl;
},
[]() { std::cout << "Completed" << std::endl; }
);
});
The lambda function that pretty prints the data sample uses ctime()
to format the date, but strips the extraneous line feed character from the string it returns for
nicer display.
The test script Test\testReactiveDDSCPP.pl starts the DCPSInfoRepo, the publisher (SensorPublisher),
the C++-based subscriber (ReactiveDDSCPP), and waits for them to terminate. The output of the test looks very similar
to the Rx.NET example:
[Temp7 Tue Oct 28 10:30:33 2014 1] [Temp7 Tue Oct 28 10:30:35 2014 2] [Temp7 Tue Oct 28 10:30:37 2014 3] [Temp7 Tue Oct 28 10:30:39 2014 4] ... Completed
We can do the same once again with RxJava. As before, the on_data_available() method of the
data reader, given the observable group and map, first obtains the observable for the instance from the map, creating
the observable if it is not found. Several types of subjects are available in RxJava, but we use a PublishSubject
as it imposes no other constraints on the elements it pushes, such as adding default values, replays, or such.
// ReactiveDDSJ\src\main\java\com\ociweb\DataReaderListenerImpl.java
if (dr != null) {
SensorDataHolder mh = new SensorDataHolder(new SensorData());
SampleInfoHolder sih = new SampleInfoHolder(
new SampleInfo(0, 0, 0,
new DDS.Time_t(), 0, 0, 0, 0, 0, 0, 0, false));
int status = dr.take_next_sample(mh, sih);
if (status == RETCODE_OK.value) {
// if instance is not found in the map, create a new
// observable for it and push it out
PublishSubject<SensorData> obs = null;
if (!_obsMap.containsKey(sih.value.instance_handle)) {
obs = PublishSubject.create();
_obsGroup.onNext(obs);
_obsMap.put(sih.value.instance_handle, obs);
} else
obs = _obsMap.get(sih.value.instance_handle);
Again, as before, the state of the observable corresponding to the instance is updated with calls to
onNext(), onCompleted(), and onError() as appropriate.
// update the observable
if (sih.value.valid_data) {
// data is valid - call OnNext with the data
obs.onNext(mh.value);
} else if (sih.value.instance_state ==
NOT_ALIVE_DISPOSED_INSTANCE_STATE.value) {
// no more data - call OnCompleted on the observable
obs.onCompleted();
}
} else {
// can't take the sample - assume we're dead and set an
// error state on all observables
Exception e = new IllegalStateException();
for (Integer k : _obsMap.keySet()) {
_obsMap.get(k).onError(e);
}
}
}
Once again, we do the same as above by first subscribing to the observable group, and then
to each instance observable. Three handlers must be provided this time, however, due to
the parameters of subscribe(). One version of subscribe() requires
only an OnNext() handler, but the other needs OnNext(), OnError()
and OnCompleted() handlers provided in that order. Therefore in order to supply an OnCompleted()
handler, all three must be provided. In addition, a handler cannot
be null, so an action with an empty body must be given as the OnError()
handler.
Also, to format the sample time correctly,
a java.util.Date() is created by passing the number of milliseconds since midnight 1/1/1970 UTC
as an argument to its constructor.
// ReactiveDDSJ\src\main\java\com\ociweb\ReactiveDDSJ.java
PublishSubject<PublishSubject<SensorData>> obsGroup =
PublishSubject.create();
DataReaderListenerImpl listener =
new DataReaderListenerImpl(obsGroup);
...
obsGroup.subscribe(new Action1<PublishSubject<SensorData>>() {
@Override
public void call(PublishSubject<SensorData> obs) {
obs.subscribe(
new Action1<SensorData>() {
@Override
public void call(SensorData sd) {
System.out.println("[" + sd.sensorID + " " +
new java.util.Date(sd.date*1000) + " " +
sd.reading + "]");
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable error) {
// no action, but handler can't be null
}
}, new Action0() {
@Override
public void call() {
System.out.println("Completed");
}
});
}
});
If Java 8 lambdas are available, the code becomes much simpler.
// ReactiveDDSJ\src\main\java\com\ociweb\ReactiveDDSJ.java
obsGroup.subscribe((obs) -> {
obs.subscribe(
(sd) -> {
System.out.println("[" + sd.sensorID + " " +
new java.util.Date(sd.date*1000) + " " +
sd.reading + "]");
},
(ex) -> {},
() -> { System.out.println("Completed"); }
);
});
The test script Test\testReactiveDDSJ.pl starts the DCPSInfoRepo, the publisher (SensorPublisher),
the Java-based subscriber (ReactiveDDSJ), and waits for them to terminate. The output of the test looks very similar
to the previous examples:
[Temp7 Tue Oct 28 10:30:58 CDT 2014 1.0] [Temp7 Tue Oct 28 10:31:00 CDT 2014 2.0] [Temp7 Tue Oct 28 10:31:02 CDT 2014 3.0] [Temp7 Tue Oct 28 10:31:04 CDT 2014 4.0] [Temp7 Tue Oct 28 10:31:06 CDT 2014 5.0] ... Completed
Although the OpenDDS publisher above simulated a physical sensor, we can simulate incoming data at even a higher level. The result of the above process was to produce an observable that produced 20 values, each two seconds apart. We can take advantage of the generation capability of observers as provided by the various frameworks to mock OpenDDS itself, which can be useful for testing. As described in talks such as this one, this is a general technique. In that talk, Bart de Smet described an observable of points obtained from mouse movements, but as it does not matter to the code where the points themselves originated, as any observable of points, such as one provided in a unit test, would suffice.
Rx.NET provides various overloaded Generate() methods on Observable. The version
below takes 5 parameters: A starting value for a numeric index (0), a function that returns true as long as the
generation should continue (stop after generating 20 samples), a function that modifies
the index (increase it by 1), a function that returns the type produced by the generation (the same
SensorEventArgs used previously, though could also be, say, an anonymous class),
and lastly, the interval between items generated (2 seconds).
// ReactiveCS\ReactiveCS.cs
var o = Observable.Generate(
0,
i => i < 20,
i => i + 1,
i => new SensorEventLib.SensorEventArgs(
"Temp7", DateTime.Now, i%10),
i => TimeSpan.FromSeconds(2)
);
For RxCpp, we could use the SensorData struct as generated from the IDL, though, for
this example, we'll create a similar structure.
// ReactiveCPP\ReactiveCPP.cpp
typedef std::chrono::system_clock clk;
typedef std::chrono::time_point<clk> tp;
struct SensorData {
std::string _sensorID;
tp _date;
double _reading;
SensorData(const std::string &sensorID, const tp &date,
const double reading) :
_sensorID(sensorID), _date(date), _reading(reading) {}
};
RxCpp provides an interval() method of observable<> which can be used for
time-based generation. It produces an increasing count, as Observable.Generate() did above,
but without end. We previously have seen the projection operator (via Select() as used in the Rx.NET OpenDDS
example), but an additional operator is needed here. The take() operator limits the number of samples from
an observable to the specified number, and map(), the equivalent of Select(),
applies a function to each element of the
generated sequence and is used here to convert the sequence into a different type.
Here, map() is used to transform the observable from producing a sequence
of int to a sequence of SensorData.
// ReactiveCPP\ReactiveCPP.cpp
auto sc = rxcpp::schedulers::make_current_thread();
auto so = rxcpp::synchronize_in_one_worker(sc);
auto o =
rxcpp::observable<>::interval(sc.now(),
std::chrono::seconds(2), so)
.take(20)
.map([](int i) {
return SensorData("Temp7", clk::now(), i%10);
});
Simulating the OpenDDS sample generation in RxJava is very similar to that of RxCpp. As before, we could have used the SensorData class generated from the IDL, but we'll use a custom one for convenience.
// ReactiveJ\src\main\java\com\oci\SensorData.java
public class SensorData {
public final String sensorID;
public final Date date;
public final double reading;
public SensorData(String sensorID, Date date,
double reading) {
this.sensorID = sensorID;
this.date = date;
this.reading = reading;
}
}
Creating a time-based observable follows the same pattern as in RxCpp. The interval() method of Observable
generates an observable that generates a sequence of integers, with each integer at two second intervals. The take()
operator limits the sequence to 20 elements, and the map() operator generates an instance of SensorData
corresponding to each integer.
// ReactiveJ\src\main\java\com\oci\ReactiveJ.java
Observable<SensorData> obs =
Observable.interval(2, TimeUnit.SECONDS)
.take(20)
.map(new Func1<Long, SensorData>() {
@Override
public SensorData call(Long i) {
return new SensorData("Temp7", new Date(), i%10);
}
});
obs.subscribe(new Action1<SensorData>() {
@Override
public void call(SensorData o) {
System.out.println("[" + o.sensorID + " " + o.date +
" " + o.reading + "]");
}
});
Implicit in the code above is the use of a scheduler, a mechanism that controls when subscriptions start,
notifications are published, and provides a notion of time. By default, a scheduler is used that is appropriate for
the context (such as by using the current thread of the process), and provides a real time clock. Rx.NET also
provides a
virtual time scheduler
in which "time" can pass at a rate managed by a test environment. A clock time,
specified in ticks (one ten-millionth of a second), is supplied as an argument to OnNext(),
OnCompleted(), or OnError() to indicate when the notification is produced.
These times are used to only sequence the operations, rather than, as with the timed
sequence generation above, to execute at a specific real-world clock time. Long running observables, by the use
of a virtual time scheduler, can now be tested at expected unit test speeds.
In Rx.NET, a useful virtual time scheduler is the TestScheduler, with methods to create observables
and observers that "tick" at the virtual time rate.
Two types
of observables can be created from a TestScheduler, a cold observable and a hot observable.
A cold observable publishes items only when an observer has subscribed, while a hot observable publishes regardless.
While a hot observable may more closely model a real-world device, such as a temperature sensor that publishes a
data reading every minute no matter whether anyone is listening, the use of a cold observable in a unit test eliminates
any timing issues in the test setup, to ensure no samples are lost before processing can be performed.
Creating an observable from a TestScheduler is done by providing a list of observer notifications in
the call to CreateHotObservable() or CreateColdObservable(). For instance, an observable
that produces the integer value 42 and then terminates can be modeled as:
// ReactiveCS\ReactiveCS.cs
var scheduler = new TestScheduler();
var xs = scheduler.CreateColdObservable(
new Recorded<Notification<int>>(100,
Notification.CreateOnNext(42)),
new Recorded<Notification<int>>(200,
Notification.CreateOnCompleted<int>())
);
The syntax can be simplified by executing the code from within a class derived from ReactiveTest.
// ReactiveCS\ReactiveCS.cs
public class Tests : ReactiveTest {
...
public void Produce42() {
var scheduler = new TestScheduler();
var xs = scheduler.CreateColdObservable(
OnNext(100, 42),
OnCompleted<int>(200)
);
...
So, to simulate OpenDDS, we can do the above, but, rather than a sequence of integers,
produce a sequence of SensorData or SensorEventArgs structures instead.
// ReactiveCS\ReactiveCS.cs
var scheduler = new TestScheduler();
var xs = scheduler.CreateColdObservable(
OnNext(100, new SensorEventArgs("Temp7", new DateTime(100),
42.0)),
OnCompleted<SensorEventArgs>(200)
);
In addition, to simulate real-world times, tick values corresponding to meaningful dates should be used. Small
numbers such as 100 and 200 are handy for clarity in simple tests, but an expression such as
new DateTime(2014, 1, 1, 0, 0, 0).Ticks can be used instead to obtain the tick count corresponding
to a point in time that would match the OpenDDS sample data being simulated. In this case, using this expression
to obtain the number of
ticks corresponding to midnight on January 1, 2014 is clearer than hardcoding the value 635241312000000000.
Once an OpenDDS data sample stream is transformed into an observable, it can be treated in the same manner as observables created from other sources of data in the application: mouse movements, file reads, database query result returns, and the like. That is, once these disparate types are all converted into observables, they can be subscribed to, transformed, filtered, and otherwise manipulated in a uniform way. This can lead to an application which is more streamlined, simpler to understand, and can provide the user with greater responsiveness.
While this article has shown how to transform OpenDDS data streams into observables, Part II will describe various operators that manipulate reactive streams, demonstrating the full power of reactive programming.