rxpy observablesobservers and subjects the following example creates subject that enriches the data it receives by adding timestamp to each data item it then republishes the data item to any observers that have subscribed to it import rx from rx subjects import subject from datetime import datetime source rx from_list([ ]class timestampsubject(subject)def on_next(selfvalue)print('subject received'valuesuper(on_next((valuedatetime now())def on_completed(self)print('data stream completed'super(on_completed(def on_error(selferror)print('in subject error occurred'errorsuper(on_error(errordef prime_number_reporter(value)print('function received'valueprint('set up'create the subject subject timestampsubject(set up multiple subscribers for the subject subject subscribe(prime_number_reportersubject subscribe(lambda valueprint('lambda received'value)subject subscribeon_next lambda valueprint('received on_next'value)on_error lambda expprint('error occurred'exp)on_completed lambdaprint('received completed notification'subscribe the subject to the observable source source subscribe(subjectprint('done'note that in the above program the observers are added to the subject before the subject is added to the source observable this ensures that the observers are subscribed before the subject starts to receive data published by the