class Rx::AutoDetachObserver

Public Class Methods

new(observer) click to toggle source
Calls superclass method Rx::ObserverBase::new
# File lib/rx/core/auto_detach_observer.rb, line 36
def initialize(observer)
  @observer = observer
  @m = SingleAssignmentSubscription.new

  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

Public Instance Methods

on_completed_core() click to toggle source
# File lib/rx/core/auto_detach_observer.rb, line 28
def on_completed_core
  begin
    @observer.on_completed
  ensure
    unsubscribe
  end
end
on_error_core(error) click to toggle source
# File lib/rx/core/auto_detach_observer.rb, line 20
def on_error_core(error)
  begin
    @observer.on_error(error)
  ensure
    unsubscribe
  end
end
on_next_core(value) click to toggle source
# File lib/rx/core/auto_detach_observer.rb, line 10
def on_next_core(value) 
  no_error = false
  begin
    @observer.on_next(value)
    no_error = true
  ensure
    unsubscribe unless no_error
  end
end
subscription=(new_subscription) click to toggle source
# File lib/rx/core/auto_detach_observer.rb, line 48
def subscription=(new_subscription)
  @m.subscription = new_subscription
end
unsubscribe() click to toggle source
Calls superclass method Rx::ObserverBase#unsubscribe
# File lib/rx/core/auto_detach_observer.rb, line 52
def unsubscribe
  super
  @m.unsubscribe
end