Module: GraphQL::AnyCable::Cleaner

Extended by:
Cleaner
Included in:
Cleaner
Defined in:
lib/graphql/anycable/cleaner.rb

Instance Method Summary collapse

Instance Method Details

#clean ⇒ Object 



8
9
10
11
12
13
 
# File 'lib/graphql/anycable/cleaner.rb', line 8

def clean
  clean_channels
  clean_subscriptions
  clean_fingerprint_subscriptions
  clean_topic_fingerprints
end

#clean_channels ⇒ Object 



15
16
17
18
19
20
 
# File 'lib/graphql/anycable/cleaner.rb', line 15

def clean_channels
  return unless config.subscription_expiration_seconds
  return unless config.use_redis_object_on_cleanup

  clean_idle_keys(adapter::CHANNEL_PREFIX)
end

#clean_fingerprint_subscriptions ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/graphql/anycable/cleaner.rb', line 29

def clean_fingerprint_subscriptions
  AnyCable.with_redis do |redis|
    each_key_batch(redis, adapter::SUBSCRIPTIONS_PREFIX) do |keys|
      bulk_readable, oversized = partition_by_size(redis, keys, :scard)

      each_capped_chunk(bulk_readable) do |chunk|
        members = redis.pipelined { |pipeline| chunk.each { |key| pipeline.smembers(key) } }
        remove_stale(redis, adapter::SUBSCRIPTION_PREFIX, chunk.zip(members), :srem)
      end

      oversized.each do |key|
        each_batch(redis.sscan_each(key, count: redis_scan_count)) do |subscription_ids|
          remove_stale(redis, adapter::SUBSCRIPTION_PREFIX, [[key, subscription_ids]], :srem)
        end
      end
    end
  end
end

#clean_subscriptions ⇒ Object 



22
23
24
25
26
27
 
# File 'lib/graphql/anycable/cleaner.rb', line 22

def clean_subscriptions
  return unless config.subscription_expiration_seconds
  return unless config.use_redis_object_on_cleanup

  clean_idle_keys(adapter::SUBSCRIPTION_PREFIX)
end

#clean_topic_fingerprints ⇒ Object 



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/graphql/anycable/cleaner.rb', line 48

def clean_topic_fingerprints
  AnyCable.with_redis do |redis|
    each_key_batch(redis, adapter::FINGERPRINTS_PREFIX) do |keys|
      redis.pipelined { |pipeline| keys.each { |key| pipeline.zremrangebyscore(key, "-inf", "0") } }

      bulk_readable, oversized = partition_by_size(redis, keys, :zcard)

      each_capped_chunk(bulk_readable) do |chunk|
        fingerprints = redis.pipelined { |pipeline| chunk.each { |key| pipeline.zrange(key, 0, -1) } }
        remove_stale(redis, adapter::SUBSCRIPTIONS_PREFIX, chunk.zip(fingerprints), :zrem)
      end

      oversized.each do |key|
        each_batch(redis.zscan_each(key, count: redis_scan_count)) do |members|
          remove_stale(redis, adapter::SUBSCRIPTIONS_PREFIX, [[key, members.map(&:first)]], :zrem)
        end
      end
    end
  end
end