अतुल्यकालिक प्रोग्रामिंग: डेटा स्ट्रीम

सामग्री

• स्ट्रीम ईवेंट प्राप्त करना
• त्रुटि घटनाएँ
• धाराओं के साथ काम करें
• स्ट्रीम प्रकार
  
  • सदस्यता धाराएँ
  • प्रसारण धाराएँ
• फ्लो प्रोसेसिंग के तरीके
• प्रवाह संशोधित करने के तरीके
  
  • रूपांतरण () फ़ंक्शन
  • फ़ाइल को पढ़ना और डिकोड करना
• सुनो () विधि
• और क्या पढ़ना है?

क्या महत्वपूर्ण है:

• धाराएँ अतुल्यकालिक डेटा अनुक्रम प्रदान करती हैं।
• डेटा अनुक्रम में फ़ाइलों से पढ़ी गई उपयोगकर्ता घटनाएं और डेटा होते हैं।
• Stream एपीआई से या listen() लिए Stream का उपयोग करके संसाधित किया जा सकता है।
• धाराएँ त्रुटियों का जवाब देने का एक तरीका प्रदान करती हैं।
• दो प्रकार की धाराएँ हैं: single subscription धाराएँ और प्रसारण।

डार्ट में एसिंक्रोनस प्रोग्रामिंग भविष्य और स्ट्रीम कक्षाओं की विशेषता है।

Future एक आस्थगित गणना है। यदि कोई सामान्य फ़ंक्शन परिणाम देता है, तो एसिंक्रोनस फ़ंक्शन Future ( future ) ऑब्जेक्ट देता है जिसमें अंततः परिणाम होगा। जब ऑपरेशन पूरा हो जाएगा तो future परिणाम लौटा देगा।

एक धारा अतुल्यकालिक घटनाओं का एक क्रम है। यह एक Iterable अतुल्यकालिक वस्तु के समान है, जहां आप अनुरोध करने पर अगली घटना प्राप्त करने के बजाय, थ्रेड घटना को रिपोर्ट करते हैं जब यह तैयार होता है।

स्ट्रीम ईवेंट प्राप्त करना

धाराएँ अलग-अलग तरीकों से बनाई जा सकती हैं, जो एक अन्य लेख का विषय है, लेकिन वे सभी एक ही तरीके से उपयोग किए जा सकते हैं: लूप के लिए एक अतुल्यकालिक (आमतौर पर जिसे प्रतीक्षा की जाती है ) स्ट्रीम घटनाओं को पुनरावृत्त करता है , जैसे एक संग्रह के माध्यम से लूप पुनरावृत्त करता है। उदाहरण के लिए:

 Future<int> sumStream(Stream<int> stream) async { var sum = 0; await for (var value in stream) { sum += value; } return sum; } 

यह कोड केवल स्ट्रीम से पूर्णांक ईवेंट प्राप्त करता है, उन्हें जोड़ता है और ( future ) राशि लौटाता है। जब लूप का शरीर समाप्त होता है, तो फ़ंक्शन अगली घटना या थ्रेड के समाप्त होने तक रुक जाता है।

फ़ंक्शन को async के साथ चिह्नित किया गया है, जो लूप के लिए वेट का उपयोग करते समय आवश्यक है।

निम्नलिखित उदाहरण ( डार्टपैड पर) async* ( नोट जनरेटर ) के साथ एक फ़ंक्शन का उपयोग करके पूर्णांकों की एक सरल धारा बनाकर पिछले कोड की जांच करता है:

 // Copyright (c) 2015, the Dart project authors. // Please see the AUTHORS file for details. // All rights reserved. Use of this source code is governed // by a BSD-style license that can be found in the LICENSE file. import 'dart:async'; Future<int> sumStream(Stream<int> stream) async { var sum = 0; await for (var value in stream) { sum += value; } return sum; } Stream<int> countStream(int to) async* { for (int i = 1; i <= to; i++) { yield i; } } main() async { var stream = countStream(10); var sum = await sumStream(stream); print(sum); // 55 } 

त्रुटि घटनाएँ

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

कुछ मामलों में, धारा समाप्त होने से पहले एक त्रुटि उत्पन्न होती है; किसी दूरस्थ सर्वर से किसी फ़ाइल को पुनर्प्राप्त करते समय शायद नेटवर्क विफलता या घटनाओं को उत्पन्न करने वाले कोड में एक त्रुटि होती है, किसी को इसके बारे में पता होना चाहिए।

स्ट्रीम डेटा इवेंट की तरह एक त्रुटि घटना की रिपोर्ट कर सकते हैं। अधिकांश थ्रेड्स पहली त्रुटि के बाद बंद हो जाते हैं, लेकिन थ्रेड्स जो एक से अधिक त्रुटि लौटाते हैं और थ्रेड्स जो किसी त्रुटि ईवेंट के बाद डेटा की रिपोर्ट करते हैं, संभव है। इस दस्तावेज़ में, हम केवल उन थ्रेड्स पर चर्चा करते हैं जो एक से अधिक त्रुटि नहीं देते हैं।

जब किसी स्ट्रीम का इंतजार करना पढ़ता है , तो लूप ऑपरेटर द्वारा एक त्रुटि डाली जाती है। यह भी लूप पूरा करता है। आप त्रुटि को पकड़ने की कोशिश कर सकते हैं। निम्न उदाहरण में ( DartPad पर), एक त्रुटि तब होती है यदि लूप इटरेटर 4 है:

 // Copyright (c) 2015, the Dart project authors. // Please see the AUTHORS file for details. // All rights reserved. Use of this source code is governed // by a BSD-style license that can be found in the LICENSE file. import 'dart:async'; Future<int> sumStream(Stream<int> stream) async { var sum = 0; try { await for (var value in stream) { sum += value; } } catch (e) { return -1; } return sum; } Stream<int> countStream(int to) async* { for (int i = 1; i <= to; i++) { if (i == 4) { throw new Exception('Intentional exception'); } else { yield i; } } } main() async { var stream = countStream(10); var sum = await sumStream(stream); print(sum); // -1 } 

धाराओं के साथ काम करें

Stream क्लास में कई सहायक विधियाँ सम्‍मिलित हैं जो एक धारा पर सामान्‍य संचालन कर सकती हैं, Iterable मेथड के समान। उदाहरण के लिए, आप Stream एपीआई से lastWhere() का उपयोग करके एक स्ट्रीम में सबसे छोटा सकारात्मक पूर्णांक पा सकते हैं।

 Future<int> lastPositive(Stream<int> stream) => stream.lastWhere((x) => x >= 0); 

स्ट्रीम प्रकार

सदस्यता धाराएँ

सबसे सामान्य प्रकार की धारा में घटनाओं का एक क्रम होता है जो एक बड़े पूरे के हिस्से होते हैं। घटनाओं को उनमें से किसी को याद किए बिना सही क्रम में वितरित किया जाना चाहिए। यह उस प्रकार की स्ट्रीम है जो आपको फ़ाइल पढ़ने या वेब अनुरोध प्राप्त करने के दौरान मिलती है।

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

प्रसारण धाराएँ

एक अन्य प्रकार की धारा व्यक्तिगत संदेशों के लिए है जिसे एक समय में संसाधित किया जा सकता है। इस तरह की एक धारा का उपयोग किया जा सकता है, उदाहरण के लिए, एक ब्राउज़र में माउस घटनाओं के लिए।

आप किसी भी समय ऐसी धारा को सुनना शुरू कर सकते हैं, और आपको सुनने के दौरान होने वाली घटनाएं प्राप्त होंगी। स्ट्रीम कई श्रोताओं को सुन सकता है। आप पिछली सदस्यता रद्द करने के बाद फिर से स्ट्रीम स्ट्रीम सुनना शुरू कर सकते हैं।

फ्लो प्रोसेसिंग के तरीके

स्ट्रीम <T> में निम्न विधियाँ धारा को संसाधित करती हैं और परिणाम लौटाती हैं:

 Future<T> get first; Future<bool> get isEmpty; Future<T> get last; Future<int> get length; Future<T> get single; Future<bool> any(bool Function(T element) test); Future<bool> contains(Object needle); Future<E> drain<E>([E futureValue]); Future<T> elementAt(int index); Future<bool> every(bool Function(T element) test); Future<T> firstWhere(bool Function(T element) test, {T Function() orElse}); Future<S> fold<S>(S initialValue, S Function(S previous, T element) combine); Future forEach(void Function(T element) action); Future<String> join([String separator = ""]); Future<T> lastWhere(bool Function(T element) test, {T Function() orElse}); Future pipe(StreamConsumer<T> streamConsumer); Future<T> reduce(T Function(T previous, T element) combine); Future<T> singleWhere(bool Function(T element) test, {T Function() orElse}); Future<List<T>> toList(); Future<Set<T>> toSet(); 

drain() और pipe() को छोड़कर ये सभी कार्य Iterable में एक समान कार्य के अनुरूप हैं । उनमें से प्रत्येक को लूप की प्रतीक्षा के साथ एक अतुल्यकालिक फ़ंक्शन का उपयोग करके आसानी से लिखा जा सकता है (या बस अन्य तरीकों में से एक का उपयोग करके)। उदाहरण के लिए, कुछ कार्यान्वयन निम्नानुसार हो सकते हैं:
नाली () और पाइप () को छोड़कर ये सभी कार्य Iterable पर समान कार्य के अनुरूप हैं। लूप की प्रतीक्षा के साथ एक async फ़ंक्शन का उपयोग करके प्रत्येक को आसानी से लिखा जा सकता है (या बस अन्य विधियों में से एक का उपयोग करके)। उदाहरण के लिए, कुछ कार्यान्वयन हो सकते हैं:

 Future<bool> contains(Object needle) async { await for (var event in this) { if (event == needle) return true; } return false; } Future forEach(void Function(T element) action) async { await for (var event in this) { action(event); } } Future<List<T>> toList() async { final result = <T>[]; await this.forEach(result.add); return result; } Future<String> join([String separator = ""]) async => (await this.toList()).join(separator); 

(वास्तविक कार्यान्वयन थोड़ा अधिक जटिल है, लेकिन मुख्यतः ऐतिहासिक कारणों से।)

प्रवाह संशोधित करने के तरीके

स्ट्रीम में निम्न विधियाँ मूल धारा के आधार पर एक नई धारा लौटाती हैं। उनमें से प्रत्येक मूल किसी को सुनने से पहले नई धारा को सुनने के लिए किसी की प्रतीक्षा करता है।

 Stream<R> cast<R>(); Stream<S> expand<S>(Iterable<S> Function(T element) convert); Stream<S> map<S>(S Function(T event) convert); Stream<R> retype<R>(); Stream<T> skip(int count); Stream<T> skipWhile(bool Function(T element) test); Stream<T> take(int count); Stream<T> takeWhile(bool Function(T element) test); Stream<T> where(bool Function(T event) test); 

उपरोक्त विधियां Iterable में समान विधियों के अनुरूप हैं, जो चलने योग्य वस्तु को किसी अन्य चलने योग्य वस्तु में परिवर्तित करते हैं। यह सब लूप की प्रतीक्षा के साथ एक अतुल्यकालिक फ़ंक्शन का उपयोग करके आसानी से लिखा जा सकता है।

 Stream<E> asyncExpand<E>(Stream<E> Function(T event) convert); Stream<E> asyncMap<E>(FutureOr<E> Function(T event) convert); Stream<T> distinct([bool Function(T previous, T next) equals]); 

asyncExpand() और asyncMap() फ़ंक्शन expand() और map() फ़ंक्शन के समान हैं, लेकिन फ़ंक्शन तर्क को एसिंक्रोनस फ़ंक्शन होने की अनुमति देते हैं। distinct() कार्य Iterable में मौजूद नहीं हैं, लेकिन इसे लागू किया जा सकता है।

 Stream<T> handleError(Function onError, {bool test(error)}); Stream<T> timeout(Duration timeLimit, {void Function(EventSink<T> sink) onTimeout}); Stream<S> transform<S>(StreamTransformer<T, S> streamTransformer); 

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

रूपांतरण () फ़ंक्शन

परिवर्तन () फ़ंक्शन केवल त्रुटि से निपटने के लिए नहीं है; यह धाराओं के लिए एक अधिक सामान्यीकृत "मानचित्र" है। एक सामान्य मानचित्र को प्रत्येक आने वाली घटना के लिए एक मूल्य की आवश्यकता होती है। हालाँकि, विशेष रूप से I / O स्ट्रीम के लिए, आउटपुट इवेंट बनाने के लिए कई इनबाउंड इवेंट की आवश्यकता हो सकती है। स्ट्रीमट्रांसफॉर्मर इसकी मदद कर सकता है। उदाहरण के लिए, Utf8Decoder जैसे डिकोडर ट्रांसफार्मर हैं। एक ट्रांसफार्मर के लिए केवल एक बाइंड () फ़ंक्शन की आवश्यकता होती है, जिसे एक अतुल्यकालिक फ़ंक्शन के माध्यम से आसानी से लागू किया जा सकता है।

 Stream<S> mapLogErrors<S, T>( Stream<T> stream, S Function(T event) convert, ) async* { var streamWithoutErrors = stream.handleError((e) => log(e)); await for (var event in streamWithoutErrors) { yield convert(event); } } 

फ़ाइल को पढ़ना और डिकोड करना

निम्न कोड फ़ाइल को पढ़ता है और स्ट्रीम में दो रूपांतरण करता है। सबसे पहले, यह UTF8 से डेटा परिवर्तित करता है, और फिर इसे LineSplitter से गुजरता है। हैशटैग ( # ) से शुरू होने वाले को छोड़कर, सभी लाइनें प्रिंट की जाती हैं।

 import 'dart:convert'; import 'dart:io'; Future<void> main(List<String> args) async { var file = File(args[0]); var lines = file .openRead() .transform(utf8.decoder) .transform(LineSplitter()); await for (var line in lines) { if (!line.startsWith('#')) print(line); } } 

सुनो () विधि

listen() विधि एक "निम्न-स्तरीय" विधि है, फ़ंक्शन के अन्य सभी कार्यों को listen() माध्यम से परिभाषित किया गया है।

 StreamSubscription<T> listen(void Function(T event) onData, {Function onError, void Function() onDone, bool cancelOnError}); 

एक नई प्रकार की स्ट्रीम बनाने के लिए, आप बस Stream क्लास को इनहेरिट कर सकते हैं और listen() विधि को लागू कर सकते हैं, अन्य सभी Stream तरीके कॉल listen() को काम करने के लिए।

सुनने () विधि आपको स्ट्रीम को सुनना शुरू करने की अनुमति देती है। जब तक आप ऐसा नहीं करते, तब तक स्ट्रीम एक अक्रिय वस्तु है जो बताती है कि आप किन घटनाओं को सुनना चाहते हैं। सुनते समय, StreamSubscription ऑब्जेक्ट लौटाया जाता है जो घटनाओं को उत्पन्न करने वाली सक्रिय धारा का प्रतिनिधित्व करता है। यह समान है कि Iterable केवल वस्तुओं का संग्रह कैसे है, और एक पुनरावृत्ति वह है जो वास्तविक पुनरावृत्ति करता है।

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

और क्या पढ़ना है?

डार्ट 2. अतुल्यकालिक प्रोग्रामिंग: वायदा