Unirx/R3 İle Reaktif Programlama
Reaktif Programlama, veri akışları oluşturma ve bu akışlara abone olarak değişikliklere tepki vermeye dayalı bir programlama paradigmasıdır.
Bu yazıda, bu paradigma üzerine kurulu olan DotnetRx/UniRx kütüphanesinin daha modern versiyonu olan R3 kütüphanesinin bazı elementlerine bakış atacağız. Ek olarak R3'ün Unity’ye özgü bazı kısımlarını da dahil edeceğiz.
Bu konuda temel olarak birkaç başlık var.
Observable bir olay dizisini (veri akışı) temsil eder.
Observer’lar Observable’lara abone olur ve yaydıkları değerlere tepki verir. Operatörler veri akışları oluşturmak,filtrelemek, birleştirmek vb işler için kullanılır.
Bunlara bir örnekte bakalım.
var observable = Observable.Create<int>(observer =>
{
observer.OnNext(1);
observer.OnNext(2);
observer.OnNext(3);
observer.OnErrorResume(new Exception("Custom Exception"));
observer.OnCompleted();
return Disposable.Empty;
});
observable.Subscribe
(onNext => Debug.Log(onNext),
onError => Debug.LogError(onError),
(_) => Debug.Log("OnCompleted"));
//1
//2
//3
//Custom Exception
//OnCompletedBu örnekte, Create operatörü ile int türünde veri akışı oluşturan bir Observable oluşturduk.
Daha sonra Observer'ın OnNext, OnErrorResume ve OnCompleted metotlarını kullanarak akışı düzenledik.
OnNext⇒ İstenen veriyi akışa yollar.
OnErrorResume⇒ Akışa bir hata verisi yollar.
OnCompleted⇒ Akışın tamamlanlandığı olayını tetikler.
Daha sonra observable’a Subscribe olarak bu 3 event’i dinlemeye başladık ve gelen veriyi logladık.
Bunlara ek olarak sık kullanılan bir kaç türe daha bakalım.
Subject
OnNext metodu ile yeni değerler yayınlarken Subscribe metodu ile bu değerleri dinleyebildiğimiz özel bir türdür. Yani hem observable hem observer olarak davranabilir. Create operatörüne göre çok daha esnek bir kullanım sağlar.
var subject = new Subject<int>();
subject.Subscribe
(onNext => Debug.Log(onNext),
onError => Debug.Log(onError),
(_) => Debug.Log("OnCompleted"));
subject.OnNext(1);
subject.OnNext(2);
subject.OnNext(3);
subject.OnErrorResume(new Exception("Custom Exception"));
subject.OnCompleted();
//1
//2
//3
//Custom Exception
//OnCompleted
ReactiveProperty
ReactiveProperty bir akışın hem son değerini tutarken hem de bu akışa klasik şekilde abone olabildiğimiz bir observable tipidir. Burada dikkat edilmesi gereken nokta bu akış sadece değer değişimlerinde tetiklenir yani üst üste aynı değer gönderilmesi durumunda aynı değerler tekrar akışa gönderilmez.
var reactiveProperty = new ReactiveProperty<int>(0);
reactiveProperty.Subscribe(onNext => Debug.Log(onNext));
reactiveProperty.Value = 1;
reactiveProperty.Value = 1;
reactiveProperty.Value = 2;
Debug.Log($"Current value {reactiveProperty.Value}");
//0
//1
//2
//Current value 2
Operators
A) Creating
Create
Yeni bir observable yaratmak için kullanılır. Daha sonra observer parametresi üzerinden akışın içeriği tasarlanabilir.
var observable = Observable.Create<string>(observer =>
{
observer.OnNext("Hello");
observer.OnNext("World");
observer.OnCompleted();
return Disposable.Empty;
});
observable.Subscribe(onNext => Debug.Log(onNext));
//Hello
//World
Timer
Belirlenen süre sonunda 1 akış yapan observable yaratır. Belirli bir sürenin sonunda işlem yaptırmak için kullanışlıdır.
var timerObservable = Observable.Timer(TimeSpan.FromSeconds(1));
timerObservable.Subscribe(_ =>
{
Debug.Log("After 1 second");
gameObject.SetActive(false);
});
//After 1 second
Interval
Tekrarlı olarak verilen süre sonunda akış yapan bir observable yaratır.
var intervalObservable = Observable.Interval(TimeSpan.FromSeconds(1));
intervalObservable.Subscribe(_ => Debug.Log($"Every second"));
//Every second
//Every second
...
IntervalFrame
Tekrarlı olarak belirtilen frame sayısı ve frame tipi sonunda akış yapan bir observable yaratır.
var intervalFrameObservable = Observable.IntervalFrame(30, UnityFrameProvider.FixedUpdate);
intervalFrameObservable.Subscribe(_ => Debug.Log($"Every 30 fixed frames"));
//Every 30 fixed frames
//Every 30 fixed frames
...
EveryUpdate
Belirlenen frame tipine göre her frame’de tetiklenen bir observable yaratır. Örneğin burada MonoBehaviour.Update işlevinde bir observable yaratılıp kullanılmıştır.
var everyUpdateObservable = Observable.EveryUpdate(UnityFrameProvider.Update);
everyUpdateObservable.Subscribe(_ => Debug.Log($"Every frame"));
//Every frame
//Every frame
B) Transforming
Select
Akıştaki öğeleri alıp istenen fonksiyon uygulanarak yeni bir akış üretmeye yarar.
Bu örnekte her framede stream yapan bir observable yaratılır. Daha sonra bu observable’ın akış tipi transform.position (Vector3)’a dönüştürülür ve subscribe edildiği yerde bu Vector3 verileri loglanır.
var everyUpdateObservable = Observable.EveryUpdate();
var positionObservable = everyUpdateObservable
.Select(_ => transform.position);
positionObservable.Subscribe(position => Debug.Log($"Position: {position}"));
SelectMany
Select gibi akıştaki öğeleri alıp istenen fonksiyon uygulanarak yeni bir akış üretmeye yarar. Fakat Select’den ayrılan birkaç önemli farkı var.
*Dönüş tipi olarak sadece Observable<T> alabilir.
*Döndürülen tip bir Enumerable ise üretilen Enumerable’ları tek bir dizide birleştirir.
Akış SelectMany ile verdiğimiz Observable üzerinden akmaya devam eder.
Select ile verdiğimiz durumda ise yeni observable akışına abone olmak yerine direkt olarak observable’ı döndürür.
void Main()
{
var observable = Observable.Create<int>(observer =>
{
observer.OnNext(1);
observer.OnNext(2);
return Disposable.Empty;
});
var select = observable.Select(ReturnDoubleAfterDelay);
var selectMany = observable.SelectMany(ReturnDoubleAfterDelay);
select.Subscribe(onNext => Debug.Log($"Select: {onNext}"));
selectMany.Subscribe(onNext => Debug.Log($"SelectMany: {onNext}"));
}
private Observable<int> ReturnDoubleAfterDelay(int value)
{
return Observable.Return<int>(value * 2).Delay(TimeSpan.FromSeconds(1));
}
//Select: R3.Delay`1[System.Int32]
//Select: R3.Delay`1[System.Int32]
//SelectMany: 2
//SelectMany: 4
Select operatörü ile Enumerable döndürdüğümüzde yaydığı değer IEnumerable<int> ‘ın kendisi olurken SelectMany kullandığımızda buradaki IEnumerable<int> dizileri tek bir dizide birleştirilir ve bize bu birleştirilmiş dizinin içindeki itemlar’ı döndürür.
var observable = Observable.Create<int>(observer =>
{
observer.OnNext(1);
observer.OnNext(2);
return Disposable.Empty;
});
var selectObservable = observable.Select(value => Enumerable.Range(1, value));
var selectManyObservable = observable.SelectMany(value => Enumerable.Range(1, value).ToObservable());
selectObservable.Subscribe(onNext: value => Debug.Log(value));
selectManyObservable.Subscribe(onNext: value => Debug.Log(value));
//System.Linq.Enumerable+RangeIterator
//System.Linq.Enumerable+RangeIterator
//1
//1
//2
Chunk/Buffer
Bir observable dizisini belirli boyutta alt dizilere ayırmak için kullanılır.
Bu örnekte Chunk operatörü ile observable akışı 2li gruplar halinde dizilere ayrılmıştır.
var observable = Observable.Create<int>(observer =>
{
observer.OnNext(1);
observer.OnNext(2);
observer.OnNext(3);
observer.OnNext(4);
return Disposable.Empty;
});
var chunk = observable.Chunk(2);
chunk.Subscribe(value => Debug.Log(value));
//int[] 1,2
//int[] 2,3Ayrıca parametre olarak TimeSpan alabilmektedir. Bu sayede akıştaki veriler verilen süreler doldukça gruplandırılabilir.
Bu örnekte Chunk içine TimeSpan vererek pozisyon verileri 0.1saniyelik gruplar halinde Vector3[] olarak bölünmekte ve bu değerler dinlenmektedir.
var everyUpdate = Observable.EveryUpdate().Select(_ => transform.position);
var timeChunk = everyUpdate.Chunk(TimeSpan.FromSeconds(0.1f));
timeChunk.Subscribe(value => Debug.Log(value));
//Vector3[] ...
//Vector3[] ...
C) Combining
Merge
Get Enescan BEKTAŞ’s stories in your inbox
Join Medium for free to get updates from this writer.
Birden fazla observable akışını tek yerde birleştirir. Herhangi bir observable’dan akış gelmesi durumunda Merge observable’ı üzerinden yayılır.
var observable1 = Observable.Timer(TimeSpan.FromSeconds(1)).Select(_ => 1);
var observable2 = Observable.Timer(TimeSpan.FromSeconds(2)).Select(_ => 2);
var mergedObservable = observable1.Merge(observable2);
mergedObservable.Subscribe(onNext => Debug.Log(onNext));
//1
//2
CombineLatest
Birden fazla observable’ı birleştirerek tek çıktıda yaymaya yarar. Combine edilen observable’lardan herhangi birinden akış gelmesi durumunda diğer observable’ın son değeriyle birleştirilerek yayılır. ResultSelector ile bu birleştirme işleminin nasıl yapılacağı ayarlanır.
var observableA = Observable.Create<string>(x =>
{
x.OnNext("A1");
Observable.Timer(TimeSpan.FromSeconds(2)).Subscribe(_ => x.OnNext("A2"));
return Disposable.Empty;
});
var observableB = Observable.Create<string>(x =>
{
Observable.Timer(TimeSpan.FromSeconds(1)).Subscribe(_ => x.OnNext("B1"));
Observable.Timer(TimeSpan.FromSeconds(3)).Subscribe(_ => x.OnNext("B2"));
return Disposable.Empty;
});
observableA.CombineLatest(observableB, (s1, s2) => s1 + s2).Subscribe(Debug.Log);
//A1B1
//A2B1
//A2B2
ZipLatest
Verilen tüm observable’ların yeni değer üretmesini bekler ve daha sonra birleştirir. ResultSelector ile bu birleştirme işleminin nasıl yapılacağı ayarlanır.
var observableA = Observable.Create<string>(x =>
{
x.OnNext("A1");
Observable.Timer(TimeSpan.FromSeconds(2)).Subscribe(_ => x.OnNext("A2"));
return Disposable.Empty;
});
var observableB = Observable.Create<string>(x =>
{
Observable.Timer(TimeSpan.FromSeconds(1)).Subscribe(_ => x.OnNext("B1"));
Observable.Timer(TimeSpan.FromSeconds(3)).Subscribe(_ => x.OnNext("B2"));
return Disposable.Empty;
});
//A1B1
//A2B2
Zip
ZipLatest’dan farklı olarak akışlar sıralamayla birleştirilir.
ObservableA 1. akışı gönderdikten sonra ObservableB’nin de 1. akışı beklenir.
ObservableA ObservableB’den 1. akış gelene kadar yeni veri yollarsa bunlar sıraya konulup bekletilir. Bu sayede tüm observable’lar aynı akış sırasıyla birleştirilmiş olur.
var observableA = Observable.Create<string>(x =>
{
x.OnNext("A1");
x.OnNext("A2");
return Disposable.Empty;
});
var observableB = Observable.Create<string>(x =>
{
Observable.Timer(TimeSpan.FromSeconds(1)).Subscribe(_ => x.OnNext("B1"));
Observable.Timer(TimeSpan.FromSeconds(2)).Subscribe(_ => x.OnNext("B2"));
return Disposable.Empty;
});
observableA.Zip(observableB, (a, b) => new{ a,b}).Subscribe(x => Debug.Log(x.a + " " + x.b));
//A1B1
//A2B2
observableA.ZipLatest(observableB, (a, b) => new{ a,b}).Subscribe(x => Debug.Log(x.a + " " + x.b));
//A2B1
Switch
Switch operatörü kullanıldığında base Observable yerine Select ile verilen yeni Observable akışı dinlenir. Bu observable base observable’a bağlıdır. Yani base her yeni akış yaydığında önceki switch’ler iptal edilir ve bu sayede sadece son Observable akışına abone olunabilir. Bunları örneklerle inceyelim.
A üzerinden Select operatörü ile B observable’ı verilmiş ve daha sonra Switch yapılmıştır. Bu noktadan itibaren bu observable’a subscribe olduğumuzda A’dan herhangi bir akış yollasak da loglanmadığını sadece B akışının dinlendiğini görüyoruz.
var observableA = new Subject<int>();
var observableB = new Subject<int>();
var switchObservable = observableA.Select(_ => observableB.AsObservable()).Switch();
switchObservable.Subscribe(onNext => Debug.Log($"B OnNext {onNext}"));
observableA.OnNext(1);
//
observableB.OnNext(2);
// B OnNext 1
observableB.OnNext(2);
// B OnNext 2
Burada, ObservableA’ya 500ms bekledikten sonra akış gönderen bir ObservableB Switch ediliyor. 500ms içinde birden fazla ObservableA akışı gönderilse bile, sadece bir kez B OnNext log’u görüyoruz. Çünkü ana observable’ın (ObservableA) her yeni akışında, bir önceki abonelikler iptal ediliyor ve bu sayede eski Observable üzerinde çalışan observableB işlemleri de iptal oluyor.
Buna en güzel örnek bir arama çubuğunda verdiğimiz text’e göre anlık olarak arama yapılması düşünülebilir. Her text değişiminden sonra bir önceki isteğin sonucu artık önemsiz olacaktır. Bu yüzden her zaman sadece son isteğin sonucunu almak isteriz.
var observableA = new Subject<int>();
var observableB = Observable.ReturnUnit().Delay(TimeSpan.FromMilliseconds(500));
var switchObservable = observableA.Select(s => observableB).Switch();
switchObservable.Subscribe(onNext => Console.WriteLine($"B OnNext"));
observableA.OnNext(1); //
observableA.OnNext(1); //
observableA.OnNext(1); //B OnNext
await Task.Delay(600);
observableA.OnNext(1); //B OnNext
D) Filtering
Where
Where operatörü, bir Observable dizisindeki öğeleri belirli bir koşula göre filtrelemek için kullanılır. Bu operatör, koşulu karşılayan öğeleri yeni bir Observable dizisine ekler ve bu yeni diziyi döndürür.
Observable.Range(1, 10)
.Where(value => value % 2 == 0)
.Subscribe(onNext => Debug.Log(onNext));
//2
//4
//6
//8
//10
Skip
Skip operatörü, bir Observable dizisinden belirli sayıda öğeyi atlamak için kullanılır. Bu operatör, belirtilen sayıda öğeyi atladıktan sonra kalan tüm öğeleri yayınlar.
Observable.Range(1, 10)
.Skip(5)
.Subscribe(onNext => Debug.Log(onNext));
//6
//7
//8
//9
//10
Take
Take operatörü, bir Observable dizisinden belirli sayıda öğeyi almak için kullanılır. Bu operatör, belirtilen sayıda öğe yayınlandıktan sonra Observable’ı tamamlar.
Observable.Range(1, 10)
.Take(5)
.Subscribe(onNext => Debug.Log(onNext));
//1
//2
//3
//4
//5
Debounce/Throttle
Bir Observable’dan yalnızca belirli bir zaman aralığı boyunca başka bir veri yollanmazsa veriyi yayar. Aksi takdirde süre dolmadan yollanan her veride süre tekrardan saymaya başlar ve eski değerleri iptal eder.
Bu örnekte Debounce ile 1 saniyelik süre tanımlanmıştır. Observable içinde birbirleri arasında 1 saniye geçmeden 1,2,3 değerleri yollanmıştır. Bu nedenle bu değerler yayınlanmaz. Fakat 4 değeri yollandıktan 1 saniye geçene kadar yeni değer yollanmadığı için 4 akışa gönderilir ve devamında 7 değerinden sonrada başka bir değer gönderilmediği için akışa gönderilir.
var observable = Observable.Create<int>(observer =>
{
observer.OnNext(1);
observer.OnNext(2);
Observable.Timer(TimeSpan.FromSeconds(0.5f)).Subscribe(_ => observer.OnNext(3));
Observable.Timer(TimeSpan.FromSeconds(1f)).Subscribe(_ => observer.OnNext(4));
Observable.Timer(TimeSpan.FromSeconds(2.1f)).Subscribe(_ => observer.OnNext(7));
return Disposable.Empty;
});
var debouncedObservable = observable.Debounce(TimeSpan.FromSeconds(1));
debouncedObservable.Subscribe(onNext => Console.WriteLine(onNext));
//4
//7
Örneğin bir kullanıcının bir metin kutusuna girdiği metni izliyorsanız ve kullanıcının yazmayı bıraktığı metni almak istiyorsanız, Debounce operatörünü kullanabilirsiniz.
inputField.onValueChanged.AsObservable()
.Debounce(TimeSpan.FromMilliseconds(500))
.Subscribe(text => Debug.Log($"User has stopped typing: {text}"));
ThrottleFirst
Belirlenen zaman periyodunda gelen ilk veriyi yayar ve süre boyunca gelen diğer akışları bloklar.
var observable = Observable.Create<int>(observer =>
{
observer.OnNext(1);
observer.OnNext(2);
Observable.Timer(TimeSpan.FromSeconds(0.25f)).Subscribe(_ => observer.OnNext(3));
Observable.Timer(TimeSpan.FromSeconds(0.5f)).Subscribe(_ => observer.OnNext(4));
Observable.Timer(TimeSpan.FromSeconds(0.75f)).Subscribe(_ => observer.OnNext(5));
Observable.Timer(TimeSpan.FromSeconds(1.1f)).Subscribe(_ => observer.OnNext(6));
return Disposable.Empty;
});
var throttleFirstObservable = observable.ThrottleFirst(TimeSpan.FromSeconds(1));
throttleFirstObservable.Subscribe(onNext => Debug.Log(onNext));
//1
//6
ThrottleLast
Bir akış geldiği andan itibaren belirlenen zaman periyodunda süre dolana kadar bekler ve o periyotta gelen son veriyi yayınlar.
var observable = Observable.Create<int>(observer =>
{
observer.OnNext(1);
observer.OnNext(2);
Observable.Timer(TimeSpan.FromSeconds(0.5f)).Subscribe(_ => observer.OnNext(3));
Observable.Timer(TimeSpan.FromSeconds(0.9f)).Subscribe(_ => observer.OnNext(4));
Observable.Timer(TimeSpan.FromSeconds(1.1f)).Subscribe(_ => observer.OnNext(5));
Observable.Timer(TimeSpan.FromSeconds(2f)).Subscribe(_ => observer.OnNext(6));
return Disposable.Empty;
});
var throttleLastObservable = observable.ThrottleLast(TimeSpan.FromSeconds(1));
throttleLastObservable.Subscribe(onNext => Debug.Log(onNext));
//4
//6
E) Utility
Do
Observable’ın akışının herhangi bir anında çalıştırılacak callback yaratır. Subscribe ile akışın son halini dinlemek yerine ara formlarla ilgili işlemler yapmak için kullanılabilir.
var observable = Observable.Create<int>(observer =>
{
observer.OnNext(1);
observer.OnNext(2);
observer.OnNext(3);
return Disposable.Empty;
});
observable.Do(onNext => Debug.Log($"Before Where {onNext}"))
.Where(onNext => onNext > 1)
.Do(onNext => Debug.Log($"After Where {onNext}"))
.Select(onNext => onNext * 2)
.Subscribe(onNext => Debug.Log($"Final {onNext}"));
//Before Where 1
//Before Where 2
//Before Where 3
//After Where 3
//Final 6Part 2 : R3/UniRx Unity Örnekleri
Kaynaklar:
Cysharp/R3: The new future of dotnet/reactive and UniRx
R3 — A New Modern Reimplementation of Reactive Extensions for C#
