Class: RxRuby::AsyncLockObserver

Inherits:
ObserverBase show all
Defined in:
lib/rx_ruby/core/async_lock_observer.rb

Instance Method Summary collapse

Methods inherited from ObserverBase

#dispose, #fail, #on_completed, #on_error, #on_next, #unsubscribe

Methods included from Observer

allow_reentrancy, #as_observer, #checked, configure, create, from_notifier, #notify_on, prevent_reentrancy, #to_notifier

Constructor Details

#initialize(observer, gate) ⇒ AsyncLockObserver

Returns a new instance of AsyncLockObserver.



33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/rx_ruby/core/async_lock_observer.rb', line 33

def initialize(observer, gate)
  @observer = observer
  @gate = gate

  config = ObserverConfiguration.new
  config.on_next(&method(:on_next_core))
  config.on_error(&method(:on_error_core))
  config.on_completed(&method(:on_completed_core))

  super(config)
end

Instance Method Details

#on_completed_core ⇒ Object 



29
30
31
 
# File 'lib/rx_ruby/core/async_lock_observer.rb', line 29

def on_completed_core
  @gate.wait { @observer.on_completed }
end

#on_error_core(error) ⇒ Object 



25
26
27
 
# File 'lib/rx_ruby/core/async_lock_observer.rb', line 25

def on_error_core(error)
  @gate.wait { @observer.on_error error }
end

#on_next_core(value) ⇒ Object 



21
22
23
 
# File 'lib/rx_ruby/core/async_lock_observer.rb', line 21

def on_next_core(value)
  @gate.wait { @observer.on_next value }
end