Iterant.resource(...).repeat doesn't release any resources until the very end

[dylemma]

I'm not sure if this is intentional, but if I construct an Iterant via the .resource helper, then .repeat it, it will repeatedly allocate the resource, then when the stream is done, it will deallocate all of the resources. I expected it to release the allocated resource each time a new "repeat" iteration started.

A minimal example:

val nums = Iterant.resource(Task{ println("Acquire") })(_ => Task{ println("Release") })
  .flatMap(_ => Iterant[Task].of(1,2,3)

If I run nums.repeat.take(9).foreach(println) I get:

Acquire
1
2
3
Acquire
1
2
3
Acquire
1
2
3
Release
Release
Release

But if I run (nums ++ nums ++ nums).foreach(println) I get what I was expecting from the "repeat" version:

Acquire
1
2
3
Release
Acquire
1
2
3
Release
Acquire
1
2
3
Release

I'm using Monix 3.0.0-RC2

[oleg-py]

Yep, current implementation is broken.

The naive implementation of repeat via concat (= this ++ repeat) mostly works, except it has issues withNextCursor nodes that aren't created in RT fashion. If I have a cursor of 30 elements and repeat it with this implementation, I get a nonterminating stream of 30 elements, and doing take on it with n > 30 will still never terminate.

And that is probably okay. Because from effectful stream being empty on first evaluation, doesn't follow that it would be empty on the next one. I.e., with current repeat this test fails:

  test("Iterant.repeat properly repeats evaluation") { _ =>
    val I = Iterant[Coeval]
    val sample = I.suspend {
      var empty = true
      I.suspend {
        if (empty) I.eval { empty = false; 0 }.drop(1)
        else I.pure(42)
      }.repeat
    }
    assertEquals(sample.take(3).toListL.value(), List(42, 42, 42))
  }

This is directly at odds with this test:

  test("Iterant.repeat terminates if the source is empty") { implicit s =>
    val source1 = Iterant[Coeval].empty[Int]
    val source2 = Iterant[Coeval].suspendS(Coeval(source1))
    val source3 = Iterant[Coeval].nextCursorS[Int](BatchCursor(), Coeval(source2))
    val source4 = Iterant[Coeval].nextBatchS[Int](Batch.empty[Int], Coeval(source3))

    assertEquals(source1.repeat.toListL.value(), List.empty[Int])
    assertEquals(source2.repeat.toListL.value(), List.empty[Int])
    assertEquals(source3.repeat.toListL.value(), List.empty[Int])
    assertEquals(source4.repeat.toListL.value(), List.empty[Int])
  }

because once the tail is in F, we have to assume it might eventually become non-empty. That would not apply if repeat would only perform effects once and memoize the results, but it doesn't.

@alexandru do you agree with my logic here? Any objections to removing non-emptiness assumption from .repeat?

[alexandru]

I fixed this in PR — #965

@oleg-py I'm also proposing a new retryIfEmpty operator for your use-case.