turbo_history_stream_from

001_example.html.erb

<!--  index.html.erb -->
<div id="messages">
  <%= turbo_history_stream_from channel, params: {channel_id: channel.id, model: Channel, cursor: "#messages .message:last-child" %>
  <%= render messages %>
</div>

<!-- _message.html.erb -->
<div id="<%= dom_id(message) %>" data-cursor="<%= message.id %>" class="message">
  <%= message.content %>
</div>

002_example_model.rb

class Channel < ApplicationRecord
  has_many :messages

  after_commit on: :create do
    broadcast_append_to channel, partial: "messages/message", locals: {message: self}, target: "messages"
  end

  def self.turbo_history(turbo_channel, last_id, params)
    channel = Channel.find(params[:channel_id])

    channel.messages
      .preload(:user)
      .where("id > ?", last_id)
      .order(id: :asc).each do |message|
      turbo_channel.transmit_append target: "messages", partial: "messages/message", locals: {message:}
    end
  end
end

01_cable_stream_source_element.js

import { Turbo, cable } from "@hotwired/turbo-rails";

const { connectStreamSource, disconnectStreamSource } = Turbo;
const { subscribeTo } = cable;

class StreamSourceElement extends HTMLElement {
  async connectedCallback() {
    connectStreamSource(this);
    let element = this;

    this.subscription = await subscribeTo(this.channel, {
      initialized() {
        this.pending = true
        this.pendingMessages = []
      },
      disconnected() {
        this.pending = true
      },
      received(data) {
        if (data.type === "history_ack") {
          this.pending = false
          while(this.pendingMessages.length > 0){
            this.received(this.pendingMessages.shift())
          }
          return
        }
        if (this.pending) {
          return this.pendingMessages.push(data)
        }

        element.dispatchMessageEvent(data)
      },
      connected() {
        let selector = element.getAttribute("cursor-selector")
        if (!selector) return

        let el = document.querySelector(selector)
        if (!el) return

        let cursor = el.getAttribute("data-cursor")
        if (!cursor) return

        this.perform("history", { cursor })
      }
    });
  }

  disconnectedCallback() {
    disconnectStreamSource(this);
    if (this.subscription) this.subscription.unsubscribe();
  }

  dispatchMessageEvent(data) {
    const event = new MessageEvent("message", { data });
    return this.dispatchEvent(event);
  }

  get channel() {
    const channel = this.getAttribute("channel");
    const signed_stream_name = this.getAttribute("signed-stream-name");

    let params = {};

    const paramsJSON = this.getAttribute("params");

    if (paramsJSON) {
      params = JSON.parse(paramsJSON);
    }

    return { ...params, channel, signed_stream_name };
  }
}

customElements.define("turbo-cable-stream-source-history", StreamSourceElement);

02_turbo_channel.rb

# frozen_string_literal: true

class TurboChannel < Turbo::StreamsChannel
  include Turbo::Streams::TransmitAction

  def history(data)
    cursor = data.fetch("cursor")
    model = params[:model].classify.constantize
    model.turbo_history(self, cursor, params)
    transmit({type: "history_ack"})
  end
end

03_transmit_action.rb

# frozen_string_literal: true

module Turbo
  module Streams
    module TransmitAction
      include Turbo::Streams::ActionHelper

      def transmit_remove(**opts)
        transmit_action(action: :remove, **opts)
      end

      def transmit_replace(**opts)
        transmit_action(action: :replace, **opts)
      end

      def transmit_update(**opts)
        transmit_action(action: :update, **opts)
      end

      def transmit_before(**opts)
        transmit_action(action: :before, **opts)
      end

      def transmit_after(**opts)
        transmit_action(action: :after, **opts)
      end

      def transmit_append(**opts)
        transmit_action(action: :append, **opts)
      end

      def transmit_prepend(**opts)
        transmit_action(action: :prepend, **opts)
      end

      def transmit_action(action:, target: nil, targets: nil, **rendering)
        payload = turbo_stream_action_tag(
          action, target: target, targets: targets,
          template: rendering.delete(:content) || (rendering.any? ? render_format(:html, **rendering) : nil)
        )

        transmit payload
      end

      private

      def render_format(format, **rendering)
        ApplicationController.render(formats: [format], **rendering)
      end
    end
  end
end

04_helper.rb

module ApplicationHelper
  def turbo_history_stream_from(*streamables, **attributes)
    attributes[:channel] = attributes[:channel]&.to_s || "TurboChannel"
    attributes[:"signed-stream-name"] = Turbo::StreamsChannel.signed_stream_name(streamables)
    attributes[:params] = attributes[:params]&.to_json
    attributes[:"cursor-selector"] = attributes.delete(:cursor)

    tag.turbo_cable_stream_source_history(**attributes)
  end
end