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 someone 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.