#NEWLINE# def activate( self , external_subscriptions , observer , deactivate ) : #NEWLINE# #INDENT# join_observers = [ ] #NEWLINE# for pattern in self . expression . patterns : #NEWLINE# #INDENT# join_observers . append ( self . plan_create_observer ( external_subscriptions , pattern , observer . on_error ) ) #NEWLINE# #UNINDENT# def on_next( * args ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# result = self . selector ( * args ) #NEWLINE# #UNINDENT# except Exception as e : #NEWLINE# #INDENT# observer . on_error ( e ) #NEWLINE# return #NEWLINE# #UNINDENT# observer . on_next ( result ) #NEWLINE# #UNINDENT# def on_completed( ) : #NEWLINE# #INDENT# for join_observer in join_observers : #NEWLINE# #INDENT# join_observer . remove_active_plan ( active_plan ) #NEWLINE# #UNINDENT# deactivate ( active_plan ) #NEWLINE# #UNINDENT# active_plan = ActivePlan ( join_observers , on_next , active_plan ) #NEWLINE# for join_observer in join_observers : #NEWLINE# #INDENT# join_observer . add_active_plan ( active_plan ) #NEWLINE# #UNINDENT# return active_plan