Skip to content

Instantly share code, notes, and snippets.

@palkan
Last active December 14, 2022 21:22
Show Gist options
  • Save palkan/65934d1f23b57e59756a61053e79a2d1 to your computer and use it in GitHub Desktop.
Save palkan/65934d1f23b57e59756a61053e79a2d1 to your computer and use it in GitHub Desktop.
turbo_history_stream_from
<!-- 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>
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
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);
# 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
# 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
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment