Also call the listener backend when a request is cancelled (#2679)

Closes #2669

Author: adamw

core/src/main/scala/sttp/client4/listener/ListenerBackend.scala

@@ -6,6 +6,7 @@ import sttp.capabilities.Effect
 import sttp.client4.wrappers.DelegateBackend
 import sttp.shared.Identity
 import java.util.concurrent.atomic.AtomicBoolean
+import scala.util.control.NoStackTrace
 
 /** A backend wrapper which notifies the given [[RequestListener]] when a request starts and completes. */
 abstract class ListenerBackend[F[_], P, L](
@@ -15,20 +16,39 @@ abstract class ListenerBackend[F[_], P, L](
   override def send[T](request: GenericRequest[T, P with Effect[F]]): F[Response[T]] =
     listener.before(request).flatMap { case tag =>
       val onBodyReceivedCalled = new AtomicBoolean
+      // #2669. It would be best to either:
+      // * have a dedicated .handleCancelled method on MonadError, implemented for monads which support it
+      // * and/or, have a dedicated callback on the RequestListener, which is called when the request is cancelled
+      // But we cannot modify either of the interfaces (due to bincompat), so we resort to using an atomic flag
+      val responseHandledOrExceptionCalled = new AtomicBoolean
       val requestToSend = request.onBodyReceived { meta =>
         onBodyReceivedCalled.set(true)
         listener.responseBodyReceived(request, meta, tag)
       }
       monad
         .handleError(delegate.send(requestToSend)) { case e: Exception =>
+          responseHandledOrExceptionCalled.set(true)
           monad.flatMap {
             ResponseException.find(e) match {
               case Some(re) => listener.responseHandled(requestToSend, re.response, tag, Some(re))
               case None     => listener.exception(requestToSend, tag, e, onBodyReceivedCalled.get())
             }
           } { _ => monad.error(e) }
         }
-        .flatMap(response => listener.responseHandled(requestToSend, response, tag, None).map(_ => response))
+        .flatMap { response =>
+          responseHandledOrExceptionCalled.set(true)
+          listener.responseHandled(requestToSend, response, tag, None).map(_ => response)
+        }
+        .ensure {
+          if (!responseHandledOrExceptionCalled.get()) {
+            listener.exception(
+              requestToSend,
+              tag,
+              new InterruptedException with NoStackTrace,
+              onBodyReceivedCalled.get()
+            )
+          } else monad.unit(())
+        }
     }
 }
 

core/src/main/scala/sttp/client4/listener/RequestListener.scala

@@ -49,6 +49,9 @@ trait RequestListener[F[_], L] {
     *
     * The [[responseBodyReceived]] might have been called before this method, but will not be called after.
     *
+    * If handling of the request is cancelled, and the effect system used doesn't rely on exceptions for cancellation,
+    * this method will still be called, with an [[InterruptedException]] exception.
+    *
     * @param responseBodyReceivedCalled
     *   Indicates if [[responseBodyReceivedCalled]] has been called before this method.
     */

effects/cats/src/test/scalajvm/sttp/client4/impl/cats/CatsListenerBackendTest.scala

@@ -0,0 +1,80 @@
+package sttp.client4.impl.cats
+
+import cats.effect.IO
+import cats.effect.kernel.Ref
+import cats.effect.std.Semaphore
+import cats.effect.unsafe.implicits.global
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import sttp.client4._
+import sttp.client4.httpclient.cats.HttpClientCatsBackend
+import sttp.client4.listener.ListenerBackend
+import sttp.client4.listener.RequestListener
+import sttp.model.ResponseMetadata
+
+class CatsListenerBackendTest extends AnyFlatSpec with Matchers {
+  // #2669
+  it should "notify the listener when a request is interrupted" in {
+    for {
+      inResponse <- Semaphore[IO](0)
+      trail <- Ref.of[IO, List[String]](Nil)
+      baseBackend = HttpClientCatsBackend.stub[IO].whenAnyRequest.thenRespondF {
+        inResponse.release *> IO.never
+      }
+      backend = ListenerBackend(baseBackend, spyingListener(trail))
+      // send the request in the background ...
+      sendingFiber <- basicRequest.get(uri"http://example.org").send(backend).void.start
+      // wait until the request is being processed ...
+      _ <- inResponse.acquire
+      // the counter should be 1, as it should be incremented in the listener's "before"
+      trailBefore <- trail.get
+      _ = trailBefore shouldBe List("before")
+      // cancel the request, interrupting the IO.never ...
+      _ <- sendingFiber.cancel
+      // and now the counter should be decreased again
+      trailAfter <- trail.get
+      _ = trailAfter shouldBe List("before", "exception")
+    } yield ()
+  }.unsafeRunSync()
+
+  it should "properly notify listener in case of normal completion" in {
+    for {
+      inResponse <- Semaphore[IO](0)
+      trail <- Ref.of[IO, List[String]](Nil)
+      baseBackend = HttpClientCatsBackend.stub[IO].whenAnyRequest.thenRespondOk()
+      backend = ListenerBackend(baseBackend, spyingListener(trail))
+      _ <- basicRequest.get(uri"http://example.org").send(backend)
+      trailAfter <- trail.get
+      _ = trailAfter shouldBe List("before", "response handled")
+    } yield ()
+  }.unsafeRunSync()
+
+  it should "properly notify listener in case of an exception" in {
+    for {
+      inResponse <- Semaphore[IO](0)
+      trail <- Ref.of[IO, List[String]](Nil)
+      baseBackend = HttpClientCatsBackend.stub[IO].whenAnyRequest.thenThrow(new RuntimeException("test exception"))
+      backend = ListenerBackend(baseBackend, spyingListener(trail))
+      _ <- basicRequest.get(uri"http://example.org").send(backend).handleErrorWith { case _ => IO.unit }
+      trailAfter <- trail.get
+      _ = trailAfter shouldBe List("before", "exception")
+    } yield ()
+  }.unsafeRunSync()
+
+  def spyingListener(trail: Ref[IO, List[String]]) = new RequestListener[IO, Unit] {
+    override def before(request: GenericRequest[_, _]): IO[Unit] = trail.update(_ :+ "before")
+    override def responseBodyReceived(request: GenericRequest[_, _], response: ResponseMetadata, tag: Unit): Unit = ()
+    override def responseHandled(
+        request: GenericRequest[_, _],
+        response: ResponseMetadata,
+        tag: Unit,
+        exception: Option[ResponseException[_]]
+    ): IO[Unit] = trail.update(_ :+ "response handled")
+    override def exception(
+        request: GenericRequest[_, _],
+        tag: Unit,
+        exception: Throwable,
+        responseBodyReceivedCalled: Boolean
+    ): IO[Unit] = trail.update(_ :+ s"exception")
+  }
+}