Add integration tests...

.github/workflows/test.yml

@@ -10,50 +10,30 @@ on:
 
 jobs:
   test:
-    name: "GraphQL-Ruby ${{ matrix.graphql }} (interpreter: ${{ matrix.interpreter }}) with AnyCable ${{ matrix.anycable }} on Ruby ${{ matrix.ruby }}"
+    name: "GraphQL-Ruby ${{ matrix.graphql }} (interpreter: ${{ matrix.interpreter }}, use_client_id: ${{ matrix.client_id }}) on Ruby ${{ matrix.ruby }}"
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
         include:
           - ruby: "3.0"
             graphql: '~> 1.12.0'
-            anycable: '~> 1.1.0'
+            client_id: 'false'
             interpreter: yes
-          - ruby: "3.0"
-            graphql: '~> 1.12.0'
-            anycable: '~> 1.1.0'
-            interpreter: no
           - ruby: 2.7
             graphql: '~> 1.12.0'
-            anycable: '~> 1.1.0'
+            client_id: 'false'
             interpreter: yes
-          - ruby: 2.7
-            graphql: '~> 1.12.0'
-            anycable: '~> 1.1.0'
-            interpreter: no
           - ruby: 2.6
             graphql: '~> 1.11.0'
-            anycable: '~> 1.0.0'
-            interpreter: yes
-          - ruby: 2.6
-            graphql: '~> 1.11.0'
-            anycable: '~> 1.0.0'
-            interpreter: no
-          - ruby: 2.5
-            graphql: '~> 1.11.0'
-            anycable: '~> 1.0.0'
-            interpreter: yes
-          - ruby: 2.5
-            graphql: '~> 1.11.0'
-            anycable: '~> 1.0.0'
+            client_id: 'true'
             interpreter: no
     container:
       image: ruby:${{ matrix.ruby }}
       env:
         CI: true
         GRAPHQL_RUBY_VERSION: ${{ matrix.graphql }}
-        ANYCABLE_VERSION: ${{ matrix.anycable }}
+        GRAPHQL_ANYCABLE_USE_CLIENT_PROVIDED_UNIQ_ID: ${{ matrix.client_id }}
     steps:
       - uses: actions/checkout@v2
       - uses: actions/cache@v2

Gemfile

@@ -9,6 +9,7 @@ gemspec
 
 gem "graphql",  ENV.fetch("GRAPHQL_RUBY_VERSION", "~> 1.12")
 gem "anycable", ENV.fetch("ANYCABLE_VERSION", "~> 1.0")
+gem "anycable-rails", github: "anycable/anycable-rails"
 
 group :development, :test do
   gem "pry"

graphql-anycable.gemspec

@@ -31,8 +31,11 @@ Gem::Specification.new do |spec|
   spec.add_dependency "graphql",       "~> 1.11"
   spec.add_dependency "redis",         ">= 4.2.0"
 
+  spec.add_development_dependency "anycable-rails"
   spec.add_development_dependency "bundler", "~> 2.0"
   spec.add_development_dependency "fakeredis"
+  spec.add_development_dependency "rack"
+  spec.add_development_dependency "railties"
   spec.add_development_dependency "rake", ">= 12.3.3"
   spec.add_development_dependency "rspec", "~> 3.0"
 end

lib/graphql/subscriptions/anycable_subscriptions.rb

@@ -225,6 +225,10 @@ def delete_legacy_subscription(subscription_id)
       def delete_channel_subscriptions(channel_or_id)
         # For backward compatibility
         channel_id = channel_or_id.is_a?(String) ? channel_or_id : read_subscription_id(channel_or_id)
+
+        # Missing in case disconnect happens before #execute
+        return unless channel_id
+
         redis.smembers(CHANNEL_PREFIX + channel_id).each do |subscription_id|
           delete_subscription(subscription_id)
         end
@@ -240,13 +244,30 @@ def anycable
       def read_subscription_id(channel)
         return channel.instance_variable_get(:@__sid__) if channel.instance_variable_defined?(:@__sid__)
 
-        channel.instance_variable_set(:@__sid__, channel.connection.socket.istate["sid"])
+        istate = fetch_channel_istate(channel)
+
+        return unless istate
+
+        channel.instance_variable_set(:@__sid__, istate["sid"])
       end
 
       def write_subscription_id(channel, val)
         channel.connection.socket.istate["sid"] = val
         channel.instance_variable_set(:@__sid__, val)
       end
+
+      def fetch_channel_istate(channel)
+        # For Rails integration
+        return channel.__istate__ if channel.respond_to?(:__istate__)
+
+        return unless channel.connection.socket.istate
+
+        if channel.connection.socket.istate[channel.identifier]
+          JSON.parse(channel.connection.socket.istate[channel.identifier])
+        else
+          channel.connection.socket.istate
+        end
+      end
     end
   end
 end

spec/graphql/anycable_spec.rb

@@ -90,6 +90,10 @@
   end
 
   describe ".delete_channel_subscriptions" do
+    before do
+      GraphQL::AnyCable.config.use_client_provided_uniq_id = false
+    end
+
     before do
       AnycableSchema.execute(
         query: query,
@@ -99,6 +103,10 @@
       )
     end
 
+    after do
+      GraphQL::AnyCable.config.use_client_provided_uniq_id = false
+    end
+
     let(:redis) { AnycableSchema.subscriptions.redis }
 
     subject do

spec/integration_helper.rb

@@ -0,0 +1,104 @@
+# frozen_string_literal: true
+
+require "anycable/rspec"
+require "rack"
+
+RSpec.shared_context "rpc" do
+  include_context "anycable:rpc:command"
+
+  let(:user) { "john" }
+  let(:schema) { nil }
+  let(:identifiers) { {current_user: "john", schema: schema.to_s} }
+  let(:channel_class) { "GraphqlChannel" }
+  let(:channel_params) { {channelId: rand(1000).to_s} }
+  let(:channel_identifier) { {channel: channel_class}.merge(channel_params) }
+  let(:channel_id) { channel_identifier.to_json }
+
+  let(:handler) { AnyCable::RPC::Handler.new }
+end
+
+# Minimal AnyCable connection implementation
+class FakeConnection
+  class Channel
+    attr_reader :connection, :params, :identifier
+
+    def initialize(connection, identifier, params)
+      @connection = connection
+      @identifier = identifier
+      @params = params
+    end
+
+    def stream_from(broadcasting)
+      connection.socket.subscribe identifier, broadcasting
+    end
+  end
+
+  attr_reader :request, :socket, :identifiers, :subscriptions,
+              :schema
+
+  def initialize(socket, identifiers: nil, subscriptions: nil)
+    @socket = socket
+    @identifiers = identifiers ? JSON.parse(identifiers) : {}
+    @request = Rack::Request.new(socket.env)
+    @schema = Object.const_get(@identifiers["schema"])
+    @subscriptions = subscriptions
+  end
+
+  def handle_channel_command(identifier, command, data)
+    parsed_id = JSON.parse(identifier)
+
+    parsed_id.delete("channel")
+    channel = Channel.new(self, identifier, parsed_id)
+
+    res =
+      case command
+      when "message"
+        data = JSON.parse(data)
+        result = 
+          schema.execute(
+            query: data["query"],
+            context: identifiers.merge(channel: channel),
+            variables: Hash(data["variables"]),
+            operation_name: data["operationName"],
+          )
+
+        transmit(
+          result: result.subscription? ? { data: nil } : result.to_h,
+          more: result.subscription?,
+        )
+      when "unsubscribe"
+        schema.subscriptions.delete_channel_subscriptions(channel)
+        true
+      else
+        raise "Unknown command"
+      end
+    res
+  end
+
+  def transmit(data)
+    socket.transmit data.to_json
+  end
+
+  def identifiers_json
+    @identifiers.to_json
+  end
+
+  def close
+    socket.close
+  end 
+end
+
+AnyCable.connection_factory = ->(socket, **options) { FakeConnection.new(socket, **options) }
+
+# Add verbose logging to exceptions
+AnyCable.capture_exception do |ex, method, message|
+  $stdout.puts "RPC ##{method} failed: #{message}\n#{ex}\n#{ex.backtrace.take(5).join("\n")}"
+end
+
+RSpec.configure do |config|
+  config.define_derived_metadata(file_path: %r{spec/integrations/}) do |metadata|
+    metadata[:integration] = true
+  end
+
+  config.include_context "rpc", integration: true
+end