#NEWLINE# @ extensionmethod ( Observable ) #NEWLINE# def transduce( self , transducer ) : #NEWLINE# #INDENT# 'Execute a transducer to transform the observable sequence.\n\n    Keyword arguments:\n    :param Transducer transducer: A transducer to execute.\n\n    :returns: An Observable sequence containing the results from the\n        transducer.\n    :rtype: Observable\n    ' #NEWLINE# source = self #NEWLINE# def subscribe( observer ) : #NEWLINE# #INDENT# xform = transducer ( Observing ( observer ) ) #NEWLINE# def on_next( v ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# xform . step ( observer , v ) #NEWLINE# #UNINDENT# except Exception as e : #NEWLINE# #INDENT# observer . on_error ( e ) #NEWLINE# #UNINDENT# #UNINDENT# def on_completed( ) : #NEWLINE# #INDENT# xform . complete ( observer ) #NEWLINE# #UNINDENT# return source . subscribe ( on_next , observer . on_error , on_completed ) #NEWLINE# #UNINDENT# return AnonymousObservable ( source )