RxPY - Konu ile Çalışma

Bir özne, gözlemlenebilir bir dizinin yanı sıra, çok noktaya yayın yapabilen bir gözlemcidir, yani abone olan birçok gözlemciyle konuşur.

Konuyla ilgili aşağıdaki konuları tartışacağız -

• Bir konu oluşturun
• Bir konuya abone ol
• Verilerin konuya aktarılması
• BehaviorSubject
• ReplaySubject
• AsyncSubject

Bir konu oluşturun

Bir konuyla çalışmak için, Konuyu aşağıda gösterildiği gibi içe aktarmamız gerekiyor -

from rx.subject import Subject

Aşağıdaki gibi bir konu nesnesi oluşturabilirsiniz -

subject_test = Subject()

Nesne, üç yöntemi olan bir gözlemcidir -

• on_next(value)
• on_error (hata) ve
• on_completed()

Bir Konuya Abone Olun

Aşağıda gösterildiği gibi konu ile ilgili birden fazla abonelik oluşturabilirsiniz -

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

Verileri Konuya Aktarma

Verileri, aşağıda gösterildiği gibi on_next (değer) yöntemini kullanarak oluşturulan konuya aktarabilirsiniz -

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

Veriler konuya eklenen tüm aboneliğe aktarılacaktır.

İşte konunun çalışan bir örneği.

Misal

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")

Subject_test nesnesi bir Subject () çağırılarak oluşturulur. Subject_test nesnesinin on_next (değer), on_error (hata) ve on_completed () yöntemlerine başvurusu vardır. Yukarıdaki örneğin çıktısı aşağıda gösterilmiştir -

Çıktı

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

Aşağıda gösterildiği gibi konu yürütmeyi durdurmak için on_completed () yöntemini kullanabiliriz.

Misal

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")

Complete'i çağırdığımızda, daha sonra çağrılan bir sonraki yöntem çağrılmaz.

Çıktı

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

Şimdi on_error (hata) yönteminin nasıl çağrılacağını görelim.

Misal

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!'))

Çıktı

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

Davranış Konu

BehaviorSubject, arandığında size en son değeri verecektir. Aşağıda gösterildiği gibi davranış konusu oluşturabilirsiniz -

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

İşte, Davranış Konusu kullanmak için çalışan bir örnek

Misal

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")

Çıktı

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

Tekrar Oynatma Konusu

Bir replaysubject, davranış öznesine benzer, burada değerleri tamponlayabilir ve aynısını yeni abonelere yeniden oynatabilir. İşte tekrar konusunun çalışan bir örneği.

Misal

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)

Yeniden oynatma konusunda kullanılan tampon değeri 2'dir. Böylece son iki değer arabelleğe alınacak ve aranan yeni aboneler için kullanılacaktır.

Çıktı

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

AsyncSubject durumunda, çağrılan son değer aboneye iletilir ve yalnızca complete () yöntemi çağrıldıktan sonra yapılır.

Misal

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.

Çıktı

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