Genericise ioRetrying

Author: bcarter97

src/test/scala/base/AsyncIntSpec.scala

@@ -1,13 +1,30 @@
 package base
 
+import cats.effect.Async
+import cats.effect.std.Dispatcher
 import cats.effect.testing.scalatest.AsyncIOSpec
+import cats.syntax.all.*
+import org.scalactic.source.Position
 import org.scalatest.OptionValues
 import org.scalatest.concurrent.Eventually
+import org.scalatest.enablers.Retrying
 import org.scalatest.matchers.should.Matchers
+import org.scalatest.time.Span
 import org.scalatest.wordspec.AsyncWordSpec
 
 import scala.concurrent.duration.*
 
-trait AsyncIntSpec extends AsyncWordSpec with AsyncIOSpec with Matchers with OptionValues with Eventually {
+trait AsyncIntSpec[F[_]] extends AsyncWordSpec with AsyncIOSpec with Matchers with OptionValues with Eventually {
   override implicit val patienceConfig: PatienceConfig = PatienceConfig(30.seconds, 500.millis)
+
+  implicit def fRetrying[T](implicit F: Async[F]): Retrying[F[T]] = new Retrying[F[T]] {
+    override def retry(timeout: Span, interval: Span, pos: Position)(fun: => F[T]): F[T] =
+      Dispatcher.sequential[F].use { dispatcher =>
+        F.fromFuture(
+          F.executionContext.map(
+            Retrying.retryingNatureOfFutureT[T](_).retry(timeout, interval, pos)(dispatcher.unsafeToFuture(fun))
+          )
+        )
+      }
+  }
 }

src/test/scala/base/KafkaSpecBase.scala

@@ -2,4 +2,4 @@ package base
 
 import utils.{EmbeddedKafka, KafkaHelpers}
 
-abstract class KafkaSpecBase[F[_]] extends AsyncIntSpec with KafkaHelpers[F] with EmbeddedKafka[F]
+abstract class KafkaSpecBase[F[_]] extends AsyncIntSpec[F] with KafkaHelpers[F] with EmbeddedKafka[F]

src/test/scala/utils/KafkaHelpers.scala

@@ -11,15 +11,14 @@ import fs2.kafka.{AutoOffsetReset, ConsumerRecord, ConsumerSettings, KafkaConsum
 import io.github.embeddedkafka.EmbeddedKafkaConfig
 import org.apache.kafka.common.TopicPartition
 import org.scalatest.Assertion
-import org.scalatest.enablers.Retrying
 import org.typelevel.log4cats.LoggerFactory
 import org.typelevel.log4cats.slf4j.Slf4jFactory
 import uk.sky.fs2.kafka.topicloader.{LoadTopicStrategy, TopicLoader}
 
 import scala.concurrent.duration.*
 
 trait KafkaHelpers[F[_]] {
-  self: AsyncIntSpec & EmbeddedKafka[F] =>
+  self: AsyncIntSpec[F] & EmbeddedKafka[F] =>
 
   val groupId    = "test-consumer-group"
   val testTopic1 = "load-state-topic-1"
@@ -95,14 +94,14 @@ trait KafkaHelpers[F[_]] {
   def publishToKafkaAndWaitForCompaction(
       partitions: NonEmptySet[TopicPartition],
       messages: Seq[(String, String)]
-  )(implicit kafkaConfig: EmbeddedKafkaConfig, F: Async[F], retrying: Retrying[F[Assertion]]): F[Unit] = for {
+  )(implicit kafkaConfig: EmbeddedKafkaConfig, F: Async[F]): F[Unit] = for {
     _ <- publishToKafkaAndTriggerCompaction(partitions, messages)
     _ <- waitForCompaction(partitions)
   } yield ()
 
   def waitForCompaction(
       partitions: NonEmptySet[TopicPartition]
-  )(implicit kafkaConfig: EmbeddedKafkaConfig, F: Async[F], retrying: Retrying[F[Assertion]]): F[Assertion] =
+  )(implicit kafkaConfig: EmbeddedKafkaConfig, F: Async[F]): F[Assertion] =
     consumeEventually(partitions) { r =>
       for {
         records    <- r
@@ -118,7 +117,7 @@ trait KafkaHelpers[F[_]] {
       groupId: String = UUID.randomUUID().toString
   )(
       f: F[List[(String, String)]] => F[Assertion]
-  )(implicit kafkaConfig: EmbeddedKafkaConfig, F: Async[F], retrying: Retrying[F[Assertion]]): F[Assertion] =
+  )(implicit kafkaConfig: EmbeddedKafkaConfig, F: Async[F]): F[Assertion] =
     eventually {
       val records = withAssignedConsumer[F[List[ConsumerRecord[String, String]]]](
         autoCommit = false,