Node.js स्ट्रीम बनाम Observables

Question

Observables के बारे में सीखने के बाद, मुझे उन्हें Node.js streams के समान लगता है। जब भी नया डेटा आता है, उपभोक्ता को सूचित करने की दोनों प्रणालियों में एक त्रुटि होती है, एक त्रुटि होती है या कोई और डेटा (ईओएफ) नहीं होता है।

मुझे दोनों के बीच वैचारिक/कार्यात्मक मतभेदों के बारे में जानना अच्छा लगेगा। धन्यवाद! अतुल्यकालिक रूप से मूल्यों का एक दृश्य की प्रक्रिया:

Answer 1

दोनों observables और Node.js के स्ट्रीम आप एक ही मूल समस्या हल करने के लिए अनुमति देते हैं। मेरा मानना ​​है कि दोनों के बीच मुख्य अंतर, उस संदर्भ से संबंधित है जो इसकी उपस्थिति को प्रेरित करता है। वह संदर्भ शब्दावली और एपीआई में परिलक्षित होता है।

पर अवलोकन पक्ष आपके पास एक्मास्क्रिप्ट का विस्तार है जो प्रतिक्रियाशील प्रोग्रामिंग मॉडल पेश करता है। यह Observer और Observable की न्यूनतम और संगत अवधारणाओं के साथ मूल्य उत्पादन और एसिंच्रोनिटी के बीच अंतर को भरने का प्रयास करता है।

node.js और स्ट्रीम पक्ष पर आप नेटवर्क धाराओं और स्थानीय फ़ाइलों के असीमित और निष्पादक प्रसंस्करण के लिए एक इंटरफ़ेस बनाना चाहते थे। शब्दावली कि प्रारंभिक संदर्भ से निकला है और आप प्राप्त pipe, chunk, encoding, flush, Duplex, Buffer, आदि एक व्यावहारिक दृष्टिकोण है कि विशेष उपयोग के मामलों आप क्योंकि यह एक समान रूप में नहीं है चीजों की रचना करने के कुछ क्षमता खो देते हैं के लिए स्पष्ट समर्थन प्रदान करता है होने से। उदाहरण के लिए, स्ट्रीम और write पर पर push का उपयोग करें, हालांकि, अवधारणात्मक रूप से, आप एक ही काम कर रहे हैं: एक मूल्य प्रकाशित करना।

तो, व्यवहार में, यदि आप अवधारणाओं को देखें, और यदि आप विकल्प { objectMode: true } उपयोग करते हैं, आपका मिलान कर Readable स्ट्रीम और ObserverWritable धारा के साथ साथ Observable। आप दो मॉडलों के बीच कुछ सरल एडाप्टर भी बना सकते हैं।

var Readable = require('stream').Readable; 
var Writable = require('stream').Writable; 
var util = require('util'); 

var Observable = function(subscriber) { 
    this.subscribe = subscriber; 
} 

var Subscription = function(unsubscribe) { 
    this.unsubscribe = unsubscribe; 
} 

Observable.fromReadable = function(readable) { 
    return new Observable(function(observer) { 
     function nop() {}; 

     var nextFn = observer.next ? observer.next.bind(observer) : nop; 
     var returnFn = observer.return ? observer.return.bind(observer) : nop; 
     var throwFn = observer.throw ? observer.throw.bind(observer) : nop; 

     readable.on('data', nextFn); 
     readable.on('end', returnFn); 
     readable.on('error', throwFn); 

     return new Subscription(function() { 
      readable.removeListener('data', nextFn); 
      readable.removeListener('end', returnFn); 
      readable.removeListener('error', throwFn); 
     }); 
    }); 
} 

var Observer = function(handlers) { 
    function nop() {}; 

    this.next = handlers.next || nop; 
    this.return = handlers.return || nop; 
    this.throw = handlers.throw || nop; 
} 

Observer.fromWritable = function(writable, shouldEnd, throwFn) { 
    return new Observer({ 
     next: writable.write.bind(writable), 
     return: shouldEnd ? writable.end.bind(writable) : function() {}, 
     throw: throwFn 
    }); 
} 

आपने ध्यान दिया होगा कि मैं कुछ नाम बदल गया है और Observer और Subscription की सरल अवधारणाओं, यहाँ पेश किया, इस्तेमाल किया Generator में observables द्वारा किया reponsibilities के अधिभार से बचने के लिए। असल में, Subscription आपको Observable से सदस्यता समाप्त करने की अनुमति देता है। वैसे भी, उपर्युक्त कोड के साथ आपके पास pipe हो सकता है।

Observable.fromReadable(process.stdin).subscribe(Observer.fromWritable(process.stdout)); 

process.stdin.pipe(process.stdout) के साथ तुलना में, आप क्या है, गठबंधन फिल्टर, और धाराएं भी डेटा के किसी भी अन्य दृश्य के लिए काम करता है को बदलने के लिए एक तरीका है। आप इसे Readable, Transform, और Writable धाराओं के साथ प्राप्त कर सकते हैं लेकिन एपीआई Readable एस को चेन करने और कार्यों को लागू करने के बजाय उप-वर्गीकरण का पक्ष लेता है। Observable मॉडल पर, उदाहरण के लिए, मूल्यों को बदलने से ट्रांसफॉर्मर फ़ंक्शन को स्ट्रीम में लागू किया जाता है। इसे Transform के नए उप प्रकार की आवश्यकता नहीं है।

Observable.just = function(/*... arguments*/) { 
    var values = arguments; 
    return new Observable(function(observer) { 
     [].forEach.call(values, function(value) { 
      observer.next(value); 
     }); 
     observer.return(); 
     return new Subscription(function() {}); 
    }); 
}; 

Observable.prototype.transform = function(transformer) { 
    var source = this; 
    return new Observable(function(observer) { 
     return source.subscribe({ 
      next: function(v) { 
       observer.next(transformer(v)); 
      }, 
      return: observer.return.bind(observer), 
      throw: observer.throw.bind(observer) 
     }); 
    }); 
}; 

Observable.just(1, 2, 3, 4, 5).transform(JSON.stringify) 
    .subscribe(Observer.fromWritable(process.stdout)) 

निष्कर्ष? प्रतिक्रियाशील मॉडल और Observable अवधारणा को कहीं भी पेश करना आसान है। उस अवधारणा के आस-पास एक संपूर्ण पुस्तकालय को कार्यान्वित करना कठिन है। उन सभी छोटे कार्यों को लगातार एक साथ काम करने की जरूरत है।आखिरकार, ReactiveX प्रोजेक्ट अभी भी इस पर जा रहा है। लेकिन अगर आपको क्लाइंट को फ़ाइल सामग्री भेजने की ज़रूरत है, तो एन्कोडिंग से निपटें, और इसे ज़िप करें, फिर नोडजेएस में इसका समर्थन, और यह बहुत अच्छी तरह से काम करता है।