javascript - RxJS - başlangıç durumu ve güncellemeler

Websocket'ten veri almam gerekiyor ve bunu yapmak için RxJS kullanmak istiyorum.

Son başlangıç verileri (~1000 kayıt) için bir websocket 1 ve artımlı güncellemeler için bir websocket 2 vardır.

İki gözlemlenebilir (observable) oluşturdum:

• initalState$ websocket 1'e gider ve başlangıç verilerini alır, ardından tamamlanır.
• updateEvent$ websocket 2'ye gider ve sürekli olarak güncellemeleri alır.

Başlangıçtaki uygulamam şuydu:

initialState.subscribe(initialData=> {
    console.log(initialData);
    updateEvent.subscribe(updateEvent => {
        console.log(updateEvent);
    });
});

Karşılaştığım sorun, initalState alındıktan ve ilk güncelleme (updateEvent) alındıktan sonra bir boşluk olmasıdır.

(İlk verileri aldığımda ve abone olmadan önce gerçekleşen güncellemeyi kaybedebilirim).

Her iki gözlemevime aynı anda abone olan ve initalState tamamlanana kadar updateEvent gözlemcisini tamponlayan ve ardından doğru sırada "önce başlangıç verisi" sonra "güncellemeler" olacak şekilde bir yeni Gözlemci oluşturmanın pratik bir yolu var mı?

Temelde initialState’i sadece “ilk” güncelleme yapmak, ama ondan sonra eksik güncelleme olmadığından emin olmak.

Cevap #1

Görünüşe göre, ilk websocket yayını veri göndermeye başlayana kadar ikinci websocket yayını için buffer kullanarak ihtiyacınız olanı elde edebilirsiniz. Ancak, bu zincir biraz daha karmaşık hale geliyor çünkü değerleri yalnızca ilk yayın veri göndermeye başladıktan sonra almaya başlamak istiyorsunuz.

const initialStateShared = initialState.pipe(share());
const updateEventShared = updateEvent.pipe(share());

merge(
  initialStateShared,
  updateEventShared.pipe( // Buffer the second stream but only once
    buffer(initialStateShared),
    take(1),
  ),
  updateEventShared.pipe( // Updates from the second stream will be buffered first and then continue comming from here
    skipUntil(initialStateShared),
  )
).subscribe(...);

Cevap #2

Eğer doğru anlıyorsam, sizin istediğiniz her iki isteği aynı anda tetiklemek ve yalnızca her ikisi de zaten mevcutsa onlara abone olmak. Sanırım aradığınız combineLatest operatörü.

combineLatest([initialState$, updateEvent$]).subscribe(([initialState, updateEvent] => {
    console.log({initialState, updateEvent});
}));

Bu şekilde birleşik observable, hem initialState$hem de updateEvent$bir şeyler yayana kadar bekleyecek ve bundan sonra birleşik observablardan herhangi biri bir şeyler yayarsa emit tetikleyecektir. Daha fazla bilgi için https://www.learnrxjs.io/operators/combination/combinelatest.html adresine bakın.

Not: Başka bir subscribe içinde subscribe yapmaktan kaçınmalısınız. Bu genellikle yanlış bir şey yapıldığının kod kokusudur.