如何加入兩個半在RxJs

Question

我有這樣

---ab---ab---a---ba---bab---ab---ab---ab---> 

流，我想這一點。

---ab---ab------ab----ab-ab-ab---ab---ab---> 

的一點是，我有開始和結束（JSON），有時數據被切斷流中的一半數據，我想再次加入他們的行列。我怎樣才能做到這一點？

Answer 1

這是我如何解決：

import Rx from 'rxjs/Rx'; 
import {last} from 'lodash'; 

const data$ = Rx.Observable.of('ab','ab','a','ba','bab','aba','b','ab'); 
const line$ = data$.flatMap(data => { 
    const lines = data.match(/[^b]+b?|b/g); // https://stackoverflow.com/a/36465144/598280 https://stackoverflow.com/a/25221523/598280 
    return Rx.Observable.from(lines); 
}); 

const isComplete$ = line$.scan((acc, value) => { 
    const isLineEndingLast = last(acc.value) === 'b'; 
    const id = isLineEndingLast ? acc.id + 1 : acc.id; 
    const complete = last(value) === 'b'; 
    return {value, id, complete}; 
}, {value: 'b', id: 0, complete: true}); 

const grouped$ = isComplete$ 
    .groupBy(data => data.id, data => data, group => group.first(data => data.complete)) 
    .flatMap(group => group.reduce((acc, data) => acc + data.value, '')); 

grouped$.subscribe(console.log); 

Answer 2

看起來像掃描操作

// substitute appropriate real-world logic 
const isProperlyFormed = (x) => x === 'ab' 
const isIncomplete = (x) => x[0] === 'a' && x.length === 1 
const startsWithEnding = (x) => x[0] === 'b' 
const getCorrected = (buffer, x) => buffer.prev + x[0] 
const getTail = (buffer, x) => x.slice(1) 

const initialBuffer = { 
    emit: [], 
    prev: null 
} 

const result = source 
    .scan((buffer, x) => { 
    if (isProperlyFormed(x)) { 
     buffer = {emit: [x], prev:null} 
    } 
    if (isIncomplete(x)) { 
     buffer = {emit: [], prev:x} 
    } 
    if (startsWithEnding(x)) { 
     const corrected = getCorrected(buffer, x) 
     const tail = getTail(buffer, x) 
     if (isProperlyFormed(tail)) { 
     buffer = {emit: [corrected, tail], prev: null} 
     } else { 
     buffer = {emit: [corrected], prev: tail} 
     } 
    } 
    return buffer 
    }, initialBuffer) 
    .flatMap(x => x.emit) 

工作CodePen

編輯

在測試輸入流尋找一份工作，我想的情況下丟失，這將打破以上。

我改變了測試從

---ab---ab---a---ba---bab---ab---ab---ab---> 

到

---ab---ab---a---ba---bab---aba---b---ab---> 

，並精簡下來的算法

const getNextBuffer = (x) => { 
    const items = x.split(/(ab)/g).filter(y => y) // get valid items plus tail 
    return { 
    emit: items.filter(x => x === 'ab'), // emit valid items 
    save: items.filter(x => x !== 'ab')[0] // save tail 
    } 
} 

const initialBuffer = { 
    emit: [], 
    save: null 
} 

const result = source 
    .scan((buffer, item) => { 
    const bufferAndItem = (buffer.save ? buffer.save : '') + item 
    return getNextBuffer(bufferAndItem) 
    }, initialBuffer) 
    .flatMap(x => x.emit) 

工作實例CodePen

Answer 3

網絡首先將流分成完全響應和部分響應。然後檢查響應是否已滿。完整的迴應是很好的。部分響應需要同步，因此我們將它們的流分成前半部分和後半部分，並將這些流壓縮在一起。

奇怪的外觀Rx.Observable.of(g.partition(x => x[0] === 'a'))是因爲partition運算符返回一對無法鏈接的observables。

const testStream = Rx.Observable.of('a1', 'a2', '_ab', 'b1', 'a3', 'b2', '_ab', 'a4', 'b3', '_ab', 'b4', 'a5', 'b5', '_ab') 
 

 
testStream 
 
    .groupBy(x => (x[0] === '_' && 'full') || 'partial') 
 
    .mergeMap(g => 
 
    Rx.Observable.if(
 
    () => g.key == 'full', 
 
     g, 
 
     Rx.Observable.of(g.partition(x => x[0] === 'a')) 
 
     .mergeMap(([as, bs]) => Rx.Observable.zip(as, bs)) 
 
    ) 
 
) 
 
    .do(x => console.log(x)) 
 
    .subscribe()

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.3/Rx.min.js"></script>