Streams in Phoenix LiveView

Streams are an exciting new feature in Phoenix. This post walks us through creating our own Slack-like chat interface that features infinite scroll back, editing past messages, deleting messages, and appending new messages to the bottom all using Streams. It’s a slick and efficient solution that avoids storing all that message data in the LiveView.

In this post, we’ll build out a LiveView chatroom app with the help of LiveView’s new streams feature. You can follow along in the open source codebase or skip ahead to play around with the finished product. We’ll see how streams seamlessly integrate into your existing live views to power interactive and efficient UIs. Along the way, we’ll look at how streams work under the hood. After this, you’ll understand how they work at a deep level.

What are LiveView Streams?

LiveView 0.18.16 ships with the new streams functionality for managing large collections of data client-side, without having to store anything in the LiveView socket. For the past few years, a question I would often hear from developers interested in LiveView was: “What about large datasets?” Users who needed to display and manage long lists of data had to store that data on the server or work with phx-update="append" feature. This new functionality allows for efficient management of large datasets in your live views, letting the client store it instead of the server.

The StreamChat App

For this project, we have a basic LiveView application set up with the following domain:

A Room has many messages.
A Message belongs to a room and a sender. A sender is a user.
A User has many messages.

We also have a Chat context that exposes the CRUD functionality for rooms and messages. All of this backs the main LiveView of the application, StreamChatWeb.ChatLive.Root. This LiveView is mapped to the /rooms and /rooms/:id live routes and this is where we’ll be building our stream-backed chatting feature.

Initialize the Stream

In the router.ex file, we have the following route definitions:

live "/rooms", ChatLive.Root, :index
live "/rooms/:id", ChatLive.Root, :show

The ChatLive.Root LiveView implements a handle_params/3 callback that queries for the room and stores it in socket assigns. We’ll add code to fetch the list of messages for the current room and store them in the stream:

def handle_params(%{"id" => id}, _uri, %{assigns: %{live_action: :show}} = socket) do
  {:noreply,
    socket
    |> assign_active_room(id)
    |> assign_active_room_messages()}
end

def assign_active_room(socket, id) do
  assign(socket, :room, Chat.get_room!(id))
end

def assign_active_room_messages(%{assigns: %{room: room}} = socket) do
  stream(socket, :messages, Chat.last_ten_messages_for(room.id))
end

List Messages with Streams

Next, we want to render a list of messages in each chat room. Here’s the UI we’re going for:

We’ll render the contents of that stream in a HEEx template. Let's add a function component, Room.show/1, to render the messages list:

defmodule StreamChatWeb.ChatLive.Room do
  use Phoenix.Component

def show(assigns) do
    ~H"""
    <div id={"room-#{@room.id}"}>
      <Messages.list_messages messages={@messages} />
      <!-- ... form for a new message -->
    </div>
    """
  end
end

This function component calls another function component, Messages.list/1, where we render the messages:

defmodule StreamChatWeb.ChatLive.Messages do
  use Phoenix.Component

def list_messages(assigns) do
    ~H"""
    <div id="messages" phx-update="stream">
      <div :for={{dom_id, message} <- @messages} id={dom_id}>
        <.message_meta message={message} />
        <.message_content message={message} />
      </div>
    </div>
    """
  end
end

Prepend Stream Messages for Infinite Scroll Back

Our app uses a JS hook to send the "load_more" event to the server when the user scrolls up to the top of the chat window. We’ll implement the event handler to fetch the previous batch of messages:

def handle_event("load_more", _params, %{assigns: %{oldest_message_id: id}} = socket) do
  messages = Chat.get_previous_n_messages(id, 5)

{:noreply,
    socket
    |> stream_batch_insert(:messages, messages, at: 0)
    |> assign_oldest_message_id(List.last(messages))}
end

Append a New Message with stream_insert

When the form for a new message is submitted, we’ll need to insert the new message into the stream. We already have support for this through PubSub:

def handle_info(%{event: "new_message", payload: %{message: message}}, socket) do
  {:noreply, insert_new_message(socket, message)}
end

def insert_new_message(socket, message) do
  socket
  |> stream_insert(:messages, Chat.preload_message_sender(message))
end

Delete a Message with stream_delete

When the user clicks the delete button for a message, we handle that event like this:

def handle_event("delete_message", %{"item_id" => message_id}, socket) do
  {:noreply, delete_message(socket, message_id)}
end

def delete_message(socket, message_id) do
  message = Chat.get_message!(message_id)
  Chat.delete_message(message)
  stream_delete(socket, :messages, message)
end

Wrap Up

LiveView’s new streams feature packs a powerful punch! It allows us to build and manage large datasets client-side with very little custom code. Our interactive, real-time chatting feature uses streams to manage chat messages fully on the client. Client-side data management with streams opens up new possibilities for LiveView developers to efficiently manage large data collections.

Fly.io is a great way to run your Phoenix LiveView app close to your users. It’s easy to get started.