Class: RedisClient::Cluster::PubSub

Inherits:
Object
  • Object
show all
Defined in:
lib/redis_client/cluster/pub_sub.rb

Defined Under Namespace

Classes: State

Constant Summary collapse

BUF_SIZE = 
Integer(ENV.fetch('REDIS_CLIENT_PUBSUB_BUF_SIZE', 1024))

Instance Method Summary collapse

Constructor Details

#initialize(router, command_builder) ⇒ PubSub

Returns a new instance of PubSub.



46
47
48
49
50
51
 
# File 'lib/redis_client/cluster/pub_sub.rb', line 46

def initialize(router, command_builder)
  @router = router
  @command_builder = command_builder
  @queue = SizedQueue.new(BUF_SIZE)
  @state_dict = {}
end

Instance Method Details

#call(*args, **kwargs) ⇒ Object 



53
54
55
 
# File 'lib/redis_client/cluster/pub_sub.rb', line 53

def call(*args, **kwargs)
  _call(@command_builder.generate(args, kwargs))
end

#call_v(command) ⇒ Object 



57
58
59
 
# File 'lib/redis_client/cluster/pub_sub.rb', line 57

def call_v(command)
  _call(@command_builder.generate(command))
end

#close ⇒ Object 



61
62
63
64
65
66
67
 
# File 'lib/redis_client/cluster/pub_sub.rb', line 61

def close
  @state_dict.each_value(&:close)
  @state_dict.clear
  @queue.clear
  @queue.close
  nil
end

#next_event(timeout = nil) ⇒ Object 



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/redis_client/cluster/pub_sub.rb', line 69

def next_event(timeout = nil)
  @state_dict.each_value(&:ensure_worker)
  max_duration = calc_max_duration(timeout)
  starting = obtain_current_time

  loop do
    break if max_duration > 0 && obtain_current_time - starting > max_duration

    case event = @queue.pop(true)
    when StandardError then raise event
    when Array then break event
    end
  rescue ThreadError
    sleep 0.005
  end
end