🤢 🍓 🍾 RxJS का उपयोग करके प्रतिक्रियाशील प्रोग्रामिंग की मूल बातें। भाग 3. उच्च आदेश वेधशाला 🏒 🐴 ⤵️

इस लेख में हम देखेंगे कि एक धागे में दूसरे को कैसे संसाधित करना संभव है, इसकी आवश्यकता क्यों है, और हायर ऑर्डर ऑब्जर्वबल्स (बाद में HOO के रूप में संदर्भित) ऑपरेटरों को इसमें कैसे मदद मिलेगी।

लेखों की श्रृंखला "RxJS का उपयोग करके प्रतिक्रियात्मक प्रोग्रामिंग के मूल तत्व":

थ्रेड्स के साथ काम करते समय, एक स्थिति अक्सर उत्पन्न होती है जब मूल्य के रूप में दूसरे के परिणामों को एक धागे में स्थानांतरित करना आवश्यक होता है। उदाहरण के लिए, हम एक अजाक्स अनुरोध को निष्पादित करना चाहते हैं और वर्तमान थ्रेड में अपनी प्रतिक्रिया को संसाधित करते हैं, या कई समानांतर अनुरोधों को चलाते हैं, पूलिंग को लागू करते हैं। मुझे लगता है कि कई लोगों को वादे जैसे तंत्र का उपयोग करके ऐसी समस्याओं को हल करने के लिए उपयोग किया जाता है। लेकिन क्या RxJS का उपयोग करके उन्हें हल करना संभव है? बेशक, और सब कुछ बहुत आसान है जितना आप सोचते हैं!

नोट : लेख के सैद्धांतिक भाग को समझने के लिए, आपको पिछले लेखों को पढ़ने की आवश्यकता नहीं है, आपको बस यह जानने की जरूरत है कि क्या अवलोकन योग्य है, ऑपरेटर और पाइप क्या हैं। व्यावहारिक भाग में, हम दूसरे लेख से उदाहरण को परिष्कृत करेंगे, जिसे आप यहां पा सकते हैं।

समस्या

आइए निम्नलिखित कार्य की कल्पना करें: हमें हर सेकंड यह पता लगाने की आवश्यकता है कि क्या सर्वर सुलभ है। हम इसे कैसे हल कर सकते हैं?

सबसे पहले, टाइमर विधि का उपयोग करके एक स्ट्रीम बनाएं:

timer(0, 1000).subscribe({ next: console.log });

अंतराल के सिद्धांत में टाइमर विधि बहुत समान है। लेकिन इसके विपरीत, यह आपको थ्रेड स्टार्ट टाइमआउट सेट करने की अनुमति देता है, जो पहले पैरामीटर द्वारा प्रेषित होता है। दूसरा पैरामीटर अंतराल को इंगित करता है जिसके माध्यम से एक नया मूल्य उत्पन्न होगा। यदि दूसरा पैरामीटर निर्दिष्ट नहीं है, तो टाइमर केवल एक मान उत्पन्न करेगा और स्ट्रीम को समाप्त करेगा।

चूंकि आपके और मेरे पास सर्वर नहीं है, मैं सुझाव देता हूं कि केवल एक फ़ंक्शन लिखें जो सर्वर के लिए अनुरोध का अनुकरण करता है:

 const makeRequest = () => { return timer(1000).pipe( mapTo('success') ) }

यह विधि क्या करती है? यह टाइमर पद्धति का उपयोग करके बनाई गई एक धारा को लौटाता है, जो एक सेकंड बीतने और समाप्त होने के बाद एक मूल्य का उत्सर्जन करता है। चूंकि टाइमर विधि केवल एक संख्या उत्पन्न करती है, हम इसे "सफलता" स्ट्रिंग के साथ बदलने के लिए मैपटो ऑपरेटर का उपयोग करते हैं।

यह वह तरीका है जो मेक रेयर विधि द्वारा बनाई गई धारा जैसा दिखता है:

अब हमारे पास एक विकल्प है: स्ट्रीम के अंदर मेकएरेस्टेस्ट विधि को कॉल करना या पर्यवेक्षक को यह जिम्मेदारी सौंपना?

पहला दृष्टिकोण बेहतर है, क्योंकि इस मामले में हम अपने ऑपरेटरों के साथ RxJS की पूरी क्षमता का उपयोग करने में सक्षम होंगे और अपने पर्यवेक्षकों को अनावश्यक कर्तव्यों से छुटकारा दिलाएंगे। हम अंतराल द्वारा अनुरोधों को निष्पादित करने के लिए टाइमर विधि का उपयोग करते हैं:

 timer(0, 1000).pipe( map(() => makeRequest()) ).subscribe({ next: console.log });

जब हम इस तरह का कोड चलाते हैं, तो हम देखेंगे कि कंसोल में। हमें पाठ "सफलता" के साथ एक संदेश नहीं मिलता है, लेकिन ऑब्जर्वेबल प्रकार की वस्तु:

जवाब काफी अपेक्षित है, क्योंकि नक्शे में हम धारा को वापस करते हैं। काम करने के लिए एक स्ट्रीम के लिए, आपको इसकी सदस्यता लेने की आवश्यकता है। खैर, आइए देखें कि यह कैसे करना है :

 timer(0, 1000).pipe( map(() => makeRequest()) ).subscribe({ next: observable => observable.subscribe({ next: console.log }); });

ऊपर दिए गए उदाहरण के साथ समस्या यह है कि हमें एक सदस्यता में सदस्यता मिलती है। लेकिन क्या होगा अगर हम एक श्रृंखला में एक से अधिक अनुरोध करना चाहते हैं? या क्या होगा अगर किसी बिंदु पर हमें अंदर प्रवाह से सदस्यता समाप्त करने की आवश्यकता है? इस मामले में, हमारा कोड अधिक से अधिक "नूडल्स" जैसा होगा। इस समस्या को हल करने के लिए, RxJS के पास HOO नामक विशेष ऑपरेटर हैं।

HOO

HOO एक विशेष प्रकार का कथन है जो धाराओं को मान के रूप में स्वीकार करता है। ऐसा ही एक ऑपरेटर मर्ज है।

जब कोई स्ट्रीम mergeAll पर आती है, तो वह इसकी सदस्यता लेती है। ऑपरेटर ने जिस स्ट्रीम को सब्सक्राइब किया है उसे आंतरिक कहा जाता है। वह धारा जिससे संचालक अन्य प्रवाह प्राप्त करता है मानों को बाह्य कहा जाता है।

जब कोई आंतरिक थ्रेड कोई मान उत्पन्न करता है, तो मर्ज सभी बाहरी थ्रेड में उस मान को धकेलता है। इस प्रकार, हमें मैन्युअल रूप से सदस्यता लेने की आवश्यकता से छुटकारा मिलता है। अगर हम बाहरी प्रवाह से अनसब्सक्राइब करते हैं, तो आंतरिक रूप से मर्ज सभी स्वतः ही अनसब्सक्राइब हो जाएगा।

आइए देखें कि कैसे हम मर्ज के साथ अपने उदाहरण को फिर से लिख सकते हैं:

 timer(0, 1000).pipe( map(() => makeRequest()) mergeAll() ).subscribe({ next: console.log });

ऊपर के उदाहरण में, टाइमर स्टेटमेंट द्वारा बाहरी धारा बनाई गई थी। और मानचित्र ऑपरेटर में जो प्रवाह बनाए जाते हैं वे आंतरिक होते हैं। प्रत्येक बनाया गया धागा मर्ज ऑल स्टेटमेंट में आता है।

संयोजन मानचित्र + मर्जएल का उपयोग अक्सर किया जाता है, इसलिए RxJS में एक मर्ज मैप विधि है:

 timer(0, 1000).pipe( mergeMap(() => makeRequest()) ).subscribe({ next: console.log });

जब कोई बाहरी थ्रेड कोई मान उत्पन्न करता है, तो मर्ज मैप ऑपरेटर कॉलबैक फ़ंक्शन को इसके पास भेज देता है, जो एक नया थ्रेड उत्पन्न करता है। फिर मर्ज मैप सबस्क्राइब्ड जनरेट स्ट्रीम को।

MergeAll / mergeMap ऑपरेटर की ख़ासियत यह है कि यदि कोई अन्य धारा इसके नीचे आती है, तो यह भी इसकी सदस्यता लेता है। इस प्रकार, एक बाहरी धारा में, हम एक साथ कई आंतरिक लोगों से मान प्राप्त कर सकते हैं। आइए निम्न उदाहरण देखें:

  timer(0, 1000)

यह है कि मर्जपेज ऑपरेटर के बिना बाहरी धारा कैसे दिखाई देगी:

और इसलिए मर्ज मैप के साथ:

 timer(0, 1000).pipe( mergeMap(() => interval(1000)) )

हर सेकंड, हम एक नया आंतरिक थ्रेड बनाते हैं और मर्ज मैप को सब्सक्राइब करते हैं। इस प्रकार, हमारे पास एक साथ काम करने वाले कई आंतरिक धागे हैं, जिनमें से मूल्य बाहरी में आते हैं:

नोट : मर्ज मैप का उपयोग करते हुए सावधान रहें, प्रत्येक नया आंतरिक धागा तब तक काम करेगा जब तक आप बाहरी से अनसब्सक्राइब नहीं करते। ऊपर के उदाहरण में, आंतरिक धागे की संख्या हर सेकंड बढ़ रही है, अंत में इतने सारे धागे हो सकते हैं कि कंप्यूटर लोड के साथ सामना नहीं कर सकता है।

संघ / संघ

जब आप आंतरिक थ्रेड्स के निष्पादन क्रम के बारे में परवाह नहीं करते हैं, तो मर्जपेज विधि बहुत बढ़िया है, लेकिन अगर आपको इसकी आवश्यकता है तो क्या होगा? मान लीजिए कि हम अगले सर्वर अनुरोध को पिछले एक से प्रतिक्रिया प्राप्त करने के बाद ही निष्पादित करना चाहते हैं?

ऐसे उद्देश्यों के लिए, HOO ऑपरेटर concatAll / concatMap उपयुक्त है। यह ऑपरेटर, आंतरिक धागे की सदस्यता ले चुका है, जब तक यह खत्म नहीं हो जाता है, तब तक इंतजार करता है और उसके बाद ही अगले एक की सदस्यता लेता है।

यदि एक धागे के निष्पादन के दौरान एक नया इसे उतरता है, तो इसे कतार में रखा जाता है जब तक कि पिछले एक पूरा न हो जाए।

 // ,  1     const firstInnerObservable = timer(1000).pipe( mapTo(1) ); // ,  2     const secondInnerObservable = timer(500).pipe( mapTo(2) ); of( firstInnerObservable, secondInnerObservable ).pipe( concatAll() ).subscribe({ next: console.log });

ऊपर दिए गए उदाहरण में, हम टाइमर विधि का उपयोग करके दो धागे बनाते हैं। स्पष्टता के लिए, मैंने विभिन्न मूल्यों को प्रदर्शित करने के लिए mapTo ऑपरेटर का उपयोग किया। पहला धागा 1 उत्पन्न करेगा, दूसरा - 2. एक बाहरी धागा विधि के उपयोग से बनाया गया है, जो इनपुट के रूप में उपरोक्त दो को लेता है।

ConcatAll स्टेटमेंट पहले FirstInnerObservable को प्राप्त करता है, इसे सब्सक्राइब करता है, और इसके पूरा होने का इंतजार करता है, और केवल पहला पूरा होने के बाद दूसरेInnerObservable को। यहाँ बाहरी धारा कैसी दिखेगी:

यदि हम concatAll को mergeAll से प्रतिस्थापित करते हैं, तो स्ट्रीम इस तरह दिखाई देगी:

 of( firstInnerObservable, secondInnerObservable ).pipe( mergeAll() ).subscribe({ next: console.log });

switchAll / switchMap

यह ऑपरेटर पिछले वाले से अलग है जब यह एक नई स्ट्रीम प्राप्त करता है, तो यह तुरंत पिछले एक से सदस्यता समाप्त कर देता है और नए को सब्सक्राइब करता है।

ऊपर दिए गए उदाहरण को लें और switchAll को बदल दें और फिर देखें कि बाहरी प्रवाह कैसे व्यवहार करता है:

 of( firstInnerObservable, secondInnerObservable ).pipe( switchAll() ).subscribe({ next: console.log });

केवल दूसरी आंतरिक धारा से मूल्य बाहरी धारा में मिला। ऐसा इसलिए है क्योंकि स्विच थ्रेड पहले से अनसब्सक्राइब हो जाता है जब वह दूसरा धागा प्राप्त करता है।

इसकी आवश्यकता कब है? उदाहरण के लिए, डेटा खोज को लागू करते समय। यदि सर्वर से प्रतिक्रिया अभी तक नहीं आई है, और हमने पहले ही एक नया अनुरोध भेजा है, तो हमें पिछले एक के लिए इंतजार करने का कोई मतलब नहीं है।

थकावट / थकावट

एग्जॉस्ट स्विचएल स्टेटमेंट के बिल्कुल विपरीत है, और इसका व्यवहार कॉनसैट के समान है। यह विधि, स्ट्रीम की सदस्यता, इसे पूरा करने के लिए इंतजार करती है। यदि कोई नई धारा इसमें आती है, तो इसे छोड़ दिया जाता है।

 of( firstInnerObservable, secondInnerObservable ).pipe( exhaust() ).subscribe({ next: console.log });

ऊपर के उदाहरण में, हमें एक ड्यूस नहीं मिला, क्योंकि उस समय ऑपरेटर पहले धागे के पूरा होने की प्रतीक्षा कर रहा था, और बस दूसरे को गिरा दिया।

मुझे लगता है कि कई लोगों के पास एक सवाल है, ऐसे व्यवहार की आवश्यकता कब हो सकती है? एक अच्छा उदाहरण लॉगिन फॉर्म है। यह सर्वर के लिए कई अनुरोध भेजने का कोई मतलब नहीं है जब तक कि एक पूरा नहीं हो जाता।

हम आवेदन को अंतिम रूप दे रहे हैं

हम दूसरे लेख से उदाहरण याद करते हैं। इसमें, हमने GitHub पर एक खोज लागू की और सर्वर पर अनुरोध भेजने के लिए मर्ज मैप ऑपरेटर का उपयोग किया। अब हम इस ऑपरेटर की विशेषताओं को जानते हैं, क्या यह वास्तव में हमारे मामले में उपयुक्त है?

 fromEvent(input, 'keyup').pipe( debounceTime(700), map(event => event.target.value), filter(val => val.length > 2), distinctUntilChanged(), mergeMap(value => { return from(getUsersRepsFromAPI(value)).pipe( catchError(err => of([])) ) }) ).subscribe({ next: reps => recordRepsToList(reps) })

मान लेते हैं कि GitHub सर्वर भारी भरकम होगा, तो हमारी प्रतिक्रिया को संसाधित करने में बहुत समय लगेगा। इस मामले में संभवतः क्या गलत हो सकता है?

मान लीजिए कि किसी उपयोगकर्ता ने कुछ डेटा दर्ज किया है, उसने उत्तर की प्रतीक्षा नहीं की, और नए में प्रवेश किया। इस मामले में, हम सर्वर को दूसरा अनुरोध भेजेंगे। हालांकि, कोई भी गारंटी नहीं देता है कि पहले अनुरोध का जवाब पहले आएगा।

चूंकि मर्जपेज ऑपरेटर आंतरिक प्रवाह को संसाधित करने के लिए किस क्रम में परवाह नहीं करता है, इस मामले में जब पहले अनुरोध को दूसरे की तुलना में बाद में निष्पादित किया जाता है, तो हम वास्तविक डेटा को मिटा देंगे। इसलिए, मैं स्विचपाइप के साथ मर्जपाइप विधि को बदलने का प्रस्ताव करता हूं:

 fromEvent(input, 'keyup').pipe( debounceTime(700), map(event => event.target.value), filter(val => val.length > 2), distinctUntilChanged(), switchMap(value => { return from(getUsersRepsFromAPI(value)).pipe( catchError(err => of([])) ) }) ).subscribe({ next: reps => recordRepsToList(reps) })

अब, यदि उपयोगकर्ता नया डेटा दर्ज करता है, तो स्विचपॉइंट पिछली स्ट्रीम से अनसब्सक्राइब करेगा और नए को सब्सक्राइब करेगा।

यह ध्यान देने योग्य है कि जब तक सर्वर इसका जवाब नहीं देता, तब तक हमारा http अनुरोध लटका रहेगा। लेकिन, जब से हमने आंतरिक धारा से सदस्यता समाप्त कर ली है, उत्तर बाहरी धारा में नहीं आएगा।

नोट : यदि आप कोणीय के साथ काम करते हैं और http के साथ काम करने के लिए HttpClient का उपयोग करते हैं, तो आप स्वयं अनुरोध को रद्द करने के बारे में चिंता नहीं कर सकते। अनसब्सक्राइब करने पर HttpClient आपके लिए ऐसा कर सकता है।

Http रद्द करें

एप्रनकंट्रोलर का उपयोग करके एसीसी अनुरोध को रद्द करने की क्षमता है। स्विचपाइप ऑपरेटर के साथ संयुक्त होने पर, यह कार्यक्षमता उपयोगकर्ता ट्रैफ़िक को बचाएगा।

आइए हमारे उदाहरण को थोड़ा फिर से लिखें। और एक विधि बनाएं जो भ्रूण कॉल को अवलोकनीय में लपेटे:

 const createCancellableRequest = (url) => { //      const controller = new AbortController(); const signal = controller.signal; return new Observable(observer => { fetch(url, { signal }) .then(response => { if (response.ok) { return response.json(); } throw new Error(''); }) //     .then(result => observer.next(result)) //   .then(() => observer.complete()) //   ,     .catch(error => observer.error(error)); // ,    return () => { //   controller.abort(); }; }); };

इसके अलावा getUsersRepsFromApi विधि बदलें:

 const getUsersRepsFromAPI = (username) => { const url = `https://api.github.com/users/${ username }/repos`; return createCancellableRequest(url); }

अब विधि प्रतिज्ञा नहीं, बल्कि अवलोकनीय है। इसलिए, हम switchMap में आवरण से निकालते हैं:

 switchMap(value => { return getUsersRepsFromAPI(value).pipe( catchError(err => of([]) ) )

नोट : RxJS संस्करण 6.5 में, उन्होंने fromFetch स्टेटमेंट जोड़ा है, जो खुद को हूड के तहत गर्भपात विधि कहता है, इसलिए आपको अब अपनी "बाइक" लिखने की आवश्यकता नहीं है।

वह सब है! सभी नमूना कोड यहां देखे जा सकते हैं ।

निष्कर्ष

आज हमने देखा कि HOO क्या है और इस श्रेणी के कुछ बहुत ही उपयोगी ऑपरेटर हैं। बेशक, ये उन सभी से बहुत दूर थे। अधिक विस्तृत और विस्तृत जानकारी के लिए, मैं RxJS प्रलेखन पर जाने की सलाह देता हूं।

अगले लेख में मैं इस बात पर विचार करने की योजना बना रहा हूं कि हॉट और कोल्ड ऑब्जर्वर में क्या अंतर है।

अंत में: सदस्यता में सदस्यता का उपयोग न करें, क्योंकि वहां HOO है!

RxJS का उपयोग करके प्रतिक्रियाशील प्रोग्रामिंग की मूल बातें। भाग 3. उच्च आदेश वेधशाला