How can I use reactive programming at multiple aggregation levels of data?

Question

When I have a reactive stream of some composite data object, I'm never able to reap the benefits of reactive programming below the highest level of the composite. Here's an example to illustrate the issue:

Say we are building an edit form for data of type Product:

Product: {
    name: string
    description: string
    comments: Array text
}

We also have a FancyControl for user input, which could contain arbitrarily complex features, e.g. autocomplete dropdowns and spell checking. Those features also rely on reactive programming ("everything is a stream"), so we have something like this:

FancyControl: (inText: Stream string, ...other input streams) => {
    asHtml: Stream HTML
    outText: Stream string
    ...other output streams
}

Note that output streams may be wired back into some input streams. For instance, outText is likely to be joined with other streams and wired back into inText.

Given products: Stream Product (a stream emitting single Products), we want a Stream HTML that visually renders the product to HTML, using our FancyControls as applicable. (Please refrain from objecting to this example goal; the reality is more performant and secure.)

If we have a function combine: (Array Stream T1, Array T1 -> T2) -> Stream T2, we can set up a stream to render name and description before we actually have a Product in our hand:

nameControl = FancyControl(products.map(p => p.name), ...);
descriptionControl = FancyControl(products.map(p => p.description), ...);

htmlStream = combine([nameControl.asHtml, descriptionControl.asHtml], 
    (nameHtml, descHtml) => 'tr' + ..., 
);

But I cannot think of any good way to set up the streams to render comments.

Attempt 1: We can try the same trick if we know (or have a good guess) the maximum number of comments any product will have:

range(0, maxComments).map(i =>
    FancyControl(products.map(p => p.comments[i]))
)

But that's unrealistic to always know, we have to guard against null and out-of-bounds indexes, and most of our streams are sitting idle most of the time. We also get bad combinatorial explosions if our arrays have arrays of arrays, and it doesn't work for other structures (e.g. dictionaries with unknown keys).

Attempt 2: Lift and flatten

If we have a function lift: value -> Stream value we can do:

commentControls = products.map(p =>
    p.comments.map(text => FancyControl(lift(text), ...))
)

We get a Stream Array FancyControl, and if we have function (Stream Stream T).flatten: () -> Stream T we can do:

commentControls
    .map(ctrlArr => combine(
        ctrlArr.map(ctrl => ctrl.asHtml),  // Array Stream HTML
        htmlArray => htmlArray.map(html => 'tr' ...)
    ))  // Stream Stream Html
    .flatten()

Which gives us the desired Stream HTML. This avoids all the disadvantages of attempt 1, but we are creating many single-use streams every time any event happens.

Streams should emit immutable values, but the user is modifying the product on this page. Therefore every modification causes products to emit a new value. FancyControl may be complex to wire up, and we are now doing it an arbitrary number of times inside of products.map. Most of the streams we create only ever emit one value and then complete. Below the highest level of the aggregate, reactive programming is (poorly) emulating synchronous value passing.

In reality, the data structures we're dealing with are much more complicated than a single Product. What if we are editing multiple products, with the ability to add and delete? We now have a Stream Array Product that, because it must emit immutable values, is creating and emitting a whole new array every time the user types a key. This is necessary; we can't short-circuit this without limiting our functionality: what if the autocomplete options offered for the product on line 17 depends on something the user entered earlier on line 5?

I am finding myself writing complicated stream-manager type classes which do diffs on aggregates to try to figure out when to open and close various streams, but this is hacky and feels like way overkill for just the relatively simple things I'm trying to do.

So my very general question is: How can I keep the nice simplicity of reactive programming below the highest level of an aggregate data structure?

Ideally, the solution would be recursively applicable. That is, if I have a nice reactive way to solve Stream BigComposite SmallComposite -> Stream Result, then I can use reactive concepts to solve Stream HugeComposite BigComposite SmallComposite -> Stream Result.

Clarifications

I am talking about Reactive Programming "in the small". This is a programming paradigm closely related to Functional Programming: the world where concepts like Stream.of(3) (creates a stream that emits the single value 3, then completes) and Stream.never() (creates a stream that never emits a value) are useful concepts, similar to how Sequence.empty() is useful. I am not talking about what could be called "Reactive Systems Architecture", where we are worried about sharded databases and elastic server allocations, where it makes sense for each actor to keep its own copy of all the data it needs to do its job, and where we wouldn't expect the concept to recursively apply to smaller and smaller problems.

This question is probing the limits of the Reactive Programming paradigm. Mature programming paradigms don't have artificial floors and ceilings below and above which they stop providing their benefits. If we heard "Object-Oriented programming no longer works as a concept if your inheritance tree is deeper than 2 levels" we would rightly scoff, just as if we heard "Functional programming doesn't work if you're trying to use it inside another function". Concepts from those paradigms apply recursively: I have the full power of functional programming even inside functions inside functions.

Yet, in reactive programming I'm finding that I'm losing the benefit of the entire paradigm when some seemingly simple constraints apply:

1. I don't know how many child streams to wire up until after I've seen a value from a parent stream.

2. Some child streams can emit signals that will logically cause the parent stream to emit a new value.

If your answer is "this is not what Reactive Programming is for", then I don't see how the paradigm is salvageable if it's not recursively applicable. If I encounter a problem and solve it using streams (this is what FancyControl is supposed to be), I want to be able to use my solution if I encounter that same problem again "inside of" another stream. Saying I can't do that is like saying I can't use functional programming concepts inside another function.

If you're getting hung up on FancyControl, please consider the more concrete example of Dropdown:.

Dropdown: (currentValue: Stream id, possibleValues: Stream Collection (id, model)) => {
    newValue: Stream id,
    htmlNodes: Stream virtualHtmlNode
}

Dropdown is not subscribing to anything, it's just a way to construct a pair of related output streams from a pair of related input streams. When the htmlNodes are eventually rendered to the screen, they provide a way for the user to cause newValue to emit a value. If I need to wire up n dropdowns, where I don't know n until I see a composite pass through some parent stream (condition 1), and some possibleValues streams may depend on this new value and other parts of the composite (condition 2), I appear to hit a floor below which reactive programming concepts no longer provide any benefit.

Answer 1

In both your examples, FancyControl is having values pushed into it, rather than have FancyControl reacting to values that it has requested.

Most discussions of reactive approaches to wiring things together revolve around ports and pipes. Pipes, in this sense, correspond to reactive streams or observer chains. Ports are the connections where data flows in or out. In this sense, FancyControl should present a set of input and output ports to the wiring manager that knows how to wire things up. The wiring gets done once for FancyControl.

Reactive Pull Model

Reactive systems pull data from data streams. Your examples are dominated by the push model.

Your FancyControl seems to have a lot of responsibilities. Let's try to break it down into 2 core functions: accept search parameters from the user, and display search results. The search parameters have a fixed set of wirings: text input, radio buttons, check boxes, menus, etc, that are pre-defined. So the input controlling here is fixed, pre-wired and pulls data from the user.

The search results has only a fixed input: a stream of search result collections. It has fixed outputs: search status (count, relevance, etc), and product summaries.

Multilevel Aggregation

If you have a search result set, say 50 products, triggered by what the user has typed so far, and the user continues to type, reducing the number of products to 15, then 10 and then 2, the aggregation process needs to provide a stream of edits rather than a stream of results. The edits consist of "add these products to the set", "remove those products from the set", etc. With a combination of debouncing, throttling, staging, etc, this reduces up-cascading, and combinatorial explosion of operations.

Answer 2

Your question has been bumping around in my brain a bit and I want to take another stab at it... I'm going to explain the problem with an analogy which hopefully will work for you.

Instead of a stream of products, let's assume we had an array of products, and a FancyControl built for arrays like this:

FancyControl: (inText: array string, ...other input arrays) => {
    asHtml: array HTML
    outText: array string
    ...other output arrays
}

How would you feel about a comment like this?

[edited for the sake of the analogy] Note that output arrays may be wired back into some input arrays. For instance, outText is likely to be joined with other arrays and wired back into inText.

To me, the above immediately stands out as odd, and so did your original comment. I assume the above statement about arrays also strikes you as odd, even though you thought it was only natural to ask it for streams.

But to go on... We can now make a FancyControl to handle the names in the products thusly:

nameControl = FancyControl(products.map(p => p.name, ...);
descriptionControl = FancyControl(products.map(p => p.description), ...);

htmlArrays = combine([nameControl.asHtml, descriptionControl.asHtml], 
    (nameHtml, descHtml) => 'tr' + ..., 
);

The idea with the above is that there is one FancyControl to render all the names and one to render all the descriptions. But what would be a good way to setup the arrays to handle comments?

We could try the same trick as your attempt 1 if we know the maximum number of comments allowed:

commentFancyControls = range(0, maxComments).map(i =>
    FancyControl(products.map(p => p.comments[i]))
)

With the above, we end up with an array of FancyControls where commentFancyControl[n] displays products.map(p => p.comment[n]). (Although, what does fancyControl[n] display if a particular product doesn't have a comment[n]?)

Or we could use your attempt 2 idea and "lift and flatten" the comments which will give us a 2-D ragged array of FancyControls which as you say:

[edited for the sake of the analogy] ... avoids all the disadvantages of attempt 1, but we are creating many single-[element] arrays for every comment.

Under this analogy, your general question becomes:

How can I keep the nice simplicity of arrays below the highest level of an aggregate data structure?

Then, according to the analogy, the rest of the question would explain how we are "probing the limits of", and "loosing the benefits of" arrays... but are we really, or is it merely a case of misunderstanding how to use arrays properly?

---

Generally, when someone posts a question here, it's because they don't understand some aspect of the technology they are asking about. In this particular case your misunderstanding is with Monads. As the analogy above shows, the question you have isn't about streams per se, rather it is about any monad; it even applies to arrays (which are also monads,) and how to handle elements that are at different levels of monadic depth.

Your FancyControl isn't at the right level of monadic depth and so it doesn't fit properly inside the monad. As a matter of fact, if you were to create a fancy control that handles strings instead of streams of strings, then it would be a simple matter of flatMapping your stream product into a stream array FancyControl.

Hopefully, this analogy will help clear your misunderstanding.

Answer 3

I think you are entirely too concerned about things that don't matter. Keep in mind that in Elm and Redux (both of which are reactive systems) the entire state of the program is kept in a single immutable object that is replaced every time the user performs an action. In Cycle.js, the entire program is nothing more than a single function that accepts streams of user (and network) input and outputs a single stream for the entire UI.

All of these systems are plenty performant. So I really don't think your "fancy control" is going too big.

Regarding a couple of specific issues you brought up:

I don't know how many child streams to wire up until after I've seen a value from a parent stream.

You say that like it's a problem, but it's the very nature of arrays. No function in any paradigm knows how many objects it will have to deal with until after the function is called with a particular array. Just write your code so it can deal with N elements and be done with it.

If you're having trouble doing that, then I suggest you ask some more specific questions so you can get useful answers.

Some child streams can emit signals that will logically cause the parent stream to emit a new value.

Well that's the whole point! Look up information about Cycle.js. Read/watch virtually anything by André Staltz. Here's a good starter talk.. A well written reactive system will push data to the user (through the screen) while the user pushes data into the system (through the keyboard and mouse). From the app, out through the screen and into the user's eyes, which mutates his/her brain. Then data goes out through his/her fingers, into the keyboard and mouse, which mutates the program's data, and around and around again...

Sorry if this answer has come off a bit snippy, but your question reads more like a rant against the paradigm rather than a legitimate question about it.