Observable.unfold

[pk044]

Solves #955

i have a few doubts though:

• i'm afraid the code is duplicated, it differs very little from fromStateAction. i was thinking of extracting the common parts somewhere and try to reuse them in both functions - not to mention that more unfold-based functions are planned to be implemented in Monix (unfoldEval, unfoldEvalF), is there perhaps a generic way to handle this?
• i tried to write a property based test, i'm not too familiar with it and I kind of gave up after having the test go out of memory :)

  test("expected not to hang and throw OOM errors but it does") { implicit s =>
    check3 { (seed: Int, f: Int => Option[(Int, Int)], i: Int) =>
      val n = i % (recommendedBatchSize * 2)

      val received = Observable.unfold(seed)(f).takeWhile(_ < n).toListL
      val expected = Task.now((seed to n).toList)

      received <-> expected
    }
  }

after dabbling with it for a little while it looks like it hangs because the size of the generated list (seed to n toList) might be a little too big. I think I'm supposed to have that list in the parameters, and come up with some sort of law so that ScalaCheck would always generate the expected list, instead of letting me doing it myself. (not sure if that was clear enough)

do you guys perhaps have any suggestions for any of the above? thanks!

[Avasil]

We could copy few tests from StateActionObservableSuite - especially to do things like:

val cancelable = Observable
    .fromStateAction(int)(s.clockMonotonic(MILLISECONDS))
    .unsafeSubscribeFn(new Subscriber[Int] {
      implicit val scheduler = s
      def onNext(elem: Int) = {
        sum += 1
        Continue
      }

      def onComplete() = wasCompleted = true
      def onError(ex: Throwable) = wasCompleted = true
})

to see if Observable completed. I think it could fail currently. Alternatively we can call methods like sumL or completedL to complete it in nicer way and execute as a Task. subscribe just processes it asynchronously in the background

Also I'd suggest naming the suite to UnfoldObservableSuite - we have also Iterant and if we add unfold for it then it can be a bit confusing to search for

[pk044]

i added the test with subscribe for now, I tried to write it like this:

test("should be cancelable") { implicit s =>
    val cancelable = Observable
      .unfold(s.clockMonotonic(MILLISECONDS))(intOption)
        .sumL
        .runToFuture

    cancelable.cancel()

    s.tick()

    assertEquals(cancelable.value, Some(Success(s.executionModel.recommendedBatchSize * 2)))
  }

but it returns None(while the subscribe one works) - i need to dig deeper and find out why

[Avasil]

Thank you @pk044 - great job overall! I've left some comments.

• i'm afraid the code is duplicated, it differs very little from fromStateAction. i was thinking of extracting the common parts somewhere and try to reuse them in both functions - not to mention that more unfold-based functions are planned to be implemented in Monix (unfoldEval, unfoldEvalF), is there perhaps a generic way to handle this?

It's fine to have duplicated code, we often do it in purpose for a little gain in efficiency. I wouldn't do it in a "normal" code base but we have an excuse. :)

• i tried to write a property based test, i'm not too familiar with it and I kind of gave up after having the test go out of memory :)

  test("expected not to hang and throw OOM errors but it does") { implicit s =>
    check3 { (seed: Int, f: Int => Option[(Int, Int)], i: Int) =>
      val n = i % (recommendedBatchSize * 2)

      val received = Observable.unfold(seed)(f).takeWhile(_ < n).toListL
      val expected = Task.now((seed to n).toList)

      received <-> expected
    }
  }

I think seed to n could be potentially -2147483648 to n

BTW It would be cool if we can come up with a property that tests Observable.unfold(seed)(f).toListL <-> Observable.fromStateAction(f)(seed).takeWhile(p).toListL instead

[pk044]

@Avasil thanks! i'll see what I can do about the property test :)

[pk044]

@Avasil i'm not too familiar with property-based testing yet, I came up with this so far:

  test("unfold and fromStateAction lists should be equal given generated inputs") { implicit s =>
    check {
      forAll(Gen.choose(0, Platform.recommendedBatchSize * 2), Gen.choose(0, Platform.recommendedBatchSize * 2)) {
        (seed, max) =>
          val f: Int => Option[(Int, Int)] = i => if (i < max) Some((i, i + 1)) else None
          val f2: Int => (Int, Int) = i => (i, i + 1)

          Observable.unfold(seed)(f).toListL <-> Observable.fromStateAction(f2)(seed).takeWhile(_ < max).toListL
      }
    }
  }

is it what we wanted more or less?

[Avasil]

I think the test is good but I'd recommend doing just check2 and val n = i % (recommendedBatchSize * 2) like in many Iterant tests, it's a bit less noise when reading tests

[pk044]

so basically something like

    check2 { (s: Int, i: Int) =>
          val seed = s % (recommendedBatchSize * 2)
          val n = i % (recommendedBatchSize * 2)

? the test would hang if i used the s variable directly, probably because the generated number is way too big

[Avasil]

I feel like seed should be fine but there's no harm in doing it this way 👍