Stream handling function does not obtain the actual state (it is always the initial one instead)

[ArtemyB]

While building an app via this library stumbled upon a problem that I can't completely understand. Should the stream handling function be designed assuming the input model/state parameter is the actual/current state or the initial one? As I understand this code, it should get the updated state. But in my app the stream's state parameter always retains the initial value.

Example

The example is simplistic and meaningless -- it is just for demo purpose.

Brief explanation

The state contains SavedInput : string option, that is taken from HTML Input element after clicking "Save" button.

The messages are SaveNewInput of string , UpperCase and LowerCase. The 1st one is simply propagated from the stream, the last two are ignored if SavedInput is None:

messages
|> AsyncRx.choose (function
    | SaveNewInput _ as msg -> Some msg
    | UpperCase
    | LowerCase as msg ->
        state.SavedInput
        |> Option.map (fun _ -> msg)

The intent is to ignore the messages UpperCase and LowerCase until user saves an entered string. So if the stream function obtained the actual state, the example would work properly. However in reality it always ignores UpperCase and LowerCase messages, because the stream function never gets an updated state -- it always has the initial value.

So this is the problem I want to clarify.

[ArtemyB]

I suspect the implied design is, after all, to propagate an updated state to a stream. In that case I have a guess why it does not work actually. The problem is in the useEffect hook. Inside it uses msgs value, which ideally should be a new Observable, produced after applying the stream function to the updated state. However the useEffect dependencies array contains only the tag value. And it means that all its "external" values, except the tag, remains the same as they were on the first run of the useStatefulStream hook -- including the msgs value. Thus, the msgs Observable being used in the useEffect hook is actually the one, that was produced initially with the initial state value (i.e. the state-parameter value of the useStatefulStream hook).

The first solution that comes to mind could be to pass msgs to the useEffect dependencies, but doing so results in stream re-subscribing on every state update and, I suppose, the messages sent during the re-subscription are being lost. In my concrete case the app gets stuck on the "in progress" state, because after stream re-creation the response ("complete progress") message does not reach the stream (and consequently the Update function too).

So the goal now is to find a way to propagate the updated state properly, not causing such re-subscriptions with messages loss.

[dbrattli]

@ArtemyB Thanks for opening this issue. I need to go out for a few hours, but hope to answer you later today

[ArtemyB]

@dbrattli ok, no problem. It seems like I've finally understood what's going on. I'll continue to experiment in order to find a solution. However my mind is still not completely immersed into the Rx-world, so I'm struggling a bit on the tasks that require a more deep understanding (at the same time it feels very comfortable on the high level, where it's all about just combining/composing ready-to-use functions).

[ArtemyB]

Actually putting msgs' (messages stream, handled by stream processing function) into the useEffect hook dependencies fixes the issue in the provided simple example. However this change involves stream re-subscription on every state change (because of msgs' depending on the state in the useMemo hook above). And now, when there is a stream re-subscription on every state change, it is not clear what the tag value is for.

[ArtemyB]

At the same time my app, that has more complex structure (it also includes WebSocket-connection) still does not work properly with the fix. Don't know, could be I misunderstood the concept of the Fable.Reaction architecture. If needed, I could try to include a corresponding repro into the example above.