Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
testClock provided to eventuallyZ
package com.wix.delay
import com.wix.testkit.eventuallyZ
import org.specs2.mutable.SpecificationWithJUnit
import zio._
import zio.duration._
import zio.test.environment.TestClock
class DelayedConsumerTestIT extends SpecificationWithJUnit with BootstrapRuntime {
"consume message with delay" in {
run(
for {
callCount <- Ref.make[Int](0)
messageHandler = doSomething(callCount)
_ <- startConsumer(Handler.create(messageHandler), delay = 1.second)
_ <- producer.produce(aMessage())
_ <- TestClock.adjust(1100.millis)
_ <- eventuallyZ(callCount.get)(_ == 1)
} yield ok
)
}
def eventuallyZ[T](f: UIO[T])(predicate: T => Boolean): ZIO[Clock, Throwable, Unit] = {
f.repeat(Schedule.spaced(100.milliseconds) && Schedule.doUntil(predicate))
.timeoutFail(new RuntimeException)(4.seconds)
.unit
}
def env: UManaged[ZEnv with TestClock] = test.environment.testEnvironment.build
def run[E, A](zio: ZIO[ZEnv with TestClock, E, A]): A =
unsafeRun(env.use(zio.provide))
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.