feat: Anycable

.github/workflows/main.yml

@@ -5,8 +5,7 @@ jobs:
     strategy:
       matrix:
         ruby-version:
-          - "2.7"
-          - "3.0"
+          - "3.1"
           - "3.2.2"
         rails-version:
           - "6.1"
@@ -22,6 +21,8 @@ jobs:
       FANOUT_FASTLY_KEY: ${{ secrets.FANOUT_FASTLY_KEY }}
       FANOUT_SERVICE_ID: ${{ secrets.FANOUT_SERVICE_ID }}
       FANOUT_SERVICE_URL: ${{ secrets.FANOUT_SERVICE_URL }}
+      ANYCABLE_URL: ${{ secrets.ANYCABLE_URL }}
+      ANYCABLE_BROADCAST_KEY: ${{ secrets.ANYCABLE_BROADCAST_KEY }}
 
     name: ${{ format('Tests (Ruby {0}, Rails {1})', matrix.ruby-version, matrix.rails-version) }}
     runs-on: ubuntu-latest
@@ -56,5 +57,10 @@ jobs:
           TURBO_TRAIN_TEST_SERVER: fanout
         run: |
           bin/test test/**/*_test.rb
-
+
+      - name: Run tests [anycable]
+        env:
+          TURBO_TRAIN_TEST_SERVER: anycable
+        run: |
+          bin/test test/**/*_test.rb
 

Gemfile

@@ -21,16 +21,10 @@ end
 
 group :test do
   gem 'puma'
-  gem 'capybara'
+  gem 'capybara', '>= 3.39.2'
   gem 'rexml'
-  # Locked because on 4.9.1 getting error:
-  # BroadcastingTest#test_Turbo::Train_broadcasts_Turbo_Streams:
-  # ArgumentError: wrong number of arguments (given 2, expected 0..1)
-  #     selenium-webdriver-4.9.1/lib/selenium/webdriver/common/logger.rb:51:in `initialize'
-  # https://github.com/SeleniumHQ/selenium/issues/12013
-  gem 'selenium-webdriver', '4.9.0'
-  gem 'webdrivers'
-  gem 'sqlite3'
+  gem 'selenium-webdriver', '4.20.0'
+  gem 'sqlite3', '~> 1.4'
 end
 
 # Start debugger with binding.b [https://github.com/ruby/debug]

README.md

@@ -6,7 +6,7 @@
      src="https://user-images.githubusercontent.com/3010927/210603861-4b265489-a4a7-4d2a-bceb-40ceccebcd96.jpg">
 
 
-Real-time page updates for your Rails app over SSE with [Mercure](https://mercure.rocks) or [Fanout Cloud](https://fanout.io/cloud) and [Hotwire Turbo](https://turbo.hotwired.dev/handbook/streams#integration-with-server-side-frameworks).
+Real-time page updates for your Rails app over SSE with [Mercure](https://mercure.rocks), [Fanout Cloud](https://fanout.io/cloud) or [AnyCable](https://anycable.io/) and [Hotwire Turbo](https://turbo.hotwired.dev/handbook/streams#integration-with-server-side-frameworks).
 
 * **Uses [SSE](https://html.spec.whatwg.org/multipage/server-sent-events.html)**. No more websockets, client libraries, JS code and handling reconnects. Just an HTTP connection. Let the [browser](https://caniuse.com/eventsource) do the work.
 * **Seamless Hotwire integration.** Use it exactly like [ActionCable](https://github.com/hotwired/turbo-rails#come-alive-with-turbo-streams). Drop-in replacement for `broadcast_action_to` and usual helpers.
@@ -66,6 +66,14 @@ We only support the cloud version today. To use [Fanout](https://fanout.io/cloud
 
 Coming soon.
 
+#### AnyCable
+
+```
+anycable-go --host=localhost --port=8080 --sse --broadcast_adapter=http --broadcast_key=test --public_streams --noauth
+```
+
+Coming soon.
+
 ## Usage
 
 If you are familiar with broadcasting from ActionCable, usage would be extremely familiar:
@@ -124,6 +132,11 @@ Turbo::Train.configure do |config|
     fanout.service_id = ...
     fanout.fastly_key = ...
   end
+
+  config.server :anycable do |fanout| 
+    ac.anycable_url = 'http://0.0.0.0:8080'
+    ac.broadcast_key = 'test'
+  end
 end
 ```
 

lib/turbo/train.rb

@@ -4,6 +4,7 @@
 require 'turbo/train/base_server'
 require 'turbo/train/mercure_server'
 require 'turbo/train/fanout_server'
+require 'turbo/train/anycable_server'
 require 'turbo/train/test_server'
 require 'turbo/train/test_helper'
 require "turbo/train/engine"

lib/turbo/train/anycable_server.rb

@@ -0,0 +1,36 @@
+module Turbo
+  module Train
+    class AnycableServer < BaseServer
+      def publish(topics:, data:)
+        uri = URI(server_config.publish_url)
+        req = Net::HTTP::Post.new(uri)
+        req['Content-Type'] = 'application/json'
+        req['Authorization'] = "Bearer #{server_config.broadcast_key}"
+
+        message = data[:data].gsub("\n", '')
+
+        opts = {
+          use_ssl: uri.scheme == 'https'
+        }
+
+        payload = []
+
+        Array(topics).each do |topic|
+          payload << { stream: topic, data: message }
+        end
+
+        req.body = payload.to_json
+
+        opts[:verify_mode] = OpenSSL::SSL::VERIFY_NONE if configuration.skip_ssl_verification
+
+        Net::HTTP.start(uri.host, uri.port, opts) do |http|
+          http.request(req)
+        end
+      end
+
+      def server_config
+        configuration.anycable
+      end
+    end
+  end
+end

lib/turbo/train/config.rb

@@ -47,13 +47,32 @@ def listen_url(topic, **)
       end
     end
 
+    class AnycableConfiguration
+      attr_accessor :anycable_url, :broadcast_key
+
+      def initialize
+        super
+        @anycable_url = 'http://localhost:8080'
+        @broadcast_key = 'test'
+      end
+
+      def publish_url
+        "#{@anycable_url}/_broadcast"
+      end
+
+      def listen_url(topic, **)
+        "#{@anycable_url}/events?stream=#{Turbo::Train.signed_stream_name(topic)}"
+      end
+    end
+
     class Configuration
-      attr_accessor :skip_ssl_verification, :mercure, :fanout, :default_server
+      attr_accessor :skip_ssl_verification, :mercure, :fanout, :anycable, :default_server
 
       def initialize
         @skip_ssl_verification = Rails.env.development? || Rails.env.test?
         @mercure = nil
         @fanout = nil
+        @anycable = nil
         @default_server = :mercure
       end
 
@@ -65,6 +84,9 @@ def server(server_name)
         when :fanout
           @fanout ||= FanoutConfiguration.new
           yield(@fanout)
+        when :anycable
+          @anycable ||= AnycableConfiguration.new
+          yield(@anycable)
         else
           raise ArgumentError, "Unknown server name: #{server_name}"
         end

lib/turbo/train/test_helper.rb

@@ -8,6 +8,8 @@ def before_setup
                         Turbo::Train::TestServer.new(Turbo::Train.mercure_server, Turbo::Train.configuration)
                       when :fanout
                         Turbo::Train::TestServer.new(Turbo::Train.fanout_server, Turbo::Train.configuration)
+                      when :anycable
+                        Turbo::Train::TestServer.new(Turbo::Train.anycable_server, Turbo::Train.configuration)
                       else
                         raise "Unknown test server: #{ENV['TURBO_TRAIN_TEST_SERVER']}"
                       end
@@ -22,6 +24,8 @@ def after_teardown
                         Turbo::Train.mercure_server
                       when :fanout
                         Turbo::Train.fanout_server
+                      when :anycable
+                        Turbo::Train.anycable_server
                       else
                         raise "Unknown test server: #{ENV['TURBO_TRAIN_TEST_SERVER']}"
                       end
@@ -60,6 +64,20 @@ def assert_body_match(r)
           assert_match "Published\n", r.body
         elsif Turbo::Train.server.real_server.is_a?(Turbo::Train::MercureServer)
           assert_match /urn:uuid:.*/, r.body
+        elsif Turbo::Train.server.real_server.is_a?(Turbo::Train::AnycableServer)
+          assert_match '', r.body
+        else
+          raise "Unknown server type"
+        end
+      end
+
+      def assert_code_ok(r)
+        if Turbo::Train.server.real_server.is_a?(Turbo::Train::FanoutServer)
+          assert_equal r.code, '200'
+        elsif Turbo::Train.server.real_server.is_a?(Turbo::Train::MercureServer)
+          assert_equal r.code, '200'
+        elsif Turbo::Train.server.real_server.is_a?(Turbo::Train::AnycableServer)
+          assert_equal r.code, '201'
         else
           raise "Unknown server type"
         end

lib/turbo/train/train.rb

@@ -24,6 +24,8 @@ def server(server = nil)
           mercure_server
         when :fanout
           fanout_server
+        when :anycable
+          anycable_server
         else
           raise ArgumentError, "Unknown server: #{server}"
         end
@@ -41,6 +43,12 @@ def fanout_server
         @fanout_server ||= FanoutServer.new(configuration)
       end
 
+      def anycable_server
+        raise ArgumentError, "Anycable configuration is missing" unless configuration.anycable
+
+        @anycable_server ||= AnycableServer.new(configuration)
+      end
+
       def stream_name_from(streamables)
         if streamables.is_a?(Array)
           streamables.map  { |streamable| stream_name_from(streamable) }.join(":")

lib/turbo/train/version.rb

@@ -1,5 +1,5 @@
 module Turbo
   module Train
-    VERSION = "0.3.0"
+    VERSION = "0.4.0"
   end
 end

test/dummy/config/initializers/turbo_train.rb

@@ -1,15 +1,26 @@
 Turbo::Train.configure do |config|
   config.default_server = ENV.fetch('TURBO_TRAIN_TEST_SERVER', 'mercure').to_sym
 
-  config.server :mercure do |mercure|
-    mercure.mercure_domain = ENV['MERCURE_DOMAIN'] || raise('MERCURE_DOMAIN is not set')
-    mercure.publisher_key = ENV['MERCURE_PUBLISHER_KEY'] || raise('MERCURE_PUBLISHER_KEY is not set')
-    mercure.subscriber_key = ENV['MERCURE_SUBSCRIBER_KEY'] || raise('MERCURE_SUBSCRIBER_KEY is not set')
+  if ENV.fetch('TURBO_TRAIN_TEST_SERVER', 'mercure').to_sym == :mercure
+    config.server :mercure do |mercure|
+      mercure.mercure_domain = ENV['MERCURE_DOMAIN'] || raise('MERCURE_DOMAIN is not set')
+      mercure.publisher_key = ENV['MERCURE_PUBLISHER_KEY'] || raise('MERCURE_PUBLISHER_KEY is not set')
+      mercure.subscriber_key = ENV['MERCURE_SUBSCRIBER_KEY'] || raise('MERCURE_SUBSCRIBER_KEY is not set')
+    end
   end
 
-  config.server :fanout do |fanout|
-    fanout.service_url = ENV['FANOUT_SERVICE_URL'] || raise('FANOUT_SERVICE_URL is not set')
-    fanout.service_id = ENV['FANOUT_SERVICE_ID'] || raise('FANOUT_SERVICE_ID is not set')
-    fanout.fastly_key = ENV['FANOUT_FASTLY_KEY'] || raise('FANOUT_FASTLY_KEY is not set')
+  if ENV.fetch('TURBO_TRAIN_TEST_SERVER', 'mercure').to_sym == :fanout
+    config.server :fanout do |fanout|
+      fanout.service_url = ENV['FANOUT_SERVICE_URL'] || raise('FANOUT_SERVICE_URL is not set')
+      fanout.service_id = ENV['FANOUT_SERVICE_ID'] || raise('FANOUT_SERVICE_ID is not set')
+      fanout.fastly_key = ENV['FANOUT_FASTLY_KEY'] || raise('FANOUT_FASTLY_KEY is not set')
+    end
   end
-end
+
+  if ENV.fetch('TURBO_TRAIN_TEST_SERVER', 'mercure').to_sym == :anycable
+    config.server :anycable do |anycable|
+      anycable.anycable_url = ENV['ANYCABLE_URL'] || raise('ANYCABLE_URL is not set')
+      anycable.broadcast_key = ENV['ANYCABLE_BROADCAST_KEY'] || raise('ANYCABLE_BROADCAST_KEY is not set')
+    end
+  end
+end

test/train/broadcasts_test.rb

@@ -11,71 +11,71 @@ class BroadcastsTest < ActiveSupport::TestCase
     assert_broadcast_on "messages", turbo_stream_action_tag("replace", target: "message_1", template: "Goodbye!") do
       r = Turbo::Train.broadcast_render_to("messages", partial: 'messages/message', locals: { message: message })
 
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end
 
   test "broadcast_action_to" do
     assert_broadcast_on 'messages', turbo_stream_action_tag("replace", target: "target", template: 'content') do
       r = Turbo::Train.broadcast_action_to('messages', action: 'replace', target: 'target', content: 'content')
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end
 
   test "broadcast_append_to" do
     assert_broadcast_on 'messages', turbo_stream_action_tag("append", target: "target", template: 'content') do
       r = Turbo::Train.broadcast_append_to('messages', target: 'target', content: 'content')
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end
 
   test "broadcast_remove_to" do
     assert_broadcast_on 'messages', turbo_stream_action_tag("remove", target: "target") do
       r = Turbo::Train.broadcast_remove_to('messages', target: 'target')
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end
 
   test "broadcast_replace_to" do
     assert_broadcast_on 'messages', turbo_stream_action_tag("replace", target: "target", template: 'content') do
       r = Turbo::Train.broadcast_replace_to('messages', target: 'target', content: 'content')
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end
 
   test "broadcast_update_to" do
     assert_broadcast_on 'messages', turbo_stream_action_tag("update", target: "target", template: 'content') do
       r = Turbo::Train.broadcast_update_to('messages', target: 'target', content: 'content')
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end
 
   test "broadcast_before_to" do
     assert_broadcast_on 'messages', turbo_stream_action_tag("before", target: "target", template: 'content') do
       r = Turbo::Train.broadcast_before_to('messages', target: 'target', content: 'content')
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end
 
   test "broadcast_after_to" do
     assert_broadcast_on 'messages', turbo_stream_action_tag("after", target: "target", template: 'content') do
       r = Turbo::Train.broadcast_after_to('messages', target: 'target', content: 'content')
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end
 
   test "broadcast_prepend_to" do
     assert_broadcast_on 'messages', turbo_stream_action_tag("prepend", target: "target", template: 'content') do
       r = Turbo::Train.broadcast_prepend_to('messages', target: 'target', content: 'content')
-      assert_equal r.code, '200'
+      assert_code_ok(r)
       assert_body_match(r)
     end
   end