Best pattern for derived signals from Resources inside For

[lukexor]

I feel like I've circled the drain on this for too long and none of the options I've tried result in the desired update behavior.

Overview of what I want to do:

• I have two Resources
  • one for status on a list of items, this is fetched during SSR, and after hydration it gets updated via a WS periodically. In the real application this is usually a sub-set of all items
  • The other is the full list of items, some of which may not have any status
• I want to render a table of the item statuses by zipping together the statuses with extra data from the items resource using
• Rows in the table can be expanded, revealing more details
• Updates to the status should not cause the to re-render, and forcing the expanded state to reset.

I've managed to reduce it down to a minimal example, but it's still fairly complex. I'm looking for the best option that fixes some core usability issues with signal updates. I have some flexibility in terms of re-architecting how data is entering the app, but fundamentally I'm still locked into this basic pattern of two overlapping resources, one being updated with a websocket.

The issue with the following example code:

• Toggling state, adding or removing an item - causes For to re-ender and Expanded to reset
• Adding an item doesn't re-render until something else causes it to - e.g. toggling state

Some items related to this:

• Combining reactive_stores with resources and hydration #3679
• <For> inside of <Suspense> or <ErrorBoundary> #1664

I'm sure Suspend here isn't the right way to do this, but I haven't had any better luck using Memo or the like.

/// Renders the home page of your application.
#[component]
fn HomePage() -> impl IntoView {
    let toggle_state = ServerMultiAction::<ToggleState>::new();
    let add_item = ServerMultiAction::<AddItem>::new();
    let remove_item = ServerMultiAction::<RemoveItem>::new();

    let statuses = Resource::new(
        move || toggle_state.version().get(),
        |_| async { RwSignal::new(get_statuses().await.unwrap_or_default()) },
    );
    Effect::new(move |_| {
        if let Some(statuses) = statuses.get() {
            let (_, rx) = mpsc::channel(1);
            spawn_local(async move {
                use futures::StreamExt;
                if let Ok(mut statuses_stream) = status_ws(rx.into()).await {
                    while let Some(Ok(new_statuses)) = statuses_stream.next().await {
                        statuses.set(new_statuses);
                    }
                }
            })
        }
    });

    let items = Resource::new(
        move || (add_item.version().get(), remove_item.version().get()),
        |_| async { get_items().await.unwrap_or_default() },
    );

    #[component]
    fn ItemRow(
        #[prop(into)] status: Signal<Status>,
        #[prop(into)] item: Signal<Item>,
        toggle_state: ServerMultiAction<ToggleState>,
        remove_item: ServerMultiAction<RemoveItem>,
    ) -> impl IntoView {
        let id = memo!(item.id);
        let name = memo!(item.name);
        let expanded = RwSignal::new(false);
        view! {
            <tr>
                <td>{id}</td>
                <td>{name}</td>
                <td>{move || status.get().state.to_string()}</td>
                <td>
                    <button on:click=move |_| {
                        toggle_state.dispatch(id.get().into())
                    }>"Toggle"</button>
                </td>
                <td>
                    <button on:click=move |_| {
                        remove_item.dispatch(id.get().into())
                    }>"Remove"</button>
                </td>
                <td>
                    <button on:click=move |_| {
                        expanded.update(|e| *e = !*e);
                    }>"Toggle Expanded"</button>
                </td>
                <Show when=move || expanded.get()>"Expanded"</Show>
            </tr>
        }
    }


    let item_rows = move || {
        Suspend::new(async move {
            let statuses = statuses.await;
            let items = items.await;
            let item_statuses = statuses
                .get()
                .into_values()
                .filter_map(|status| {
                    items
                        .get(&status.item_id)
                        .cloned()
                        .map(|item| (status, item))
                })
                .collect::<Vec<_>>();

            view! {
                <For
                    each=move || item_statuses.clone()
                    key=move |(status, _)| status.item_id
                    let:((status, item))
                >
                    <ItemRow status item toggle_state remove_item />
                </For>
            }
        })
    };

    view! {
        <Transition>
            <div>
                <table>
                    <thead>
                        <th>"ID"</th>
                        <th>"Name"</th>
                        <th>"State"</th>
                    </thead>
                    <tbody>{item_rows}</tbody>
                </table>
            </div>
            <button on:click=move |_| add_item.dispatch(Item::new().into())>"Add Item"</button>
        </Transition>
    }
}

#[cfg(feature = "ssr")]
static STATUSES: LazyLock<Mutex<BTreeMap<usize, Status>>> = LazyLock::new(|| {
    Mutex::new(BTreeMap::from([
        (
            0,
            Status {
                item_id: 0,
                state: State::Active,
            },
        ),
        (
            1,
            Status {
                item_id: 1,
                state: State::Active,
            },
        ),
    ]))
});

#[cfg(feature = "ssr")]
static ITEMS: LazyLock<Mutex<BTreeMap<usize, Item>>> = LazyLock::new(|| {
    Mutex::new(BTreeMap::from([
        (
            0,
            Item {
                id: 0,
                name: "Item 0".into(),
            },
        ),
        (
            1,
            Item {
                id: 1,
                name: "Item 1".into(),
            },
        ),
    ]))
});

#[leptos::server]
async fn get_statuses() -> Result<BTreeMap<usize, Status>, ServerFnError> {
    Ok(STATUSES.lock().unwrap().clone())
}

#[leptos::server]
async fn get_items() -> Result<BTreeMap<usize, Item>, ServerFnError> {
    Ok(ITEMS.lock().unwrap().clone())
}

#[leptos::server]
async fn toggle_state(item_id: usize) -> Result<(), ServerFnError> {
    let mut statuses = STATUSES.lock().unwrap();
    match statuses
        .values_mut()
        .find(|status| status.item_id == item_id)
    {
        Some(status) => {
            status.state = match status.state {
                State::Active => State::Inactive,
                State::Inactive => State::Active,
            };
            Ok(())
        }
        None => Err(ServerFnError::new("invalid `item_id`")),
    }
}

#[leptos::server]
async fn add_item(item: Item) -> Result<(), ServerFnError> {
    let mut items = ITEMS.lock().unwrap();
    let mut statuses = STATUSES.lock().unwrap();
    statuses.insert(
        item.id,
        Status {
            item_id: item.id,
            state: State::default(),
        },
    );
    items.insert(item.id, item);
    Ok(())
}

#[leptos::server]
async fn remove_item(item_id: usize) -> Result<(), ServerFnError> {
    let mut items = ITEMS.lock().unwrap();
    let mut statuses = STATUSES.lock().unwrap();
    items.remove(&item_id);
    statuses.remove(&item_id);
    Ok(())
}

#[server(protocol = Websocket<JsonEncoding, JsonEncoding>)]
async fn status_ws(
    _input: BoxedStream<(), ServerFnError>,
) -> Result<BoxedStream<BTreeMap<usize, Status>, ServerFnError>, ServerFnError> {
    use futures::SinkExt;

    let (mut tx, rx) = mpsc::channel(1);

    tokio::spawn(async move {
        loop {
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
            let statuses = {
                let mut statuses = STATUSES.lock().unwrap();
                if let Some(status) = statuses.values_mut().next() {
                    status.state = match status.state {
                        State::Active => State::Inactive,
                        State::Inactive => State::Active,
                    };
                }
                statuses.clone()
            };
            tx.send(Ok(statuses)).await.unwrap();
        }
    });

    Ok(rx.into())
}

[lukexor]

Ideally what I'd like is an idiomatic way to fetch one or more resources for some collections, generate a Store from them, and be able to have fine-grain reactivity when they get refetched (or is patched from a websocket message)

[gbj]

Ideally what I'd like is an idiomatic way to fetch one or more resources for some collections, generate a Store from them, and be able to have fine-grain reactivity when they get refetched

I would also like this.

It's not something I've had time to work on recently, and I'm not sure when in the future I'll have time to work on it; unfortunately I do all this in my free time, which ebbs and flows.

It raises some interesting API design questions (do we do things like my_store.field().await? or is it my_store.await.field()? same with the Option<_>ality)

[lukexor]

Ideally what I'd like is an idiomatic way to fetch one or more resources for some collections, generate a Store from them, and be able to have fine-grain reactivity when they get refetched

I would also like this.

It's not something I've had time to work on recently, and I'm not sure when in the future I'll have time to work on it; unfortunately I do all this in my free time, which ebbs and flows.

It raises some interesting API design questions (do we do things like my_store.field().await? or is it my_store.await.field()? same with the Option<_>ality)

Totally understand and really appreciate all the features so far.

Are there custom options for a work around you can think of? E.g. some custom primitive. I'm just not sure where to start or how to still leverage the SSR/serialize features of Resources while still constructing a client side reactive graph that can be patched on refetch

At the moment I feel a little blocked and worried I may not be able to continue migrating to leptos until I can find a pattern that's ergonomic enough (even if I have to write something custom) for my other team members to use.

Certainly happy to bring back any work in that area to help out with a PR - but I've seen some of the leptos internals and it's fairly complex in places!

[gbj]

(To be clear, I am in the middle of writing a response to the main post!)

[gbj]

So, looking at the example here:

<For
    each=move || item_statuses.clone()

This is often an indicator that <For/> isn't quite set up right here, in one way or another. Specifically, the value returned by each here will never change. It does not track any reactive value, and it will never cause <For/> to re-run.

Completely understood that this is because you're using Suspend to read the value asynchronously. If you want to use <For/>, though, you need to read from it synchronously in each, so that the <For/> is triggered to run again when it changes, rather than being wiped out and re-rendered.

Here's what I mean by "read from it synchronously":

    let item_rows = move || match (statuses.get(), items.get()) {
        (Some(statuses), Some(items)) => statuses
            .into_values()
            .filter_map(|status| {
                items
                    .get(&status.item_id)
                    .cloned()
                    .map(|item| (status, item))
            })
            .collect::<Vec<_>>(),
        _ => vec![],
    };
    
    // ... 
    
     <For
           each=item_rows

Rather than wiping out the whole Suspend block every time one of the two resources changes, this derives a new signal from their values and passes it into the each directly. (This is how all resource reads worked prior to the async support in 0.7, and this is still very useful for control-flow components like For! There is a way to do this using Suspend by directly using the keyed function in the tachys rendering library, but you might want to keep using For, which requires a synchronous each like this.)

Now, the rows are keyed only by status.item_id, and not by status. This is exactly what you want, because you don't want to lose that Expanded state if the status changes. However, it means you need to derive some signal or memo to look up the status for each row

                       <ItemRow
                            status=Memo::new(move |_| {
                                statuses
                                    .read()
                                    .as_ref()
                                    .and_then(|statuses| {
                                        statuses.get(&status.item_id).cloned()
                                    })
                                    .unwrap_or_default()
                            })

Granted this is a bit of annoying boilerplate, it's just one of the downsides of fine-grained reactivity combined with keyed list iteration.

Note also that I removed the RwSignal being returned from the resource. Returning a signal from a resource is (almost?) never what you want. Resources can be mutated directly, so this is fine and avoids the possibility of "signal diposed" issues happening (as the RwSignal from each previous resource run should be disposed every time the resource gets a new value, otherwise it's a memory leak.)

    let statuses = Resource::new(
        move || toggle_state.version().get(),
        |_| async { get_statuses().await.unwrap_or_default() },
    );

Here's a complete implementation including imports, for my own sake if I need to look back at it later. Apologies for any infidelities to your types, here!

use futures::channel::mpsc;
use leptos::{
    prelude::*,
    server_fn::{codec::JsonEncoding, BoxedStream, Websocket},
    task::spawn_local,
};
use serde::{Deserialize, Serialize};
use std::{
    collections::BTreeMap,
    fmt::Display,
    sync::{LazyLock, Mutex},
};

#[derive(Debug, Default, Clone, Serialize, Deserialize)]
struct Item {
    id: usize,
    name: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
struct Status {
    item_id: usize,
    state: State,
}

#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)]
enum State {
    #[default]
    Inactive,
    Active,
}

impl Display for State {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Active => write!(f, "Active"),
            Self::Inactive => write!(f, "Inactive"),
        }
    }
}

pub fn shell(options: LeptosOptions) -> impl IntoView {
    view! {
        <!DOCTYPE html>
        <html lang="en">
            <head>
                <meta charset="utf-8"/>
                <meta name="viewport" content="width=device-width, initial-scale=1"/>
                <AutoReload options=options.clone()/>
                <HydrationScripts options/>
            </head>
            <body>
                <App/>
            </body>
        </html>
    }
}
/// Renders the home page of your application.
#[component]
pub fn App() -> impl IntoView {
    let toggle_state = ServerMultiAction::<ToggleState>::new();
    let add_item = ServerMultiAction::<AddItem>::new();
    let remove_item = ServerMultiAction::<RemoveItem>::new();

    let statuses = Resource::new(
        move || toggle_state.version().get(),
        |_| async { get_statuses().await.unwrap_or_default() },
    );

    Effect::new(move |_| {
        let (_, rx) = mpsc::channel(1);
        spawn_local(async move {
            use futures::StreamExt;
            if let Ok(mut statuses_stream) = status_ws(rx.into()).await {
                while let Some(Ok(new_statuses)) = statuses_stream.next().await
                {
                    statuses.set(Some(new_statuses));
                }
            }
        })
    });

    let items = Resource::new(
        move || (add_item.version().get(), remove_item.version().get()),
        |_| async { get_items().await.unwrap_or_default() },
    );

    #[component]
    fn ItemRow(
        #[prop(into)] status: Signal<Status>,
        #[prop(into)] item: Signal<Item>,
        toggle_state: ServerMultiAction<ToggleState>,
        remove_item: ServerMultiAction<RemoveItem>,
    ) -> impl IntoView {
        let id = memo!(item.id);
        let name = memo!(item.name);
        let expanded = RwSignal::new(false);
        view! {
            <tr>
                <td>{id}</td>
                <td>{name}</td>
                <td>{move || status.get().state.to_string()}</td>
                <td>
                    <button on:click=move |_| {
                        toggle_state.dispatch(id.get().into())
                    }>"Toggle"</button>
                </td>
                <td>
                    <button on:click=move |_| {
                        remove_item.dispatch(id.get().into())
                    }>"Remove"</button>
                </td>
                <td>
                    <button on:click=move |_| {
                        expanded.update(|e| *e = !*e);
                    }>"Toggle Expanded"</button>
                </td>
                <Show when=move || expanded.get()>"Expanded"</Show>
            </tr>
        }
    }

    let item_rows = move || match (statuses.get(), items.get()) {
        (Some(statuses), Some(items)) => statuses
            .into_values()
            .filter_map(|status| {
                items
                    .get(&status.item_id)
                    .cloned()
                    .map(|item| (status, item))
            })
            .collect::<Vec<_>>(),
        _ => vec![],
    };

    view! {
        <Transition>
            <div>
                <table>
                    <thead>
                        <th>"ID"</th>
                        <th>"Name"</th>
                        <th>"State"</th>
                    </thead>
                    <tbody>
                    <For
                        each=item_rows
                        key=move |(status, _)| status.item_id
                        let:((status, item))
                    >
                        <ItemRow
                            status=Memo::new(move |_| {
                                statuses
                                    .read()
                                    .as_ref()
                                    .and_then(|statuses| {
                                        statuses.get(&status.item_id).cloned()
                                    })
                                    .unwrap_or_default()
                            })
                            item
                            toggle_state
                            remove_item
                        />
                    </For>
                    </tbody>
                </table>
            </div>
            <button on:click=move |_| add_item.dispatch(Item::default().into())>"Add Item"</button>
        </Transition>
    }
}

#[cfg(feature = "ssr")]
static STATUSES: LazyLock<Mutex<BTreeMap<usize, Status>>> =
    LazyLock::new(|| {
        use std::sync::Mutex;

        Mutex::new(BTreeMap::from([
            (
                0,
                Status {
                    item_id: 0,
                    state: State::Active,
                },
            ),
            (
                1,
                Status {
                    item_id: 1,
                    state: State::Active,
                },
            ),
        ]))
    });

#[cfg(feature = "ssr")]
static ITEMS: LazyLock<Mutex<BTreeMap<usize, Item>>> = LazyLock::new(|| {
    Mutex::new(BTreeMap::from([
        (
            0,
            Item {
                id: 0,
                name: "Item 0".into(),
            },
        ),
        (
            1,
            Item {
                id: 1,
                name: "Item 1".into(),
            },
        ),
    ]))
});

#[leptos::server]
async fn get_statuses() -> Result<BTreeMap<usize, Status>, ServerFnError> {
    Ok(STATUSES.lock().unwrap().clone())
}

#[leptos::server]
async fn get_items() -> Result<BTreeMap<usize, Item>, ServerFnError> {
    Ok(ITEMS.lock().unwrap().clone())
}

#[leptos::server]
async fn toggle_state(item_id: usize) -> Result<(), ServerFnError> {
    let mut statuses = STATUSES.lock().unwrap();
    match statuses
        .values_mut()
        .find(|status| status.item_id == item_id)
    {
        Some(status) => {
            status.state = match status.state {
                State::Active => State::Inactive,
                State::Inactive => State::Active,
            };
            Ok(())
        }
        None => Err(ServerFnError::new("invalid `item_id`")),
    }
}

#[leptos::server]
async fn add_item(item: Item) -> Result<(), ServerFnError> {
    let mut items = ITEMS.lock().unwrap();
    let mut statuses = STATUSES.lock().unwrap();
    statuses.insert(
        item.id,
        Status {
            item_id: item.id,
            state: State::default(),
        },
    );
    items.insert(item.id, item);
    Ok(())
}

#[leptos::server]
async fn remove_item(item_id: usize) -> Result<(), ServerFnError> {
    let mut items = ITEMS.lock().unwrap();
    let mut statuses = STATUSES.lock().unwrap();
    items.remove(&item_id);
    statuses.remove(&item_id);
    Ok(())
}

#[server(protocol = Websocket<JsonEncoding, JsonEncoding>)]
async fn status_ws(
    _input: BoxedStream<(), ServerFnError>,
) -> Result<BoxedStream<BTreeMap<usize, Status>, ServerFnError>, ServerFnError>
{
    use futures::SinkExt;

    let (mut tx, rx) = mpsc::channel(1);

    tokio::spawn(async move {
        loop {
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
            let statuses = {
                let mut statuses = STATUSES.lock().unwrap();
                if let Some(status) = statuses.values_mut().next() {
                    status.state = match status.state {
                        State::Active => State::Inactive,
                        State::Inactive => State::Active,
                    };
                }
                statuses.clone()
            };
            tx.send(Ok(statuses)).await.unwrap();
        }
    });

    Ok(rx.into())
}

[lukexor]

So I tried out this version and the state updates work exactly as desired! Remove seems to work fine as well

However, Add Item does not with the example as-written.

Simply adding add_item.version().get() and remove_item.version().get() to the statuses Resource resolves it. I think because of the item_rows match statement and the looking up one resource in another, they both have to stay in sync.

So this is fine and should work for my case, I'm going to try integrating it today - it does beg the question, is there any way to prevent item_rows from running twice? I tried using a Memo - but it considers it a warning, reading Resource outside of Suspense. I don't think it's particularly egregious - but if this list were super large or expensive to calculate - it could be a performance concern.

[lukexor]

Thank you for the details! I did figure out the For reactivity around each, but not in time for my original post. I've tried so many iterations at this point it's been hard to figure out which ones I've already tried in certain combinations. I'll give these a go tomorrow.

The reason I have the signal currently from the resource is because 0.8.2 doesn't allow mutating resources, so I'd have to rely on a GitHub patch in Cargo.toml.

Hopefully I can sort things out now, I'll follow up and thanks again!

[lukexor]

Also apparently I was wrong about 0.8.2 not allowing mutating resources? Maybe it was a carry over from 0.8.0 or 0.8.1 and I forgot to update things.

[gbj]

I think that was included in the patch release 0.8.2, that's correct.

[lukexor]

Things have been going swimmingly since I made these changes. Thanks so much!