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을 한 번만 사용하면 나머지는 모두 Iterator입니다. 그것은 당신이 firsts.flatMap 일 때 Observable.fromIterator(ilr)에 아무것도 남지 않았으므로 아무 것도 생산되지 않는다는 것을 의미합니다.

기본적으로 메모리에 접두어를 저장할 수 없다면 너비 우선 검색을 수행 할 수 있다고 생각하지 않습니다. 그러나 nexts이 이미 List을 반환하기 때문에 메모리에 해당 목록의 복사본이 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() 

    } 
}