rx-scala

0熱

2答えて

を放射されたX項目よりも少ない場合、私はOBS1、obs2、obs3、...、それらのそれぞれ（のMongoDBデータベースからの）項目の数を発することができる観測のリストを持っているスイッチは、私が興味を持っています最初のN個のアイテムにのみ表示されます。私は、必要に応じてオブザーバブルのクエリが実行されるようにしたいと思います。つまり、たとえばobs1がNより大きい値を生成した場合、obs2

1熱

1答えて

RxScalaでサブスクリプションを管理するベストプラクティスは何ですか？

私はこの use subscription = Observable.subscribe ... を行うだろうRxの/ F＃の私のサブスクリプションためVARを使用して回避しながら、私のサブスクリプションのためにも例外の存在下での範囲を制御したいとサブスクリプションがありますスコープを離れるときに自動的に処理されます。 Scalaのベストプラクティスは何ですか？

0熱

2答えて

RxJava/RxScalaでobservableをキャッシュする方法

私は、RxScalaに基づいて反応性スカラドライバを使用してmongodbを呼び出すREST APIを持っています。私のAPIコントローラまたはサービス層で、私はhazelcast（またはその他のキャッシュツール）を使用して、MongoDBのために呼び出しを避けるために、キャッシュを使用する必要がすべての私のサービスは非同期であり、唯一、観察を返し、どのように私ができる任意のアイデア観察可能な

1熱

1答えて

TCPソケットを観測可能にする[Byte]

私のAndroidアプリでは、Socketを使用してバイト配列を送受信する必要があります。便宜上、私はSocketに接続されたObservableで作業したいと思っています。私はこのコードを発見した、インターネット上で探している相手：「こんにちは」の文字列を送信したとき、出力され、たとえば、 import rx.lang.scala.Observable val s = Observable

1熱

1答えて

RxScala ConnectableObservable.doOnSubscribeは決して呼び出されませんか？

val observable = Observable(...) .publish val subscription = observable.connect observable.doOnsubscribe(() => doSomething) .subscribe() doSomethingは決して呼び出されません。 RxJavaとまったく同じコードが正しく動作し

0熱

1答えて

oncompleteのみが呼び出されたときのscala rx observables上のマッピング

私はscala observablesをcouchbaseから取得するためにmap、flatMap、zipを使用して結果を変換しています。問題は、アイテムがcouchbaseに存在しない場合、たとえば.zipはonCompleteだけ呼び出されないことです。例： import rx.lang.scala._ def getIdsWithValues(ids: Seq[String]): Map

0熱

2答えて

RxScala "zip"複数オブザーバブル

RxScalaでは、2つ以上のオブザーバブルを "ジップ"する方法は？ val ob1: Observable[Int] = Observable.from(Future(10)) val ob2: Observable[Int] = Observable.from(Future(20)) val ob3: Observable[Int] = Observable.from(Future(30

0熱

1答えて

RxJava/RxScalaのgroupByとflatMap（maxConcurrent、...）を組み合わせる

私は着信処理要求を持っていますが、共有リソースが枯渇して同時に処理したくないと思っています。私はまた、いくつかのユニークな鍵を共有する要求が同時に実行されないことを好むだろう：キーごとに観察できるが、決して完了しないため def process(request: Request): Observable[Answer] = ??? requestsStream .groupBy(re

0熱

1答えて

Akka Streams - 最新の操作を結合します

私は最新のAkka Streamsをhereのように組み合わせたいと思います。私はそれを行う方法を理解できません - 助けてください！ありがとう、 Ryan。

2熱

1答えて

Observableには無限のメモリが必要ですか？

各Observableには最初に発行された値までトレースできるキャッシュがあるので、このキャッシュを格納するために使用されるメモリの量は制限されていないようです。 Observable.interval(1.microsecond).map(_ => System.currentTimeMillis) .subscribe(x =>()) をアプリの実行中に実際にメモリ使用量が着実に