make the "close connection on monix task *" tests generic

Author: halotukozak

rest/jetty/src/test/scala/io/udash/rest/jetty/CloseStaleJettyConnectionsOnMonixTimeout.scala

@@ -6,7 +6,9 @@ import io.udash.rest.{RestApiTestScenarios, ServletBasedRestApiTest}
 import org.eclipse.jetty.client.HttpClient
 
 final class JettyRestCallTest extends ServletBasedRestApiTest with RestApiTestScenarios {
-  val client: HttpClient = new HttpClient
+  val client: HttpClient = new HttpClient() {
+    setMaxConnectionsPerDestination(MaxConnections)
+  }
 
   def clientHandle: HandleRequest =
     JettyRestClient.asHandleRequest(client, s"$baseUrl/api", maxPayloadSize)

rest/jetty/src/test/scala/io/udash/rest/jetty/JettyRestCallTest.scala

@@ -1,15 +1,25 @@
 package io.udash
 package rest
 
-import com.avsystem.commons._
+import com.avsystem.commons.*
+import com.avsystem.commons.misc.ScalaDurationExtensions.durationIntOps
 import io.udash.rest.raw.RawRest
 import io.udash.rest.raw.RawRest.HandleRequest
+import monix.eval.Task
 import monix.execution.Scheduler
 import org.scalactic.source.Position
+import org.scalatest.concurrent.PatienceConfiguration.Timeout
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.funsuite.AnyFunSuite
 
+import scala.concurrent.duration.FiniteDuration
+
 abstract class RestApiTest extends AnyFunSuite with ScalaFutures {
+
+  protected final val MaxConnections: Int = 1 // to timeout quickly
+  protected final val Connections: Int = 10 // > MaxConnections
+  protected final val CallTimeout: FiniteDuration = 300.millis // << idle timeout
+
   implicit def scheduler: Scheduler = Scheduler.global
 
   final val serverHandle: RawRest.HandleRequest =
@@ -30,6 +40,9 @@ abstract class RestApiTest extends AnyFunSuite with ScalaFutures {
     case arr: Array[_] => IArraySeq.empty[AnyRef] ++ arr.iterator.map(mkDeep)
     case _ => value
   }
+
+  def getNeverGetCounter(): Int = RestTestApi.Impl.neverGetCounter.get()
+  def resetNeverGetCounter(): Unit = RestTestApi.Impl.neverGetCounter.set(0)
 }
 
 trait RestApiTestScenarios extends RestApiTest {
@@ -89,6 +102,29 @@ trait RestApiTestScenarios extends RestApiTest {
   test("body using third party type") {
     testCall(_.thirdPartyBody(HasThirdParty(ThirdParty(5))))
   }
+
+  test("close connection on monix task timeout") {
+    resetNeverGetCounter()
+    Task
+      .traverse(List.range(0, Connections))(_ => Task.deferFuture(proxy.neverGet).timeout(CallTimeout).failed)
+      .map(_ => assertResult(expected = Connections)(actual = getNeverGetCounter())) // neverGet should be called Connections times
+      .runToFuture
+      .futureValue(Timeout(30.seconds))
+  }
+
+  test("close connection on monix task cancellation") {
+    resetNeverGetCounter()
+    Task
+      .traverse(List.range(0, Connections)) { i =>
+        val cancelable = Task.deferFuture(proxy.neverGet).runAsync(_ => ())
+        Task.sleep(100.millis)
+          .restartUntil(_ => getNeverGetCounter() >= i)
+          .map(_ => cancelable.cancel())
+      }
+      .map(_ => assertResult(expected = Connections)(actual = getNeverGetCounter())) // neverGet should be called Connections times
+      .runToFuture
+      .futureValue(Timeout(30.seconds))
+  }
 }
 
 class DirectRestApiTest extends RestApiTestScenarios {

rest/src/test/scala/io/udash/rest/RestApiTest.scala

@@ -9,6 +9,7 @@ import com.avsystem.commons.serialization.json.JsonStringOutput
 import io.udash.rest.openapi.adjusters.*
 import io.udash.rest.openapi.{Header as OASHeader, *}
 import io.udash.rest.raw.*
+import monix.execution.atomic.Atomic
 import monix.execution.{FutureUtils, Scheduler}
 
 import scala.concurrent.Future
@@ -94,6 +95,8 @@ case class ErrorWrapper[T](error: T)
 object ErrorWrapper extends HasPolyGenCodec[ErrorWrapper]
 
 trait RestTestApi {
+  final val neverGetCounter = Atomic(0)
+
   @GET @group("TrivialGroup") def trivialGet: Future[Unit]
   @GET @group("TrivialDescribedGroup") @tagDescription("something") def failingGet: Future[Unit]
   @GET def jsonFailingGet: Future[Unit]
@@ -181,7 +184,10 @@ object RestTestApi extends DefaultRestApiCompanion[RestTestApi] {
     def failingGet: Future[Unit] = Future.failed(HttpErrorException.plain(503, "nie"))
     def jsonFailingGet: Future[Unit] = Future.failed(HttpErrorException(503, HttpBody.json(JsonValue(JsonStringOutput.write(ErrorWrapper("nie"))))))
     def moreFailingGet: Future[Unit] = throw HttpErrorException.plain(503, "nie")
-    def neverGet: Future[Unit] = Future.never
+    def neverGet: Future[Unit] = {
+      neverGetCounter.transform(_ + 1)
+      Future.never
+    }
     def wait(millis: Int): Future[String] = FutureUtils.delayedResult(millis.millis)(s"waited $millis ms")
     def getEntity(id: RestEntityId): Future[RestEntity] = Future.successful(RestEntity(id, s"${id.value}-name"))
     def complexGet(p1: Int, p2: String, h1: Int, h2: String, q1: Int, q2: String, q3: Opt[Int], c1: Int, c2: String): Future[RestEntity] =