RxPY - Operatori collegabili

pubblicare

Questo metodo convertirà l'osservabile in un osservabile collegabile.

Sintassi

publish(mapper=None)

Parametri

mappatore: opzionale. Una funzione utilizzata per trasmettere più volte i valori di origine, senza dover eseguire più sottoscrizioni.

Esempio

from rx import create, range, operators as op
import random
def test_observable(observer, scheduler):
   observer.on_next(random.random())
   observer.on_completed()
source = create(test_observable).pipe(op.publish())
test1 = source.subscribe(on_next = lambda i: print("From subscriber 1 - {0}".format(i)))
test2 = source.subscribe(on_next = lambda i: print("From subscriber 2 –
{0}".format(i)))
source.connect()

Produzione

E:\pyrx>python testrx.py
From subscriber 1 - 0.14751607273318490
From subscriber 2 - 0.1475160727331849

ref_count

Questo operatore renderà l'osservabile una normale osservabile.

Sintassi

ref_count()

Esempio

from rx import create, operators as op
import random
def test_observable(observer, scheduler):
   observer.on_next(random.random())
source = create(test_observable).pipe(op.publish(),op.ref_count())
test1 = source.subscribe(on_next = lambda i: print("From subscriber 1 - {0}".format(i)))
test2 = source.subscribe(on_next = lambda i: print("From subscriber 2 - {0}".format(i)))

Produzione

E:\pyrx>python testrx.py
From subscriber 1 - 0.8230640432381131

replay

Questo metodo funziona in modo simile a replaySubject. Questo metodo restituirà gli stessi valori, anche se l'osservabile ha già emesso e alcuni degli abbonati sono in ritardo nella sottoscrizione.

Sintassi

replay()

Esempio

from rx import create, range, operators as op
import random
from threading import Timer
def test_observable(observer, scheduler):
   observer.on_next(random.random())
   observer.on_completed()
source = create(test_observable).pipe(op.replay())
test1 = source.subscribe(on_next = lambda i: print("From subscriber 1 - {0}".format(i)))
test2 = source.subscribe(on_next = lambda i: print("From subscriber 2 - {0}".format(i)))
source.connect()
print("subscriber called after delay ")
def last_subscriber():
   test3 = source.subscribe(on_next = lambda i: print("From subscriber 3 - {0}".format(i)))
t = Timer(5.0, last_subscriber)
t.start()

Produzione

E:\pyrx>python testrx.py
From subscriber 1 - 0.8340998157725388
From subscriber 2 - 0.8340998157725388
subscriber called after delay
From subscriber 3 - 0.8340998157725388