Class: MTProto::Updates::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/mtproto/updates/runner.rb

Overview

Drives the high-level update loop over a Client (Client#run_updates is the entry point). The on_update callback runs on the receiver fiber and only ENQUEUES raw containers; a single worker fiber drains the queue and feeds the state machine, so recovery RPCs never block the receiver (no deadlock) and never race each other on pts (serial by construction). A reconnect enqueues a re-sync. An optional on_start runs once inside the started mainloop before the first update, the point at which host startup that needs the API belongs.

Instance Method Summary collapse

Constructor Details

#initialize(client:, store:, logger: nil) ⇒ Runner

Returns a new instance of Runner.



16
17
18
19
20
21
 
# File 'lib/mtproto/updates/runner.rb', line 16

def initialize(client:, store:, logger: nil)
  @client = client
  @store = store
  @logger = logger
  @machine = StateMachine.new(client: client, store: store, logger: logger)
end

Instance Method Details

#run(on_start: nil, &block) ⇒ Object

Raises:

  • (ArgumentError) 


23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/mtproto/updates/runner.rb', line 23

def run(on_start: nil, &block)
  raise ArgumentError, 'block is required' unless block

  @client.run_mainloop do
    queue = Async::Queue.new
    @client.replace_update_callback { |constructor, body| queue.enqueue([constructor, body]) }
    @client.on_reconnect { queue.enqueue(:reconnect) }
    on_start&.call
    @machine.resume(&block)
    drain(queue, &block)
  end
end