reactive-programming

0熱度

1回答

我正在使用Java Reactor核心，並且我有一個被動對象Flux。對於Flux的每個對象，我需要做一個外部查詢，爲每個輸入返回一個不同的對象。然後新生成的Flux需要與原始的一個壓縮 - 所以2 Flux的項目必須以相同的順序同步和生成。我只是相同的流量的兩倍，這樣重新使用： Flux<MyObj> aTest = Flux.fromIterable(aListOfObj); Flux<

0熱度

1回答

並行Rx Java地圖函數

RxJava和Reactive Programming的新手可以這麼說。我試圖將兩個函數並行映射爲單個Observable管道的一部分，但似乎不能這樣工作。這是我的代碼。 Observable.fromCallable(thatReturnsNumberOne()) .observeOn(newThread()) .map(doubleIt())

0熱度

1回答

如何在Reactor 3中編寫操作符？

我已經通過實施org.reactivestreams.Publisher實施了Reactor操作員，如下所示。但是，我不知道這是否是正確的方法™使用反應堆。手動實現用戶看起來有點麻煩。在這方面Operators類似乎沒有幫助。 class MyOperator implements Publisher<Integer> { private final Publisher<Integer

1熱度

1回答

RxJava：模式爲暫時中止選擇的源觀測量無有狀態變量

在更大RxJava應用程序，我有許多熱，無限源觀測量。這些排放物被合併，然後由下游觀察員處理。然後，處理結果需要用於暫時中止某些源觀測數據的排放，而非懸浮觀測數據預計將繼續排放，觀測者將消耗非暫停的觀測數據。對於任何在暫停期間發生的事件，觀察對象都可以安全地忽略/丟棄。我到目前爲止唯一能夠提出的解決方案是應用具有全局，有狀態變量的過濾器。下面的代碼顯示了原理。爲了簡單起見，我將源Observab

1熱度

1回答

observable forkjoin永遠不會執行可觀察的angular2

我試圖使用forkJoin執行多個XHR請求（在getImages函數中），每個請求都包含在可觀察對象（在getImage函數內）中。然而，沒有任何請求被執行，即從不執行observer.next()。我曾嘗試使用zip而不是forkJoin，但這也不起作用。我確信我可以在我創建的所有觀察對象上都有complete。 getImages(urls: string[]) { Observa

1熱度

1回答

如何對rxjs5進行單元測試？

我讀過這個「大理石串」文檔，但我堅持不明白。我想測試的基本上是一個「協議」。我有一個連接到Observable併發出數據的「東西」。在與Rx.NET和C＃我的後端代碼，我已經寫了一個助手來測試協議，使我的測試是這樣的： verifier.Send(new AddMaterialPickCommand(commandId, orderId, materialId: Guid.NewGuid(),

2熱度

2回答

f＃可觀察到的叉子和副作用

我有一個可觀察的鏈，最初的觀察者來自網絡，並且每當消息準備被讀取時就會被觸發。接下來的處理程序然後讀取消息並將其反序列化。現在我有一個observable分支，一個是消息處理程序，另一個是記錄消息。的問題是，因爲我觀察到的使用我實際上將嘗試讀取消息的兩倍。據我所知，使用事件，而不是可觀察就能解決問題，但我會再有一個垃圾收集問題，可能導致無法收集套接字。一種解決方案我想的是插入某種分離器的其中

0熱度

1回答

如何在RxJava2中默默跳過異常？

我有一個這樣的數據流： Observable .fromFuture( CompletableFuture.supplyAsync { // First remote call returns Future<List<Type>> listOf(1, 2, 3, 57005, 5) }, Schedulers.computation()

1熱度

1回答

無效編程規範化時間序列值

我有一個遠程程序，它通過套接字連接每10毫秒發送一次更新的度量。在我的客戶端程序中，我將這個套接字包裝在一個可產生這些測量結果的觀察值中對於我的用例來說，測量到達10毫秒間隔很重要。當然，這種情況不會發生，因爲網絡延遲會使信息早或晚地到達每條信息。所以基本上我在遠程PC上有一個程序在套接字連接上發送它。 --爲10毫秒 o--o--o--o--o--o--o--o--o--o--... 這成

2熱度

1回答

Swift 3：轉換PromiseKit推遲到RxSwift

我目前正在用RxSwift取代PromiseKit，並且需要將我的推遲承諾轉換爲RxSwift。在PromiseKit 當前實現的例子： private var deferredDidLayout = Promise<()>.pending() override func layoutSubviews() { super.layoutSubviews() self.def