Alter Action Cable Element to be morph-compatible

Closes [#638][]

Recent changes to integrate with morphing have altered the mental model
for some Turbo custom elements, including the
`<turbo-cable-stream-source>` element.

Custom Elements' `connectedCallback()` and `disconnectedCallback()`
(along with Stimulus' `connect()` and `disconnect()`) improved upon
invoking code immediately, or listening for `DOMContentLoaded` events.

There are similar improvements to be made to integrate with morphing.
First, [observe attribute changes][] by declaring their own `static
observedAttributes` properties along with
`attributeChangedCallback(name, oldValue, newValue)` callbacks. Those
callbacks execute the same initialization code as their current
`connectedCallback()` and `disconnectedCallback()` methods.

That'll help resolve this issue. In addition to those changes, it's
important to emphasize this pattern for consumer applications moving
forward. JavaScript code (whether Stimulus controller or otherwise)
should be implemented in a way that' resilient to both asynchronous
connection and disconnection *as well as* asynchronous modification of
attributes.

[#638]: #638
[observe attribute changes]: https://developer.mozilla.org/en-US/docs/Web/API/Web_Components/Using_custom_elements#responding_to_attribute_changes

app/assets/javascripts/turbo.js

@@ -5261,6 +5261,7 @@ function walk(obj) {
 }
 
 class TurboCableStreamSourceElement extends HTMLElement {
+  static observedAttributes=[ "channel", "signed-stream-name" ];
   async connectedCallback() {
     connectStreamSource(this);
     this.subscription = await subscribeTo(this.channel, {
@@ -5272,6 +5273,17 @@ class TurboCableStreamSourceElement extends HTMLElement {
   disconnectedCallback() {
     disconnectStreamSource(this);
     if (this.subscription) this.subscription.unsubscribe();
+    this.subscriptionDisconnected();
+  }
+  attributeChangedCallback(name) {
+    switch (name) {
+     case "channel":
+     case "signed-stream-name":
+      {
+        this.disconnectedCallback();
+        this.connectedCallback();
+      }
+    }
   }
   dispatchMessageEvent(data) {
     const event = new MessageEvent("message", {

app/javascript/turbo/cable_stream_source_element.js

@@ -3,6 +3,8 @@ import { subscribeTo } from "./cable"
 import snakeize from "./snakeize"
 
 class TurboCableStreamSourceElement extends HTMLElement {
+  static observedAttributes = ["channel", "signed-stream-name"]
+
   async connectedCallback() {
     connectStreamSource(this)
     this.subscription = await subscribeTo(this.channel, {
@@ -15,6 +17,17 @@ class TurboCableStreamSourceElement extends HTMLElement {
   disconnectedCallback() {
     disconnectStreamSource(this)
     if (this.subscription) this.subscription.unsubscribe()
+    this.subscriptionDisconnected()
+  }
+
+  attributeChangedCallback(name) {
+    switch (name) {
+      case "channel":
+      case "signed-stream-name": {
+        this.disconnectedCallback()
+        this.connectedCallback()
+      }
+    }
   }
 
   dispatchMessageEvent(data) {

test/dummy/app/controllers/messages_controller.rb

@@ -1,6 +1,9 @@
 class MessagesController < ApplicationController
   def show
-    @message = Message.new(id: 1, content: "My message")
+    respond_to do |format|
+      format.turbo_stream { @message = Message.new(id: 1, content: "My message") }
+      format.html { @message = Message.find(params[:id]) }
+    end
   end
 
   def index

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

@@ -0,0 +1,5 @@
+<%= turbo_stream_from @message %>
+
+<div id="<%= dom_id @message %>">
+  <%= render @message %>
+</div>

test/system/broadcasts_test.rb

@@ -2,6 +2,7 @@
 
 class BroadcastsTest < ApplicationSystemTestCase
   include ActiveJob::TestHelper
+  extend Turbo::Streams::StreamName
 
   test "Message broadcasts Turbo Streams" do
     visit messages_path
@@ -118,8 +119,30 @@ class BroadcastsTest < ApplicationSystemTestCase
     end
   end
 
+  test "the turbo-cable-stream-source observes attribute changes" do
+    original, broadcasted = Message.create! [{content: "original content"}, {content: "broadcasted content"}]
+
+    visit message_path(original)
+    wait_for_stream_to_be_connected
+    reconnect_cable_stream_source from: original, to: broadcasted
+
+    assert_text original.content
+    assert_broadcasts_text broadcasted.content, to: dom_id(original) do
+      broadcasted.broadcast_update target: dom_id(original)
+    end
+    assert_no_text original.content
+  end
+
   private
 
+  def reconnect_cable_stream_source(from:, to:)
+    cable_stream_source = find("turbo-cable-stream-source[signed-stream-name=#{signed_stream_name(from)}]")
+
+    cable_stream_source.execute_script <<~JS, signed_stream_name(to)
+      this.setAttribute("signed-stream-name", arguments[0])
+    JS
+  end
+
   def wait_for_stream_to_be_connected
     assert_selector "turbo-cable-stream-source[connected]", visible: false
   end