El siguiente ejemplo crea un Asunto que enriquece los datos que recibe al agregar una marca de tiempo a cada elemento de datos. A continuación, vuelve a publicar el elemento de datos en cualquier Los observadores que se han suscrito a ella. importar rx de rx.subjects asunto de importación desde fechahora fechahora de importación fuente = rx.from_list([2, 3, 5, 7]) clase TimeStampSubject (Asunto): def on_next(self, valor): print(‘Asunto recibido’, valor) super().on_next((valor, fechahora.ahora())) def on_completed(self): print(‘Flujo de datos completado’) super().on_completed() def on_error(self, error): print(‘En Asunto - Ocurrió un error’, error) super().on_error(error) def prime_number_reporter(valor): print(‘Función Recibida’, valor) Configuración de impresión’)

Crear el Asunto

asunto = TimeStampSubject()

Configure múltiples suscriptores para el tema

asunto.subscribe(principal_numero_reportero) asunto.subscribe(valor lambda: print(‘Lambda recibido’, valor)) asunto.subscribe( en_siguiente = valor lambda: print(‘Recibido en_siguiente’, valor), on_error = lambda exp: print(‘Ocurrió un error’, exp), on_completed = lambda: print(‘Recibido completado notificación’) )

Suscribir el Sujeto a la fuente Observable

source.subscribe(asunto) imprimir(‘Terminado’) Tenga en cuenta que en el programa anterior, los Observadores se agregan al Sujeto antes el Asunto se agrega a la fuente Observable. Esto asegura que los observadores están suscritos antes de que el Sujeto comience a recibir datos publicados por el 434 36 RxPy Observables, Observadores y Sujetos