monixの観測可能な再帰をどのように処理するのですか？

Question

Using monix私はObservable [Node]を構築し、幅広い最初のアルゴリズムを使ってグラフをたどろうとしています。 しかし、そこに私は再帰問題のビットがあります。ここに私の問題を示すスニペットがあります：

package gp 

import monix.eval.Task 
import monix.execution.Scheduler.Implicits.global 
import monix.reactive._ 


object HelloObservable { 

    type Node = Int 

    //real case fetch next node across the network so the signature 
    //has to be Node -> List[Task[Node]] 
    def nexts(i : Node) : List[Task[Node]] = 
    List(Task(i), Task(i+1)) 

    def go(i :Node) : Task[Iterator[List[Node]]] = 
    Task.sequence(nexts(i).sliding(100,100).map(Task.gatherUnordered)) 

    def explore(r: Node): Observable[Node] = { 
    val firsts = for { 
     ilr <- Observable.fromTask(go(r)) 
     lr <- Observable.fromIterator(ilr) 
     r <- Observable.fromIterable(lr) 
    } yield r 

    firsts ++ firsts.flatMap(explore) 
    } 


    def main(args : Array[String]) : Unit = { 

    val obs = explore(0) 

    val cancelable = obs 
     .dump("O") 
     .subscribe() 

    scala.io.StdIn.readLine() 

    } 

} 

最初の反復後の観測可能な停止。誰も私をなぜヒントできますか？

Answer 1

私は問題が再帰に関連していないと思います。私はそれがを返すslidingを使用するという事実から来ていると思います。 IteratorとIterableの主な違いは、Iteratorを1回だけ消費することができることです。残りのすべてが空のIteratorです。 firsts.flatMapの場合は、Observable.fromIterator(ilr)に何も残っておらず、何も生産されていません。

基本的には、メモリ内の接頭辞（大部分）を保持できない場合は、幅優先検索を行うことはできません。しかし、あなたのnextsはすでにListを返すので、私はそのリストの2つのコピーをメモリに持たせることができると思います。 2番目のコピーはslidingのマテリアライズされた結果です。固定コードは次のようなものになります。

object HelloObservable { 

    import monix.eval.Task 
    import monix.execution.Scheduler.Implicits.global 
    import monix.reactive._ 

    type Node = Int 

    //real case fetch next node across the network so the signature 
    //has to be Node -> List[Task[Node]] 
    def nexts(i: Node): List[Task[Node]] = List(Task(i), Task(i + 1)) 

    def go(i: Node): Task[List[List[Node]]] = 
     Task.sequence(nexts(i).sliding(100, 100).toList.map(Task.gatherUnordered)) 


    def explore(r: Node): Observable[Node] = { 
     val firsts = for { 
     ilr <- Observable.fromTask(go(r)) 
     lr <- Observable.fromIterable(ilr) 
     r <- Observable.fromIterable(lr) 
     } yield r 
     firsts ++ firsts.flatMap(explore) 
    } 


    def main(args: Array[String]): Unit = { 

     val obs = explore(0) 

     val cancelable = obs 
     .dump("O") 
     .subscribe() 

     scala.io.StdIn.readLine() 

    } 
}