Support for page refreshes and broadcasting

app/assets/javascripts/turbo.js

@@ -4060,14 +4060,14 @@ var turbo_es2017Esm = Object.freeze({
 let consumer;
 
 async function getConsumer() {
-  return consumer || setConsumer(createConsumer().then(setConsumer));
+  return consumer || setConsumer(createConsumer$1().then(setConsumer));
 }
 
 function setConsumer(newConsumer) {
   return consumer = newConsumer;
 }
 
-async function createConsumer() {
+async function createConsumer$1() {
   const {createConsumer: createConsumer} = await Promise.resolve().then((function() {
     return index;
   }));
@@ -4083,7 +4083,7 @@ var cable = Object.freeze({
   __proto__: null,
   getConsumer: getConsumer,
   setConsumer: setConsumer,
-  createConsumer: createConsumer,
+  createConsumer: createConsumer$1,
   subscribeTo: subscribeTo
 });
 
@@ -4309,6 +4309,8 @@ ConnectionMonitor.staleThreshold = 6;
 
 ConnectionMonitor.reconnectionBackoffRate = .15;
 
+var ConnectionMonitor$1 = ConnectionMonitor;
+
 var INTERNAL = {
   message_types: {
     welcome: "welcome",
@@ -4320,7 +4322,8 @@ var INTERNAL = {
   disconnect_reasons: {
     unauthorized: "unauthorized",
     invalid_request: "invalid_request",
-    server_restart: "server_restart"
+    server_restart: "server_restart",
+    remote: "remote"
   },
   default_mount_path: "/cable",
   protocols: [ "actioncable-v1-json", "actioncable-unsupported" ]
@@ -4337,7 +4340,7 @@ class Connection {
     this.open = this.open.bind(this);
     this.consumer = consumer;
     this.subscriptions = this.consumer.subscriptions;
-    this.monitor = new ConnectionMonitor(this);
+    this.monitor = new ConnectionMonitor$1(this);
     this.disconnected = true;
   }
   send(data) {
@@ -4353,11 +4356,12 @@ class Connection {
       logger.log(`Attempted to open WebSocket, but existing socket is ${this.getState()}`);
       return false;
     } else {
-      logger.log(`Opening WebSocket, current state is ${this.getState()}, subprotocols: ${protocols}`);
+      const socketProtocols = [ ...protocols, ...this.consumer.subprotocols || [] ];
+      logger.log(`Opening WebSocket, current state is ${this.getState()}, subprotocols: ${socketProtocols}`);
       if (this.webSocket) {
         this.uninstallEventHandlers();
       }
-      this.webSocket = new adapters.WebSocket(this.consumer.url, protocols);
+      this.webSocket = new adapters.WebSocket(this.consumer.url, socketProtocols);
       this.installEventHandlers();
       this.monitor.start();
       return true;
@@ -4369,7 +4373,7 @@ class Connection {
     if (!allowReconnect) {
       this.monitor.stop();
     }
-    if (this.isActive()) {
+    if (this.isOpen()) {
       return this.webSocket.close();
     }
   }
@@ -4399,6 +4403,9 @@ class Connection {
   isActive() {
     return this.isState("open", "connecting");
   }
+  triedToReconnect() {
+    return this.monitor.reconnectAttempts > 0;
+  }
   isProtocolSupported() {
     return indexOf.call(supportedProtocols, this.getProtocol()) >= 0;
   }
@@ -4438,6 +4445,9 @@ Connection.prototype.events = {
     const {identifier: identifier, message: message, reason: reason, reconnect: reconnect, type: type} = JSON.parse(event.data);
     switch (type) {
      case message_types.welcome:
+      if (this.triedToReconnect()) {
+        this.reconnectAttempted = true;
+      }
       this.monitor.recordConnect();
       return this.subscriptions.reload();
 
@@ -4452,7 +4462,16 @@ Connection.prototype.events = {
 
      case message_types.confirmation:
       this.subscriptions.confirmSubscription(identifier);
-      return this.subscriptions.notify(identifier, "connected");
+      if (this.reconnectAttempted) {
+        this.reconnectAttempted = false;
+        return this.subscriptions.notify(identifier, "connected", {
+          reconnected: true
+        });
+      } else {
+        return this.subscriptions.notify(identifier, "connected", {
+          reconnected: false
+        });
+      }
 
      case message_types.rejection:
       return this.subscriptions.reject(identifier);
@@ -4487,6 +4506,8 @@ Connection.prototype.events = {
   }
 };
 
+var Connection$1 = Connection;
+
 const extend = function(object, properties) {
   if (properties != null) {
     for (let key in properties) {
@@ -4556,10 +4577,12 @@ class SubscriptionGuarantor {
   }
 }
 
+var SubscriptionGuarantor$1 = SubscriptionGuarantor;
+
 class Subscriptions {
   constructor(consumer) {
     this.consumer = consumer;
-    this.guarantor = new SubscriptionGuarantor(this);
+    this.guarantor = new SubscriptionGuarantor$1(this);
     this.subscriptions = [];
   }
   create(channelName, mixin) {
@@ -4636,7 +4659,8 @@ class Consumer {
   constructor(url) {
     this._url = url;
     this.subscriptions = new Subscriptions(this);
-    this.connection = new Connection(this);
+    this.connection = new Connection$1(this);
+    this.subprotocols = [];
   }
   get url() {
     return createWebSocketURL(this._url);
@@ -4657,6 +4681,9 @@ class Consumer {
       return this.connection.open();
     }
   }
+  addSubProtocol(subprotocol) {
+    this.subprotocols = [ ...this.subprotocols, subprotocol ];
+  }
 }
 
 function createWebSocketURL(url) {
@@ -4674,7 +4701,7 @@ function createWebSocketURL(url) {
   }
 }
 
-function createConsumer$1(url = getConfig("url") || INTERNAL.default_mount_path) {
+function createConsumer(url = getConfig("url") || INTERNAL.default_mount_path) {
   return new Consumer(url);
 }
 
@@ -4687,17 +4714,17 @@ function getConfig(name) {
 
 var index = Object.freeze({
   __proto__: null,
-  Connection: Connection,
-  ConnectionMonitor: ConnectionMonitor,
+  Connection: Connection$1,
+  ConnectionMonitor: ConnectionMonitor$1,
   Consumer: Consumer,
   INTERNAL: INTERNAL,
   Subscription: Subscription,
   Subscriptions: Subscriptions,
-  SubscriptionGuarantor: SubscriptionGuarantor,
+  SubscriptionGuarantor: SubscriptionGuarantor$1,
   adapters: adapters,
   createWebSocketURL: createWebSocketURL,
   logger: logger,
-  createConsumer: createConsumer$1,
+  createConsumer: createConsumer,
   getConfig: getConfig
 });
 

app/channels/turbo/streams/broadcasts.rb

@@ -33,6 +33,10 @@ def broadcast_prepend_to(*streamables, **opts)
     broadcast_action_to(*streamables, action: :prepend, **opts)
   end
 
+  def broadcast_refresh_to(*streamables, **opts)
+    broadcast_stream_to(*streamables, content: turbo_stream_refresh_tag)
+  end
+
   def broadcast_action_to(*streamables, action:, target: nil, targets: nil, attributes: {}, **rendering)
     broadcast_stream_to(*streamables, content: turbo_stream_action_tag(action, target: target, targets: targets, template:
       rendering.delete(:content) || rendering.delete(:html) || (rendering[:render] != false && rendering.any? ? render_format(:html, **rendering) : nil),
@@ -64,6 +68,12 @@ def broadcast_prepend_later_to(*streamables, **opts)
     broadcast_action_later_to(*streamables, action: :prepend, **opts)
   end
 
+  def broadcast_refresh_later_to(*streamables, request_id: Turbo.current_request_id, **opts)
+    refresh_debouncer_for(*streamables, request_id: request_id).debounce do
+      Turbo::Streams::BroadcastStreamJob.perform_later stream_name_from(streamables), content: turbo_stream_refresh_tag(request_id: request_id, **opts)
+    end
+  end
+
   def broadcast_action_later_to(*streamables, action:, target: nil, targets: nil, attributes: {}, **rendering)
     Turbo::Streams::ActionBroadcastJob.perform_later \
       stream_name_from(streamables), action: action, target: target, targets: targets, attributes: attributes, **rendering
@@ -81,6 +91,9 @@ def broadcast_stream_to(*streamables, content:)
     ActionCable.server.broadcast stream_name_from(streamables), content
   end
 
+  def refresh_debouncer_for(*streamables, request_id: nil) # :nodoc:
+    Turbo::ThreadDebouncer.for("turbo-refresh-debouncer-#{stream_name_from(streamables.including(request_id))}")
+  end
 
   private
     def render_format(format, **rendering)

app/controllers/concerns/turbo/request_id_tracking.rb

@@ -0,0 +1,12 @@
+module Turbo::RequestIdTracking
+  extend ActiveSupport::Concern
+
+  included do
+    around_action :turbo_tracking_request_id
+  end
+
+  private
+    def turbo_tracking_request_id(&block)
+      Turbo.with_request_id(request.headers["X-Turbo-Request-Id"], &block)
+    end
+end

app/helpers/turbo/drive_helper.rb

@@ -26,4 +26,12 @@ def turbo_exempts_page_from_preview
   def turbo_page_requires_reload
     provide :head, tag.meta(name: "turbo-visit-control", content: "reload")
   end
+
+  def turbo_refreshes_with(method: :replace, scroll: :reset)
+    raise ArgumentError, "Invalid refresh option '#{method}'" unless method.in?(%i[ replace morph ])
+    raise ArgumentError, "Invalid scroll option '#{scroll}'" unless scroll.in?(%i[ reset preserve ])
+
+    provide :head, tag.meta(name: "turbo-refresh-method", content: method)
+    provide :head, tag.meta(name: "turbo-refresh-scroll", content: scroll)
+  end
 end

app/helpers/turbo/streams/action_helper.rb

@@ -24,7 +24,7 @@ module Turbo::Streams::ActionHelper
   #   # => <turbo-stream action="remove" target="special_message_1"></turbo-stream>
   #
   def turbo_stream_action_tag(action, target: nil, targets: nil, template: nil, **attributes)
-    template = action.to_sym == :remove ? "" : tag.template(template.to_s.html_safe)
+    template = action.to_sym.in?(%i[ remove refresh ]) ? "" : tag.template(template.to_s.html_safe)
 
     if target = convert_to_turbo_stream_dom_id(target)
       tag.turbo_stream(template, **attributes, action: action, target: target)
@@ -35,6 +35,10 @@ def turbo_stream_action_tag(action, target: nil, targets: nil, template: nil, **
     end
   end
 
+  def turbo_stream_refresh_tag(request_id: Turbo.current_request_id, **attributes)
+    turbo_stream_action_tag(:refresh, **{ "request-id": request_id }.compact, **attributes)
+  end
+
   private
     def convert_to_turbo_stream_dom_id(target, include_selector: false)
       if Array(target).any? { |value| value.respond_to?(:to_key) }

app/jobs/turbo/streams/broadcast_stream_job.rb

@@ -0,0 +1,7 @@
+class Turbo::Streams::BroadcastStreamJob < ActiveJob::Base
+  discard_on ActiveJob::DeserializationError
+
+  def perform(stream, content:)
+    Turbo::StreamsChannel.broadcast_stream_to(stream, content: content)
+  end
+end