RxJS, एक गतिशील टाइमस्टैम्प

Question

मैं RxJS के लिए नया हूँ और मैं एक ऐप्लिकेशन है जो निम्नलिखित चीजों को पूरा करेगा लिखने के लिए कोशिश कर रहा हूँ का उपयोग कर लगातार अपडेट की रिकॉर्ड के लिए जाँच करने के लिए एक API मतदान करने के लिए कैसे: प्रस्तुत होने पर

1. , बनाना वस्तुओं की एक सूची लाने के लिए एक AJAX अनुरोध (सादगी के लिए fetchItems() के रूप में faked)।
2. इसके बाद हर दूसरे, आइटम प्राप्त करने के लिए AJAX अनुरोध करें।
3. नई वस्तुओं की जांच करते समय, सबसे हालिया टाइमस्टैम्प को वापस करने के बाद केवल आइटम बदल दिए जाते हैं।
4. अवलोकनों के बाहर कोई भी राज्य नहीं होना चाहिए।

मेरे first attempt बहुत सीधे आगे था और लक्ष्यों को 1, 2 और 4.

var data$ = Rx.Observable.interval(1000) 
    .startWith('run right away') 
    .map(function() { 
    // `fetchItems(modifiedSince)` returns an array of items modified after `modifiedSince`, but 
    // I am not yet tracking the `modifiedSince` timestamp yet so all items will always be returned 
    return fetchItems(); 
    }); 

अब मैं उत्साहित हूँ से मुलाकात की, कि आसान था, यह है कि बहुत कठिन नहीं किया जा सकता लक्ष्य पूरा करने के लिए 3 ... कई घंटे बाद में इस where I am at है:

var modifiedSince = null; 
var data$ = Rx.Observable.interval(1000) 
    .startWith('run right away') 
    .flatMap(function() { 
    // `fetchItems(modifiedSince)` returns an array of items modified after `modifiedSince` 
    return fetchItems(modifiedSince); 
    }) 
    .do(function(item) { 
    if(item.updatedAt > modifiedSince) { 
     modifiedSince = item.updatedAt; 
    } 
    }) 
    .scan(function(previous, current) { 
    previous.push(current); 
    return previous; 
    }, []); 

इस लक्ष्य 3 को हल करती है, लेकिन लक्ष्य 4. मैं अब नमूदार के बाहर राज्य भंडारण कर रहा हूँ पर regresses।

मुझे लगता है कि वैश्विक modifiedSince और .do() ब्लॉक इसे पूरा करने का सबसे अच्छा तरीका नहीं है। किसी भी मार्गदर्शन की काफी सराहना की जाएगी।

संपादित करें: उम्मीद है कि मैं इस प्रश्न के साथ क्या देख रहा हूं।

Answer 1

यहां एक और समाधान है जो बंद या 'बाहरी राज्य' का उपयोग नहीं करता है।

मैं निम्नलिखित परिकल्पना बनाया:

• fetchItems रिटर्न मदों की एक Rx.Observable, मदों की एक सरणी मतलब यह कि

यह जो मूल्यों जो एक का पालन करें फेंकना करने की अनुमति देता expand ऑपरेटर का उपयोग करता है x_n+1 = f(x_n) प्रकार के पुनरावर्ती संबंध। आप x_n+1 को उस अवलोकन योग्य लौटकर पास करते हैं जो उस मान को उत्सर्जित करता है, उदाहरण के लिए Rx.Observable.return(x_n+1) और आप Rx.Observable.empty() लौटकर रिकर्सन को समाप्त कर सकते हैं। यहां ऐसा लगता है कि आपके पास कोई स्थायी स्थिति नहीं है इसलिए यह हमेशा के लिए चलेगा।

scan भी एक पुनरावर्ती संबंध (x_n+1 = f(x_n, y_n)) के बाद मूल्यों को उत्सर्जित करने की अनुमति देता है। अंतर यह है कि scan आपको एक सिंक्रोनस फ़ंक्शन का उपयोग करने के लिए मजबूर करता है (इसलिए x_n+1y_n के साथ सिंक्रनाइज़ किया गया है), जबकि expand के साथ आप एक अवलोकन के रूप में एक एसिंक्रोनस फ़ंक्शन का उपयोग कर सकते हैं।

कोड का परीक्षण नहीं किया गया है, इसलिए यदि यह काम करता है या नहीं तो मुझे अपडेट रखें।

प्रासंगिक दस्तावेज: expand, combineLatest

var modifiedSinceInitValue = // put your date here 
var polling_frequency = // put your value here 
var initial_state = {modifiedSince: modifiedSinceInitValue, itemArray : []} 
function max(property) { 
    return function (acc, current) { 
    acc = current[property] > acc ? current[property] : acc; 
    } 
}  
var data$ = Rx.Observable.return(initial_state) 
    .expand (function(state){ 
      return fetchItem(state.modifiedSince) 
        .toArray() 
        .combineLatest(Rx.Observable.interval(polling_frequency).take(1), 
        function (itemArray, _) { 
         return { 
         modifiedSince : itemArray.reduce(max('updatedAt'), modifiedSinceInitValue), 
         itemArray : itemArray 
         } 
        } 

    }) 

Answer 2

इस बारे में कैसे:

var interval = 1000; 
function fetchItems() { 
    return items; 
} 

var data$ = Rx.Observable.interval(interval) 
    .map(function() { return fetchItems(); }) 
    .filter(function(x) {return x.lastModified > Date.now() - interval} 
    .skip(1) 
    .startWith(fetchItems()); 

यह केवल नए आइटम के लिए स्रोत फ़िल्टर करना चाहिए, साथ ही आप पूरा संग्रह के साथ शुरू। अपने डेटा स्रोत के लिए उपयुक्त होने के लिए केवल फ़िल्टर फ़ंक्शन लिखें।

या fetchItems लिए एक तर्क पारित करके:

var interval = 1000; 
function fetchItems(modifiedSince) { 
    var retVal = modifiedSince ? items.filter(function(x) {return x.lastModified > modifiedSince}) : items 
    return retVal; 
} 

var data$ = Rx.Observable.interval(interval) 
    .map(function() { return fetchItems(Date.now() - interval); }) 
    .skip(1) 
    .startWith(fetchItems()); 

Answer 3

तुम्हारा मतलब है कि modifiedSince राज्य आप ले जाने का हिस्सा है लग रहे हैं, तो यह scan में दिखाई देनी चाहिए। आप स्कैन में do में कार्रवाई को क्यों नहीं ले जाते? तब आपका बीज {modifiedSince: null, itemArray: []} होगा।

एरर, मैंने सोचा कि यह काम नहीं कर सकता है, क्योंकि आपको modifiedSince को fetchItem फ़ंक्शन को अपस्ट्रीम पर फ़ीड करने की आवश्यकता है। क्या आपके पास यहां कोई चक्र नहीं है? इसका मतलब है कि आपको चक्र को तोड़ने के लिए किसी विषय का उपयोग करना होगा। वैकल्पिक रूप से आप एक बंद में encapsulated modifiedSince रखने की कोशिश कर सकते हैं। जैसे

function pollItems (fetchItems, polling_frequency) { 
var modifiedSince = null; 
var data$ = Rx.Observable.interval(polling_frequency) 
    .startWith('run right away') 
    .flatMap(function() { 
    // `fetchItems(modifiedSince)` returns an array of items modified after `modifiedSince` 
    return fetchItems(modifiedSince); 
    }) 
    .do(function(item) { 
    if(item.updatedAt > modifiedSince) { 
     modifiedSince = item.updatedAt; 
    } 
    }) 
    .scan(function(previous, current) { 
    previous.push(current); 
    return previous; 
    }, []); 

return data$; 
} 

मैं नए साल का जश्न मनाने के लिए, कि अगर काम नहीं करता है बाहर चलाने के लिए है कुछ है, मैं (शायद expand ऑपरेटर, scan के अन्य संस्करण का उपयोग कर) बाद में एक और आज़मा सकें।