Allow turbo-cable-stream-source to be compatible with data-turbo-permanent

app/assets/javascripts/turbo.js

@@ -5386,25 +5386,44 @@ function walk(obj) {
 
 class TurboCableStreamSourceElement extends HTMLElement {
   static observedAttributes=[ "channel", "signed-stream-name" ];
+  constructor() {
+    super();
+    this.beforeTurboRender = this.beforeTurboRender.bind(this);
+    this.afterTurboRender = this.afterTurboRender.bind(this);
+  }
   async connectedCallback() {
+    document.addEventListener("turbo:before-render", this.beforeTurboRender);
+    document.addEventListener("turbo:render", this.afterTurboRender);
+    if (!this.withinTurboRender) {
+      await this.subscribe();
+    }
+  }
+  async disconnectedCallback() {
+    document.removeEventListener("turbo:before-render", this.beforeTurboRender);
+    document.removeEventListener("turbo:render", this.afterTurboRender);
+    if (!this.withinTurboRender) {
+      this.unsubscribe();
+    }
+  }
+  async attributeChangedCallback() {
+    if (this.subscription) {
+      this.unsubscribe();
+      await this.subscribe();
+    }
+  }
+  async subscribe() {
     connectStreamSource(this);
     this.subscription = await subscribeTo(this.channel, {
       received: this.dispatchMessageEvent.bind(this),
       connected: this.subscriptionConnected.bind(this),
       disconnected: this.subscriptionDisconnected.bind(this)
     });
   }
-  disconnectedCallback() {
+  unsubscribe() {
     disconnectStreamSource(this);
     if (this.subscription) this.subscription.unsubscribe();
     this.subscriptionDisconnected();
   }
-  attributeChangedCallback() {
-    if (this.subscription) {
-      this.disconnectedCallback();
-      this.connectedCallback();
-    }
-  }
   dispatchMessageEvent(data) {
     const event = new MessageEvent("message", {
       data: data
@@ -5417,6 +5436,15 @@ class TurboCableStreamSourceElement extends HTMLElement {
   subscriptionDisconnected() {
     this.removeAttribute("connected");
   }
+  beforeTurboRender() {
+    this.withinTurboRender = true;
+  }
+  afterTurboRender() {
+    if (this.withinTurboRender && !this.isConnected) {
+      this.unsubscribe();
+    }
+    this.withinTurboRender = false;
+  }
   get channel() {
     const channel = this.getAttribute("channel");
     const signed_stream_name = this.getAttribute("signed-stream-name");

app/javascript/turbo/cable_stream_source_element.js

@@ -1,11 +1,42 @@
-import { connectStreamSource, disconnectStreamSource } from "@hotwired/turbo"
-import { subscribeTo } from "./cable"
+import {connectStreamSource, disconnectStreamSource} from "@hotwired/turbo"
+import {subscribeTo} from "./cable"
 import snakeize from "./snakeize"
 
 class TurboCableStreamSourceElement extends HTMLElement {
   static observedAttributes = ["channel", "signed-stream-name"]
 
+  constructor() {
+    super()
+    this.beforeTurboRender = this.beforeTurboRender.bind(this)
+    this.afterTurboRender = this.afterTurboRender.bind(this)
+  }
+
   async connectedCallback() {
+    document.addEventListener("turbo:before-render", this.beforeTurboRender)
+    document.addEventListener("turbo:render", this.afterTurboRender)
+
+    if (!this.withinTurboRender) {
+      await this.subscribe()
+    }
+  }
+
+  async disconnectedCallback() {
+    document.removeEventListener("turbo:before-render", this.beforeTurboRender)
+    document.removeEventListener("turbo:render", this.afterTurboRender)
+
+    if (!this.withinTurboRender) {
+      this.unsubscribe()
+    }
+  }
+
+  async attributeChangedCallback() {
+    if (this.subscription) {
+      this.unsubscribe()
+      await this.subscribe()
+    }
+  }
+
+  async subscribe() {
     connectStreamSource(this)
     this.subscription = await subscribeTo(this.channel, {
       received: this.dispatchMessageEvent.bind(this),
@@ -14,19 +45,12 @@ class TurboCableStreamSourceElement extends HTMLElement {
     })
   }
 
-  disconnectedCallback() {
+  unsubscribe() {
     disconnectStreamSource(this)
     if (this.subscription) this.subscription.unsubscribe()
     this.subscriptionDisconnected()
   }
 
-  attributeChangedCallback() {
-    if (this.subscription) {
-      this.disconnectedCallback()
-      this.connectedCallback()
-    }
-  }
-
   dispatchMessageEvent(data) {
     const event = new MessageEvent("message", { data })
     return this.dispatchEvent(event)
@@ -40,10 +64,21 @@ class TurboCableStreamSourceElement extends HTMLElement {
     this.removeAttribute("connected")
   }
 
+  beforeTurboRender() {
+    this.withinTurboRender = true;
+  }
+
+  afterTurboRender() {
+    if (this.withinTurboRender && !this.isConnected) {
+      this.unsubscribe();
+    }
+    this.withinTurboRender = false;
+  }
+
   get channel() {
     const channel = this.getAttribute("channel")
     const signed_stream_name = this.getAttribute("signed-stream-name")
-    return { channel, signed_stream_name, ...snakeize({ ...this.dataset }) }
+    return {channel, signed_stream_name, ...snakeize({...this.dataset})}
   }
 }
 

test/dummy/app/controllers/messages_controller.rb

@@ -11,6 +11,10 @@ def index
     @messages = Message.all
   end
 
+  def permanent
+    @messages = Message.all
+  end
+
   def section
   end
 

test/dummy/app/views/messages/permanent.html.erb

@@ -0,0 +1,12 @@
+<% navigations = params.fetch(:navigations, 0).to_i %>
+
+<h1>Permanent Messages</h1>
+
+<p>Navigations: <%= navigations %></p>
+<%= link_to "Navigate", permanent_messages_path(navigations: navigations + 1) %>
+
+<div id="messages-wrapper" data-turbo-permanent>
+  <%= turbo_stream_from "messages" %>
+  <div id="messages">
+  </div>
+</div>

test/dummy/config/routes.rb

@@ -5,6 +5,7 @@
   resources :messages do
     collection do
       get :section
+      get :permanent
     end
   end
   resources :trays

test/system/broadcasts_test.rb

@@ -120,6 +120,39 @@ class BroadcastsTest < ApplicationSystemTestCase
     assert_no_text original.content
   end
 
+  test "the turbo-cable-stream-source does not unsubscribe+resubscribe within turbo-permanent" do
+    visit permanent_messages_path
+    assert_selector "turbo-cable-stream-source[connected]"
+
+    cable_stream_source = find("turbo-cable-stream-source")
+    cable_stream_source.execute_script <<~JS
+      const el = this;
+
+      let removedOnce = false;
+      el.reconnectedObserver = new MutationObserver((mutations) => {
+        mutations.forEach((m) => {
+          const isConnected = el.hasAttribute("connected");
+          if (m.oldValue === "" && !isConnected) {
+            removedOnce = true;
+            el.setAttribute("disconnected", "");
+          }
+          if (removedOnce && m.oldValue == null && isConnected) {
+            el.setAttribute("reconnected", "");
+          }
+        });
+      });
+      el.reconnectedObserver.observe(el, { attributes: true, attributeOldValue: true, attributeFilter: ["connected"] });
+    JS
+
+    assert_text "Navigations: 0"
+    click_link "Navigate"
+    assert_text "Navigations: 1"
+
+    assert_selector "turbo-cable-stream-source[connected]"
+    assert_no_selector "turbo-cable-stream-source[disconnected]"
+    assert_no_selector "turbo-cable-stream-source[reconnected]"
+  end
+
   private
 
   def reconnect_cable_stream_source(from:, to:)