Stream pagination and subscribe/unsubscribe to individual items

Hi,

I am paginating through a bunch of classes, displaying 10 at a time. Every time I change a page, I need to subscribe to the new classes count, and unsubscribe from those shown previously. (I want the count to live-update in the table, so I will be re-inserting the classes to the stream when update is broadcast)

The only thing I was able to make work is

unsubscribe (kinda workaround)>

Sorted.PubSub
    |> Registry.keys(self())
    |> Enum.map(fn topic ->
      Phoenix.PubSub.unsubscribe(Sorted.PubSub, topic)
    end)

and subscribe>

class_ids =
        socket.assigns.streams.classes.inserts
        # Convert stream to list eagerly
        |> Enum.to_list()
        # Extract IDs
        |> Enum.map(fn {_dom_id, _, class, _} -> class.id end)

      class_ids |> Enum.each(&Sorted.Pubsub.subscribe_class_count/1)

It works, but it kinda smells to me…

I was hoping I could do something like this for both unsubscribe and subscribe

for {_dom_id, class} <- socket.assigns.streams.classes do
        Sorted.Pubsub.subscribe_class(class.id)
      end

but this is throwing

** (ArgumentError) streams can only be consumed directly by a for comprehension.
If you are attempting to consume the stream ahead of time, such as with
`Enum.with_index(@streams.classes)`, you need to place the relevant information
within the stream items instead.

On top of that, even with the above subscribe code, I was not able to unsubscribe before changing the page, as the list was always empty..

So what’s the proper way to go about this? Thank you very much.