Kasus menarik untuk menggunakan operator RxJs khusus | Oleh Enrico Pisinin Juni 2022

Pustaka RxJs hadir dengan seperangkat operator yang mampu menangani sebagian besar hal yang perlu kita tangani saat menangani aliran yang terlihat.

Namun, masih ada situasi di mana kemungkinan membangun operator kustom baru dapat bermanfaat. Mari kita lihat kasus tertentu yang terinspirasi oleh pertanyaan yang diposting di stackoverflow.

Pertimbangkan skenario di mana kita memiliki aliran catatan, misalnya, dari “kotak hitam” mobil, yang mengalir ke sistem penerima yang harus menyimpannya dalam database secepat mungkin.

Sistem penerima memiliki API, saveRec, Yang dapat menerima larik catatan untuk disimpan dan merespons secara tidak sinkron ketika semua catatan ditulis. Sayangnya kami juga memiliki batasan: saveRec Tidak dapat dipanggil secara bersamaan, dengan kata lain ketika kita dapat memanggil API, kita harus menunggu jawaban sebelum menelepon kembali.

Ketika tingkat konkurensi terbatas pada 1, metode yang biasa digunakan untuk RxJs adalah dengan menggunakannya concatMap Operator (yang tidak lain adalah mergeMap Dengan concurrent Parameter diatur 1).

Menggunakan concatMap Dalam hal ini, solusinya mungkin tidak optimal. Sebagai contoh, mari kita asumsikan bahwa interval waktu antara record yang masuk kurang dari waktu yang dibutuhkan saveRec Untuk menjawab ini, kami mengumpulkan catatan yang menunggu untuk disimpan (catatan tersebut disimpan dalam buffer internal. concatMap Operator, tetapi kami akan kembali ke titik ini nanti).

Mengingat bahwa saveRec Itu dapat menerima array catatan, idealnya kita harus memanggilnya dengan semua catatan buffer selama panggilan sebelumnya.

Misalnya, jika kita menerima rec_4 Dan rec_5 Selama eksekusi saveRec Operasi di rec_3Kemudian di panggilan berikutnya kami ingin melewati keduanya rec_4 Dan rec_5 Untuk diselamatkan.

Dan inilah yang akan kami terapkan dengan operator kustom baru bufferConcatMap. faktanya bufferConcatMap Adalah Kinerja pabrik operator pipaKarena ini adalah fungsi yang mengembalikan a saat dipanggil Operator pipa Yang kemudian digunakan sebagai parameter pipe Metode dari Observable.

Sebelum mengimplementasikan, mungkin ada baiknya untuk mendefinisikan RxJs Operator pipa Dan apa hasilnya? Kinerja pabrik operator pipa Adalah.

Operator pipa Adalah fungsi yang menerima Observable sebagai input dan mengembalikan Observable sebagai output (kode RxJs mewakili Operator pipa Ketik dengan antarmuka OperatorFunction<T, R> Atau salah satu ekstensinya).

Kinerja pabrik operator pipa Adalah fungsi yang mengembalikan Operator pipa.

Jadi operator kustom baru kami bufferConcatMap Ini adalah fungsi yang mengembalikan fungsi yang mengharapkan pengamat sebagai input dan mengembalikan pengamat.

Sementara ini adalah satu Kinerja pabrik operator pipaMulai sekarang, kami hanya akan menyebutnya sebagai Operator khususKarena ini adalah metode yang diterima untuk merujuk ke jenis fungsi ini.

Operator kustom sederhana dapat dikodekan dengan cara ini.

Operator kustom ini membuat Observable baru yang mencerminkan sumber Observable. Dengan kata lain, ia tidak melakukan apa-apa, tetapi masih menunjukkan bagaimana mekanisme itu bekerja.

Sekarang, misalkan kita ingin mengonversi nilai apa pun yang dideklarasikan oleh sumber yang terlihat dan kemudian mengirim nilai yang dikonversi ke pelanggan dengan indeksnya (dengan kata lain, kita ingin membuat rxJs.map Operator dari awal).

Ini bisa dilakukan

• Menyeberang ke Kinerja pabrik operator pipa Fungsi yang digunakan untuk mengonversi nilai apa pun yang diterima dari sumber
• Gunakan konsep “tutup” JavaScript untuk mendefinisikan variabel (variabel index Dalam contoh ini) yang diinisialisasi setiap kali Observable dikembalikan dan kapan pun next Fungsinya dipanggil subscriber

Sangat menarik untuk dicatat bahwa menggunakan penutupan JavaScript, kami telah melampirkan beberapa “status” ( index Variabel) ke suatu fungsi, the next Fungsi ditentukan oleh subscriber. Kemampuan untuk menambahkan status ke fungsi penting dalam eksekusinya bufferConcatMap.

Sekarang setelah kita menguasai dasar-dasar operator kustom, mari kita mulai membangunnya sendiri bufferConcatMap.

Perilaku yang diharapkan

Perilaku yang diharapkan dapat diekspresikan dengan diagram kelereng, menggunakan contoh yang dijelaskan di atas.

Hal ini diungkapkan dalam kata-kata yang dapat kita katakan:

• Jika upstream memberi tahu dan tidak ada pemrosesan yang sedang berlangsung, maka nilai yang dinyatakan akan segera diproses
• Jika upstream memberi tahu dan pemrosesan sedang berlangsung, nilai yang dideklarasikan disimpan dalam buffer.
• Segera setelah pemrosesan selesai, jika ada sesuatu di buffer, itu segera diproses, jika tidak, kami menunggu elemen berikutnya untuk memulai pemrosesan baru.

Demi kesederhanaan, kami tidak menutupi kata “persyaratan” error Dan complete (Tapi mereka ada dalam kode).

Kami membutuhkan sebuah negara

Ya, kami perlu menyimpan beberapa informasi status agar operator kami berfungsi seperti yang diharapkan:

• Kita perlu memiliki buffer untuk menampung nilai-nilai yang berasal dari upstream saat masih memproses permintaan.
• “Sementara kami memproses permintaan”? Ini berarti kita harus menemukan cara untuk mengetahui apakah kita sedang memproses aplikasi.

Jadi mode kami terdiri dari 2 variabel:

• buffer (Array) untuk menampung hal-hal yang tidak dapat kita proses.
• processing Bendera (boolean) tempat kita menyimpan apakah ada permintaan dalam penerbangan atau tidak.

Anatomi Operator

Pada bufferConcatMap Operator sebenarnya adalah Kinerja pabrik operator pipa Yang mengharapkan project Bertindak sebagai parameter input dan mengembalikan a Operator pipa.

Parameter masukan

Pada project Fungsi adalah fungsi yang mengharapkan array nilai pada input dan mengembalikan Observable. Mengapa array nilai sebagai input? Karena bufferConcatMap Menyimpan nilai yang diterima dalam buffer yang merupakan array dan buffer ini ke project Bekerja segera setelah tidak ada pemrosesan dengan cepat.

Mengapa mengembalikan Observable? Karena bufferConcatMap Sebagai berbagai concatMap Dan sebagai concatMap Apakah, mengharapkan fungsi yang dikirim sebagai input menjadi fungsi yang mengembalikan Observable.

Kembali Operator pipa

Pada Operator pipa Dikembalikan sebagai output itu sendiri adalah fungsi yang mengharapkan a source Mengembalikan Terlihat sebagai Input dan “Tampilan Konversi Baru”.

“Tampilan yang Dikonversi Baru” berperilaku serupa dengan apa yang dibuatnya concatMap Jika tidak ada permintaan yang tertunda: panggil saja project Merupakan fungsi dan menunggu jawabannya sebelum panggilan berikutnya, perbedaannya adalah project Fungsi tersebut disebut dengan buffer Nilai yang timbul dari source Terlihat (hulu) dan tidak hanya dengan nilai. Di sisi lain, jika permintaan sedang diproses, itu akan menerima jumlah yang diumumkan olehnya source Itu terlihat dan menyimpannya di buffer internal tanpa memanggil project Fungsi (dengan kata lain tanpa efek hilir).

Kode operator

Akhirnya, datang kode lengkap dari operator kustom kami.

Berikut adalah contoh cara menggunakannya:

Mari kita lihat beberapa konsep kunci yang mendasari logika ini.

Sebagian besar fungsi eksternal (bufferConcatMap). Fungsi paling eksternal, a Kinerja pabrik operator pipaDisebut sebagai bagian dari panggilan pipe Yang mewakili parameter untuk itu (dengan kata lain, lihat contoh di atas, ketika newTransformerObservable telah dibuat).

Pada Operator pipa. Fungsi dikembalikan oleh bufferConcatMap Nyata Operator pipaDisebut hanya jika newTransformerObservable Apakah umum.

Status internal operator. Kapan Operator pipa (Pada saat berlangganan) dipanggil dari variabel yang memegang status operator (bufferedNotifications Dan processing Dalam kasus kami) diinisialisasi. Variabel-variabel ini adalah bagian dari rentang leksikal dari setiap fungsi yang didefinisikan dalam tubuh Operator pipa (Misalnya fungsi didefinisikan sebagai parameter yang dikirim ke concatMap Operator tersemat dalam eksekusi bufferConcatMap) Dan oleh karena itu dibagikan di antara semua panggilan ke fungsi-fungsi ini, setiap kali sumber yang terlihat (hulu) mengumumkan nilai baru.

project Dan EMPTY. implementasi dari bufferConcatMap Ada berbagai di sekitar concatMap Yang menggunakan mode internal untuk menjalankan perilaku yang diinginkan. concatMap Sebuah fungsi harus dikirim sebagai input yang mengembalikan Observable. Jika beberapa pemrosesan sedang berlangsung saat nilai baru diberitahukan oleh upstream, bufferedConcatMap Hanya perlu jumlah seperti itu dan jangan lakukan apapun. Dapat dilihat bahwa itu tidak melakukan apa-apa EMPTY Terlihat sama mudahnya terlihat completeSegera setelah Anda menerima pemberitahuan pertama dari hulu. Jadi itu sebabnya bufferConcatMap menggunakan EMPTY Kapan processing Adalah true. Sebaliknya, jika tidak ada proses yang terjadi, project Fungsi dipanggil dengan nilai-nilai yang disimpan di buffer, selama nilai-nilai itu ada di dalamnya.

Sejauh ini bagus, setidaknya seperti itu. Tapi ada sesuatu yang kita abaikan.

Mari kita asumsikan kita membuat menggunakan Observable bufferConcatMap Seperti ini dan kemudian kami dua kali, secara bersamaan, membagikannya, seperti ini

const newObs = source.pipe(bufferConcatMap((val) => process(val)));newObs.subscribe(values => // do something);newObs.subscribe(values => // do something else);

Dalam hal ini, variabel mode (bufferedNotifications Dan processing) Ketika mereka diinisialisasi bufferConcatMap Disebut, dengan kata lain ketika newObs Telah dibuat.

Oleh karena itu, variabel yang sama dibagikan di kedua akun, yang salah dan dapat menyebabkan kesalahan yang sangat halus. Kita perlu memastikan bahwa setiap langganan memiliki salinan sendiri dari variabel status yang diinisialisasi pada saat berlangganan.

Untungnya dalam hal ini operator rxJs defer Ini membantu kami sehingga versi akhir operator yang benar adalah sebagai berikut.

Dalam kebanyakan kasus, operator yang disediakan oleh perpustakaan RxJS sudah cukup, tetapi terkadang memiliki kemampuan untuk membangun operator khusus dapat bermanfaat. Jika demikian, kita harus ingat bahwa yang dapat diamati hanyalah fungsi khusus. Operator hanyalah fungsi yang bekerja pada fungsi khusus tersebut.

Oleh karena itu, membangun operator kustom adalah latihan kombinasi fungsi yang harus mengikuti seperangkat aturan terbatas. Dengan menggunakan paket, kita dapat menambahkan status internal ke operator untuk meningkatkan kemampuan mereka sambil mempertahankan perilaku eksternal.

Semua kode di belakang bufferConcatMapTes semacam itu dapat ditemukan di repositori ini.