Musings on RxAda

Musings on RxAda

[Alejandro R. Mosteo]

Hello all,

having been recently working on Java/Android apps, I have been exposed to 
the ReactiveX framework [1]. I think it originated on C# though.

[1] http://reactivex.io/

Anyway, the gist of it is that you chain function calls (operators in their 
jargon) to process data asynchronously. This, in Java, with lambda functions 
and parameter inference leads to extremely compact, legible code that may 
notably simplify multithreading code.

But I'm not here to advocate RxJava. For now I see it as an interesting 
challenge from an implementor point of view. When using it I started to 
think how this could be implemented in Ada. My ideas are not very elegant. 
I'd like to know your opinion/ideas on if this is doable in Ada (which could 
lead to a potential RxAda library ;-)).

For simplicity, I'm going to concentrate on the "map" operator, as seen on 
this Java example:

Observable.just("Hello, world!")
    .map(s -> s + " -Dan")
    .map(s -> s.hashCode())
    .map(i -> Integer.toString(i))
    .subscribe(s -> System.out.println(s));

Basically, the map operation takes some input, changes it somehow and 
outputs another type. The chain begins at an Observable (some data 
generator) and ends at a subscriber (which does something with the result). 
The above example takes a string, appends a signature, hashes it, and 
outputs the hash as a string. This does not necessarily has to happen at the 
moment of declaration; in general, observables emit data asynchronously.

Moving into Ada, we need an Observable type able to take different map 
implementations. That could be (not compiled, bear with my mistakes. Allow 
for 2012 syntax):

type Observable is [limited?] tagged record;
type Datum is abstract interface; -- or maybe tagged null record;

function Map (This : in out Observable; 
              Map  : access function (X : Datum'Class) return Datum'Class)
return Observable; -- or access Observable if limited

Here the 'Class are needed to avoid multiple dispatch, I think. Then you'll 
have to declare beforehand any mappings you need (which is more verbose and, 
lacking lambdas, would somehow defeat the purpose of having the logic in the 
chain declaration, but I see no way around it, unless the new simple 
expression functions can appear in-place?). Also, the need for 'Class 
parameters will force explicit casts in the user mappings, which I find ugly 
and leaves the type checks to runtime.

I guess an implementation of "RxAda" could provide many Map profile 
overloads for basic Ada types (from String to String and so on), but still 
that's a poor's man attempt.

That's as far as I've got, which is not much, but I'm a bit rusty on Ada 
right now. I can think of even more contrived ways using tagged types plus 
generics but I'm not sure they lead anywhere. I think it boils down to Ada 
not having lambda functions nor implicit template types a-la C++.

Your thoughts welcome!

[1] http://reactivex.io/

Re: Musings on RxAda

[brbarkstrom]

This approach may also tie in with the communities working on "Asynchronicity".
They would like to create interfaces that identify "Promises" that appear to
be unelaborated tasks.  When a user responds to a message, the controlling
task elaborates the fulfilled promise (of a response to an asynchronous
message and lets the elaborated task work.

See, for example,
<crockford.com/pp/asynchronicity.pptx>
or
<http://webapplog.com/asynchronicity-in-node-js/>
A site that discusses "Promises" in javascript is
<http://12devs.co.uk/articles/promises-an-alternative-way-to-approach-asynchronous-javascript/>
It looks like the use of "Promises" is to provide a low cost way of
formalizing responses to asynchronous I/O.

Bruce B.

Re: Musings on RxAda

[Hadrien Grasland]

Interesting problem ! Let me try to play with it too.

If you need asynchronicity, the standard Ada answer is tasks. So here, you would like an "observable task", which can hold data of any type, perform any action on this data, and return data of another (possibly different) type.

The type-safe and statically checkable way to hold data of any type is to build a generic package. So I would argue that this is the idiomatic Ada way to solve this problem, and a better alternative to the type erasure you attempt to implement using null interfaces. It also has the important advantage of working with built-in Ada tyes.

If try to make a generic task which fits our needs, we might get something like this as a first attempt (NOTE : This code is wrong, read below to know why) :

====================

generic
   type Input_Data is private;
   type Output_Data is private;
package Observables is
   
   -- Any valid mapping function for this observable can be matched by this access type
   type Mapping_Function is access function (Input : Input_Data) return Output_Data;
   
   -- Any subscribing procedure can be matched by this access type
   type Subscribing_Proc is access procedure (Input : Input_Data);
   
   -- This task performs the asynchronous mapping
   task type Observable is
      entry Just (Data : Input_Data; Output : access Observable);
      entry Just (Data : Input_Data; Action : Subscribing_Proc);
      entry Map (Mapping : Mapping_Function; Output : access Observable);
      entry Map (Mapping : Mapping_Function; Action : Subscribing_Proc);
   end Observable;
   
end Observables;

====================

Notice that Map and Just must return results through a parameter, because Ada task entries cannot be functions. This is somewhat annoying, but probably motivated by the fact that a return in an accept statement would be highly confusing.

Notice also that I have removed Subscribe, and replaced it by overloadings of Just and Map. That is because an Ada task may return results immediately within the bodies of Just and Map, in which case subscribing later would only be a recipe for overhead and race conditions.

More problematic, however, are the facts that...

   1/ This code is illegal. Task entries Map and Just cannot take an access parameter.
   2/ Even if it were legal, it wouldn't do what we want. The Observer that would be passed as a parameter to Map would expect Input_Type as input, whereas we would want it to accept Output_Type as input and do not care about its output.

The "do not care" statement above is a telltale sign that we need another layer of abstraction. We need something that can hold data (results) of a given type, is thread-safe, and allows us not to think about the data after it is sent.

I would argue that this something is a protected object :

====================

generic
   type Data_Type is private;
package Async_Data is
   
   -- This container is promised to hold a chunk of data someday
   protected type Datum is
      procedure Send (Result : Data_Type);
      entry Receive (Result : out Data_Type);
   private
      Data_Ready : Boolean := False;
      Data : Data_Type;
   end Datum;
   
end Async_Data;

====================

This is really a textbook example of a protected type, and indeed you will find one like this in almost all Ada textbook. Nevertheless, for the sake of completeness, here is an implementation :

====================

package body Async_Data is
   
   protected body Datum is
      
      procedure Send (Result : Data_Type) is
      begin
         Data := Result;
         Data_Ready := True;
      end Send;
      
      entry Receive (Result : out Data_Type) when Data_Ready is
      begin
         Result := Data;
      end Receive;
      
   end Datum;
   
end Async_Data;

====================

Okay, so now we have asynchronous communication channels. We can use them to rewrite the Observable code above so that its Map function actually sends data to an asynchronous output channel, and that it can accept asynchronous input as well, like so :

====================

with Async_Data;

generic
   with package Async_Input is new Async_Data (others => <>);
   with package Async_Output is new Async_Data (others => <>);
package Observables is
   
   -- Let's clarify our input and output data types
   subtype Input_Data is Async_Input.Data_Type;
   subtype Output_Data is Async_Output.Data_Type;
   
   -- Any valid mapping function for this observable can be matched by this access type
   type Mapping_Function is access function (Input : Input_Data) return Output_Data;
   
   -- Any subscribing procedure can be matched by this access type
   type Subscribing_Proc is access procedure (Data : Output_Data);
   
   -- This task performs the asynchronous mapping
   task type Observable (Input : access Async_Input.Datum) is
      entry Map (Mapping : Mapping_Function; Output : in out Async_Output.Datum);
      entry Map (Mapping : Mapping_Function; Action : Subscribing_Proc);
   end Observable;
   
end Observables;

====================

Notice that I have suppressed the "Just" entry, whose complexity has become unnecessary due to our use of asynchronous input objects. Also, here is a possible implementation :

====================

package body Observables is

   task body Observable is
      Data : Input_Data;
   begin
      Input.Receive (Data);
      select
         accept Map (Mapping : Mapping_Function; Action : Subscribing_Proc) do
            Action.all (Mapping (Data));
         end Map;
      or
         accept Map (Mapping : Mapping_Function; Output : in out Async_Output.Datum) do
            Output.Send (Mapping (Data));
         end Map;
      end select;
   end Observable;
   
end Observables;

====================

How would we use such an object ? Well, any self-respecting would have a bunch of predefined instances of these generic packages for standard types, like this one :

====================

with Async_Data;
with Observables;

package Predefined is
   
   -- In a real library, basic asynchronous data types like this would be predefined
   package Async_Characters is new Async_Data (Character);
   subtype Async_Character is Async_Characters.Datum;
   package Async_Naturals is new Async_Data (Natural);
   subtype Async_Natural is Async_Naturals.Datum;
   
   -- Same for basic observables like this
   package Char_To_Nat_Observables is new Observables (Async_Input  => Async_Characters,
                                                       Async_Output => Async_Naturals);
   subtype Char_To_Nat_Observable is Char_To_Nat_Observables.Observable;
   package Nat_To_Char_Observables is new Observables (Async_Input  => Async_Naturals,
                                                       Async_Output => Async_Characters);
   subtype Nat_To_Char_Observable is Nat_To_Char_Observables.Observable;
   
end Predefined;

====================

And we would then want to use them like this :

====================

with Ada.Text_IO;
with Predefined;

procedure Main is
   -- Here, we will try to map a character to a natural and back, then print the result
   function Char_To_Nat_Mapping (Char : Character) return Natural is (Character'Pos (Char));
   function Nat_To_Char_Mapping (Nat : Natural) return Character is (Character'Val (Nat));
   
   procedure Print_Char (Char : Character) is
   begin
      Ada.Text_IO.Put (Char);
   end Print_Char;
   
   -- Let's try it !
   Input : constant Character := 'a';
   Char_To_Nat : Predefined.Char_To_Nat_Observable (new Predefined.Async_Character);
   Nat_To_Char : Predefined.Nat_To_Char_Observable (new Predefined.Async_Natural);
begin
   Char_To_Nat.Input.Send (Input);
   Char_To_Nat.Map (Char_To_Nat_Mapping'Access, Nat_To_Char.Input.all);
   Nat_To_Char.Map (Nat_To_Char_Mapping'Access, Print_Char'Access);
end Main;

====================

Sadly, this does not work. The Ada accessibility rules prevent us to use the access to subprograms Char_To_Nat_Mapping, Nat_To_Char_Mapping, and Print_Char, on the grounds that the task might subsequently leak pointers to nonexistent subprograms. So we would need to declare our functions in a separate package.

And this is where I will stop, because I think I have made my point : yes, it is possible to implement something like the Observers you mentioned in Ada. But I think it is highly unlikely to be anywhere near as practical as it is in C# or Java.

The reason lies in Ada's strict enforcement of pointer accessibility rules, including for function pointers, something which garbage-collected languages need not care about but which is of critical importance for languages where scope determines variable and function lifetimes.

Have a nice day !
Hadrien

Re: Musings on RxAda

[Hadrien Grasland]

Reading through my post again, I should 1/Apologize for the missing words here and there and 2/Point out that my implementation of Observable is mostly synchronous, because all the costly mapping work occurs within the accept rendezvous.

To make an asynchronous Observer would require quite a couple more tweaks, allowing the Observer task to save its entry parameters and use them once the computation is over. In particular, the output channel should be passed by access, rather than by in out value.

This does not, however, void my core point that Ada's accessibility rules would end up getting in our way if we wanted to implement something like ReactiveX for Ada.

Re: Musings on RxAda

[Dmitry A. Kazakov]

On Wed, 21 Oct 2015 05:12:40 -0700 (PDT), Hadrien Grasland wrote:

> Reading through my post again, I should 1/Apologize for the missing words
> here and there and 2/Point out that my implementation of Observable is
> mostly synchronous, because all the costly mapping work occurs within the
> accept rendezvous.
> 
> To make an asynchronous Observer would require quite a couple more tweaks,
> allowing the Observer task to save its entry parameters and use them once
> the computation is over. In particular, the output channel should be
> passed by access, rather than by in out value.
> 
> This does not, however, void my core point that Ada's accessibility rules
> would end up getting in our way if we wanted to implement something like
> ReactiveX for Ada.

Not going to pretend understanding what are you talking about, but
synchronous vs. asynchronous in the context of publisher / subscriber is
determined by whether the publisher is blocked by subscriber(s) or not.

From that point of view using tasks or other synchronization mechanism in
the non-blocking (and thus asynchronous, and thus marshaling) scenario is
straight out wrong. See? If you block, it is not asynchronous.

A proper design pattern for non-blocking (and thus non-delivery guaranteed)
1-n publisher / subscriber scenario is a blackboard or sets of queues on
the subscriber's side. Neither require tasks, nor access types (no in the
form that something gets allocated in a conventional memory pool).

Yes, at some point you will need to defeat the type system. That is a job
for stream attributes, "for address use", or a fake memory pool. Either
would work.

-- 
Regards,
Dmitry A. Kazakov
http://www.dmitry-kazakov.de

Re: Musings on RxAda

[Hadrien Grasland]

Yes, thinking about it some more, this implementation of Observer is REALLY wrong. It actually is blocking in two different places where it shouldn't block : before the accept, and within the accept.

I do believe, however, that a task which blocks or performs CPU work after the rendezvous is acceptable, and could be used to achieve some parallelism in the context of a true data pipeline (i.e. more than 1 input).

Re: Musings on RxAda

[Dmitry A. Kazakov]

On Wed, 21 Oct 2015 09:18:12 -0700 (PDT), Hadrien Grasland wrote:

> I do believe, however, that a task which blocks or performs CPU work after
> the rendezvous is acceptable,

But it is still blocking. For a publisher/subscriber service no mediator
task is needed. Such a task would not solve any of the problems of such
service anyway.

> and could be used to achieve some
> parallelism in the context of a true data pipeline (i.e. more than 1
> input).

Pipeline is just a method of buffered exchange. As such it is blocking
(publisher) when the subscriber is unable to process all published data. 

Moreover it is unsuitable for 1-n communications. If you are OK with
blocking the publisher, then no buffering is ever needed. Buffering
(marshaling) is required by non-blocking but not ensures it.

And no mediator task is needed anyway, not even a protected object. 1-1
FIFO requires none of Ada's synchronization primitives.

-- 
Regards,
Dmitry A. Kazakov
http://www.dmitry-kazakov.de

Re: Musings on RxAda

[Hadrien Grasland]

So you would essentially store a pipeline of data and operations in some container, then perform all of them on the same thread when the output of the pipeline is requested ?

This seems much more complex to implement to me, since your operation queue needs to be able to store data of any type and operation function pointers, all in a type-safe way. As far as I can tell, you cannot use streams for that because a stream requires you to know what you are reading from it.

Re: Musings on RxAda

[Dmitry A. Kazakov]

On Wed, 21 Oct 2015 12:09:58 -0700 (PDT), Hadrien Grasland wrote:

> So you would essentially store a pipeline of data and operations in some
> container, then perform all of them on the same thread when the output of
> the pipeline is requested ?

I don't know your requirements. In a FIFO scenario, yes, the publisher
pushes the object. The subscriber pulls it out. In a blackboard scenario
(non-blocking) the publisher pushes the object and the subscribers scan the
blackboard for updates.

Of course in a modern typed language like Ada you would not mess with "data
and operations." There are objects for that. The type of the object
determines the operations.

> This seems much more complex to implement to me, since your operation
> queue needs to be able to store data of any type and operation function
> pointers, all in a type-safe way. As far as I can tell, you cannot use
> streams for that because a stream requires you to know what you are
> reading from it.

That is because you are trying to think about it in terms of C. Ada is not
C, luckily. It is quite straightforward to marshal T'Class objects with a
dispatching operation Do_It. And yes, it is as much easy to marshal handles
to reference-counted objects if objects are expected large (and you have
shared memory).

-- 
Regards,
Dmitry A. Kazakov
http://www.dmitry-kazakov.de

Re: Musings on RxAda

[Hadrien Grasland]

Le mercredi 21 octobre 2015 21:35:26 UTC+2, Dmitry A. Kazakov a écrit :
> On Wed, 21 Oct 2015 12:09:58 -0700 (PDT), Hadrien Grasland wrote:
> 
> > So you would essentially store a pipeline of data and operations in some
> > container, then perform all of them on the same thread when the output of
> > the pipeline is requested ?
> 
> I don't know your requirements. In a FIFO scenario, yes, the publisher
> pushes the object. The subscriber pulls it out. In a blackboard scenario
> (non-blocking) the publisher pushes the object and the subscribers scan the
> blackboard for updates.
> 
> Of course in a modern typed language like Ada you would not mess with "data
> and operations." There are objects for that. The type of the object
> determines the operations.
> 
> > This seems much more complex to implement to me, since your operation
> > queue needs to be able to store data of any type and operation function
> > pointers, all in a type-safe way. As far as I can tell, you cannot use
> > streams for that because a stream requires you to know what you are
> > reading from it.
> 
> That is because you are trying to think about it in terms of C. Ada is not
> C, luckily. It is quite straightforward to marshal T'Class objects with a
> dispatching operation Do_It. And yes, it is as much easy to marshal handles
> to reference-counted objects if objects are expected large (and you have
> shared memory).
> 
> -- 
> Regards,
> Dmitry A. Kazakov
> http://www.dmitry-kazakov.de

Ah, I see. I agree that an object oriented solution could indeed solve this serialization problem well !

Re: Musings on RxAda

[Alejandro R. Mosteo]

Hadrien Grasland wrote:

> Le mercredi 21 octobre 2015 21:35:26 UTC+2, Dmitry A. Kazakov a écrit :
>> On Wed, 21 Oct 2015 12:09:58 -0700 (PDT), Hadrien Grasland wrote:
>> 
>> > So you would essentially store a pipeline of data and operations in
>> > some container, then perform all of them on the same thread when the
>> > output of the pipeline is requested ?
>> 
>> I don't know your requirements. In a FIFO scenario, yes, the publisher
>> pushes the object. The subscriber pulls it out. In a blackboard scenario
>> (non-blocking) the publisher pushes the object and the subscribers scan
>> the blackboard for updates.
>> 
>> Of course in a modern typed language like Ada you would not mess with
>> "data and operations." There are objects for that. The type of the object
>> determines the operations.
>> 
>> > This seems much more complex to implement to me, since your operation
>> > queue needs to be able to store data of any type and operation function
>> > pointers, all in a type-safe way. As far as I can tell, you cannot use
>> > streams for that because a stream requires you to know what you are
>> > reading from it.
>> 
>> That is because you are trying to think about it in terms of C. Ada is
>> not C, luckily. It is quite straightforward to marshal T'Class objects
>> with a dispatching operation Do_It. And yes, it is as much easy to
>> marshal handles to reference-counted objects if objects are expected
>> large (and you have shared memory).
>> 
>> --
>> Regards,
>> Dmitry A. Kazakov
>> http://www.dmitry-kazakov.de
> 
> Ah, I see. I agree that an object oriented solution could indeed solve
> this serialization problem well !

Thank you for your musings, both of you. I guess we would all need to have a 
firmer grasp of ReactiveX semantics for more accurate speculation.

Anyway, to not leave it here, I can precise that indeed the observable-
subscriber relationship is 1:n. As I understand it, the decoupling may 
happen at every observable (but it's not mandatory). 

I omitted a piece of information that I see now is fundamental here. When an 
Observable has a new piece of data (or when you override an Operator), it 
will call the onNext(datum) method of its subscribers. So there is a push 
mechanism. Indeed there are also operators to deal with "backpressure" (too 
fast data emission by an observable).

This means that if your operation is fast it needs not to be asynchronous, 
but it could be if desired and there are also operators to force it. For 
example (sorry about the pseudocode, not Java nor Ada):

Observable.fromClickEvents(SomeButton)
    .onWorkerThread()  -- To ensure we are not blocking the GUI thread
    .count(during => one_second) -- This could be any costly operation
    .onGuiThread()               -- Switch back to GUI thread
    .subscribe(count -> SomeTextLabel.setText(count));

This would count clicks on a button and output the count every second in a 
GUI field.

What I see interesting of this methodology is that it basically hides (in 
theory, of course then you read about nightmare histories) the explicit 
definition of tasks, problems with race conditions and data integrity, 
because these aspects are within the implementation of the base Observable, 
Operator and Subscriber objects, and the user sees only a pipeline of data 
being transformed that can jump from thread to thread as needed, and without 
blocking any other threads (as long as synchronization is not needed, I 
guess). This, in languages which such poor high-level concurrency built-in 
mechanisms like C++ or Java, is a blessing.

The power of the setup comes with the huge number of predefined operators 
that exist in the library. It's the closer to "chaining blocks" in something 
like Simulink I've experienced in code. But then, I stress, the syntax of 
Java 8 is mandatory or else you get a nightmare of boilerplate code also, 
which is what worries me about an Ada implementation.

I have a vague notion in my head that, in Ada, by using generics one could 
have all the boilerplate in instantiations and keep the chain definition 
terse (which is the goal) and with type safety. But still the function 
bodies would have to go elsewhere (unlike in the last line of the example 
above).

I see a parallelism on how the new syntax in Ada 2012 for looping over 
containers cuts down on boilerplate enormously, enabling having the actual 
loop body code in the loop. And if you look at any example on how to achieve 
this for your own types, there is indeed quite the boilerplate, but you do 
it once to spare it from your library users.

Re: Musings on RxAda

[Dmitry A. Kazakov]

On Thu, 22 Oct 2015 13:02:46 +0200, Alejandro R.  Mosteo wrote:

> Observable.fromClickEvents(SomeButton)
>     .onWorkerThread()  -- To ensure we are not blocking the GUI thread
>     .count(during => one_second) -- This could be any costly operation
>     .onGuiThread()               -- Switch back to GUI thread
>     .subscribe(count -> SomeTextLabel.setText(count));

This looks like a plain active object. On click you simply call its
primitive operation Count, straight from the event handler. The
implementation of Count calls a task-safe SetText which marshals back to
the GUI messages loop. Basically, 2 source code lines.

There is not much use in this anyway. In a real-life GUI it is far more
complicated. When a detached asynchronous operation starts, you normally
need to be able to indicate its progress, to be able to cancel it, have to
deal with MVC mess etc. As an example see how GPS handles the "build"
button clicks. A crude mechanism like Observable would not scale.

> What I see interesting of this methodology is that it basically hides (in 
> theory, of course then you read about nightmare histories) the explicit 
> definition of tasks,

It does not if you have to specify tasks. Active object does.

> problems with race conditions and data integrity,

I don't see why and how. It is too low level to judge. Especially
integrity, because you seem ignore GUI elements bound to the detached
execution. A typical problem is when SomeTextLabel gets killed while your
thing keeps on counting...
 
> The power of the setup comes with the huge number of predefined operators 
> that exist in the library. It's the closer to "chaining blocks" in something 
> like Simulink I've experienced in code.

It is a poor analogy because Simulink is 100% synchronous. (We are using
that to make real-time out of the simulated one. In a custom Simulink block
we do "delay until", and here you are, all simulation gets blocked.)

-- 
Regards,
Dmitry A. Kazakov
http://www.dmitry-kazakov.de

Re: Musings on RxAda

[Alejandro R. Mosteo]

Dmitry A. Kazakov wrote:

> On Thu, 22 Oct 2015 13:02:46 +0200, Alejandro R.  Mosteo wrote:
> 
>> Observable.fromClickEvents(SomeButton)
>>     .onWorkerThread()  -- To ensure we are not blocking the GUI thread
>>     .count(during => one_second) -- This could be any costly operation
>>     .onGuiThread()               -- Switch back to GUI thread
>>     .subscribe(count -> SomeTextLabel.setText(count));
> 
> This looks like a plain active object. On click you simply call its
> primitive operation Count, straight from the event handler. The
> implementation of Count calls a task-safe SetText which marshals back to
> the GUI messages loop. Basically, 2 source code lines.

It might be so in this particular made-up example. I guess many cases of 
observables and filters in the RX framework are indeed active objects.

> There is not much use in this anyway. In a real-life GUI it is far more
> complicated. When a detached asynchronous operation starts, you normally
> need to be able to indicate its progress, to be able to cancel it, have to
> deal with MVC mess etc. As an example see how GPS handles the "build"
> button clicks. A crude mechanism like Observable would not scale.

The GUI was an example, the methodology has no specific purpose. For me, the 
interest is in how such a thing (the RX framework) could be implemented in 
Ada resulting in similar effort and clarity for the user as in other 
languages.

Here's a real use example if you're interested:
http://techblog.netflix.com/2013/02/rxjava-netflix-api.html

It would take someone more proficient on the framework than me to discuss 
the MVC/scalability issue.

>> What I see interesting of this methodology is that it basically hides (in
>> theory, of course then you read about nightmare histories) the explicit
>> definition of tasks,
> 
> It does not if you have to specify tasks. Active object does.
> 
>> problems with race conditions and data integrity,
> 
> I don't see why and how. It is too low level to judge. Especially
> integrity, because you seem ignore GUI elements bound to the detached
> execution. A typical problem is when SomeTextLabel gets killed while your
> thing keeps on counting...

I've left out everything about unsubscribing and error management in my 
effort to keep things simple.

>> The power of the setup comes with the huge number of predefined operators
>> that exist in the library. It's the closer to "chaining blocks" in
>> something like Simulink I've experienced in code.
> 
> It is a poor analogy because Simulink is 100% synchronous. (We are using
> that to make real-time out of the simulated one. In a custom Simulink
> block we do "delay until", and here you are, all simulation gets blocked.)

I was referring to the user experience connecting blocks, not about the 
implementation of the blocks that Simulink uses. Sorry if I was unclear. The 
Lego Mindstorms graphical blocks language would be a similar experience. 
Perhaps Scratch too.

Re: Musings on RxAda

[Jacob Sparre Andersen]

Alejandro R. Mosteo wrote:

> Anyway, the gist of it is that you chain function calls (operators in
> their jargon) to process data asynchronously.

This sounds like how you chain filters in a Unix shell using pipes.

Continuing that thought, an obvious mapping to Ada is tasks (acting as
filters) and protected objects (acting as pipes/buffers).

Even though Ada doesn't explicitly have lambda functions, localised
scopes allow similar (if somewhat more verbose) formulations.

> Observable.just("Hello, world!")
>     .map(s -> s + " -Dan")
>     .map(s -> s.hashCode())
>     .map(i -> Integer.toString(i))
>     .subscribe(s -> System.out.println(s));

   Q1 : String_Queues.Queue;

   task Hello_World;
   task body Hello_World is
   begin
      Q1.Enqueue (+"Hello, world!");
   end Hello_World;

   Q2 : String_Queues.Queue;

   task Add_Source;
   task body Add_Source is
      Buffer : Unbounded_String;
   begin
      Q1.Dequeue (Buffer);
      Q2.Enqueue (Buffer & " -Dan");
   end Add_Source;

   Q3 : Hash_Queues.Queue;

   task Hash_String;
   task body Hash_String is
      Buffer : Unbounded_String;
   begin
      Q2.Dequeue (Buffer);
      Q3.Enqueue (Hash (Buffer));
   end Hash_String;

   Q4 : String_Queues.Queue;

   task Hash_As_String;
   task body Hash_As_String is
      Buffer : Hash_Type;
   begin
      Q3.Dequeue (Buffer);
      Q4.Enqueue (+Hash_Type'Image (Buffer));
   end Hash_As_String;

   task Output;
   task body Output is
      Buffer : Unbounded_String;
   begin
      Q4.Dequeue (Buffer);
      Put_Line (+Buffer);
   end Output;

See <https://bitbucket.org/sparre/ada-2012-examples>
("src/chained_calls.adb") for the full, compilable version.

I'm tempted to make the tasks task types with references to the input
and output queues as discriminants, but I'm not sure it really would be
an improvement.

Greetings,

Jacob
-- 
"If it's a mess, hide it..." -- J-P. Rosen