RxPY - Lavorare con l'oggetto

Un soggetto è una sequenza osservabile, così come un osservatore che può trasmettere in multicast, cioè parlare con molti osservatori che si sono iscritti.

Discuteremo i seguenti argomenti sull'argomento:

• Crea un soggetto
• Iscriviti a un argomento
• Passaggio dei dati al soggetto
• BehaviorSubject
• ReplaySubject
• AsyncSubject

Crea un soggetto

Per lavorare con un oggetto, dobbiamo importare l'oggetto come mostrato di seguito:

from rx.subject import Subject

Puoi creare un oggetto-soggetto come segue:

subject_test = Subject()

L'oggetto è un osservatore che ha tre metodi:

• on_next(value)
• on_error (errore) e
• on_completed()

Iscriviti a un argomento

È possibile creare più abbonamenti sull'argomento come mostrato di seguito:

subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)

Passaggio di dati al soggetto

È possibile passare i dati all'oggetto creato utilizzando il metodo on_next (valore) come mostrato di seguito -

subject_test.on_next("A")
subject_test.on_next("B")

I dati verranno passati a tutti gli iscritti, aggiunti sull'argomento.

Ecco un esempio funzionante dell'argomento.

Esempio

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.on_next("A")
subject_test.on_next("B")

L'oggetto subject_test viene creato chiamando un Subject (). L'oggetto subject_test fa riferimento ai metodi on_next (value), on_error (error) e on_completed (). L'output dell'esempio precedente è mostrato di seguito:

Produzione

E:\pyrx>python testrx.py
The value is A
The value is A
The value is B
The value is B

Possiamo usare il metodo on_completed (), per interrompere l'esecuzione del soggetto come mostrato di seguito.

Esempio

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.subscribe(
   lambda x: print("The value is {0}".format(x))
)
subject_test.on_next("A")
subject_test.on_completed()
subject_test.on_next("B")

Una volta chiamato complete, il metodo successivo chiamato in seguito non viene richiamato.

Produzione

E:\pyrx>python testrx.py
The value is A
The value is A

Vediamo ora come chiamare il metodo on_error (error).

Esempio

from rx.subject import Subject
subject_test = Subject()
subject_test.subscribe(
   on_error = lambda e: print("Error : {0}".format(e))
)
subject_test.subscribe(
   on_error = lambda e: print("Error : {0}".format(e))
)
subject_test.on_error(Exception('There is an Error!'))

Produzione

E:\pyrx>python testrx.py
Error: There is an Error!
Error: There is an Error!

BehaviorSubject

BehaviorSubject ti darà il valore più recente quando viene chiamato. Puoi creare un soggetto comportamentale come mostrato di seguito:

from rx.subject import BehaviorSubject
behavior_subject = BehaviorSubject("Testing Behaviour Subject"); // initialized the behaviour subject with value:Testing Behaviour Subject

Ecco un esempio funzionante per utilizzare Behavior Subject

Esempio

from rx.subject import BehaviorSubject
behavior_subject = BehaviorSubject("Testing Behaviour Subject");
behavior_subject.subscribe(
   lambda x: print("Observer A : {0}".format(x))
)
behavior_subject.on_next("Hello")
behavior_subject.subscribe(
   lambda x: print("Observer B : {0}".format(x))
)
behavior_subject.on_next("Last call to Behaviour Subject")

Produzione

E:\pyrx>python testrx.py
Observer A : Testing Behaviour Subject
Observer A : Hello
Observer B : Hello
Observer A : Last call to Behaviour Subject
Observer B : Last call to Behaviour Subject

Ripeti oggetto

Un replaysubject è simile al behavior subject, in cui può bufferizzare i valori e riprodurre lo stesso ai nuovi abbonati. Ecco un esempio funzionante di soggetto replay.

Esempio

from rx.subject import ReplaySubject
replay_subject = ReplaySubject(2)
replay_subject.subscribe(lambda x: print("Testing Replay Subject A: {0}".format(x)))
replay_subject.on_next(1)
replay_subject.on_next(2)
replay_subject.on_next(3)
replay_subject.subscribe(lambda x: print("Testing Replay Subject B: {0}".format(x)));
replay_subject.on_next(5)

Il valore del buffer utilizzato è 2 sull'oggetto della riproduzione. Quindi, gli ultimi due valori verranno bufferizzati e utilizzati per i nuovi abbonati chiamati.

Produzione

E:\pyrx>python testrx.py
Testing Replay Subject A: 1
Testing Replay Subject A: 2
Testing Replay Subject A: 3
Testing Replay Subject B: 2
Testing Replay Subject B: 3
Testing Replay Subject A: 5
Testing Replay Subject B: 5

AsyncSubject

Nel caso di AsyncSubject, l'ultimo valore chiamato viene passato al sottoscrittore e verrà eseguito solo dopo che il metodo complete () è stato chiamato.

Esempio

from rx.subject import AsyncSubject
async_subject = AsyncSubject()
async_subject.subscribe(lambda x: print("Testing Async Subject A: {0}".format(x)))
async_subject.on_next(1)
async_subject.on_next(2)
async_subject.on_completed()
async_subject.subscribe(lambda x: print("Testing Async Subject B: {0}".format(x)))
Here, before complete is called, the last value passed to the subject is 2, and the same is given to the subscribers.

Produzione

E:\pyrx>python testrx.py
Testing Async Subject A: 2
Testing Async Subject B: 2