if (stopped.compareAndSet(false, true)) {
      val activeConns = connectionTracker.activeConnectionCount
      val activeReqs = connectionTracker.activeRequestCount
      info(
        s"Stopping ${config.name} server at ${localAddress} " +
          s"(activeConnections=$activeConns, activeRequests=$activeReqs, " +
          s"quietPeriod=${config.shutdownQuietPeriodSeconds}s, timeout=${config.shutdownTimeoutSeconds}s)"
      )

      // Unregister shutdown hook to prevent double-stop during JVM shutdown
      unregisterShutdownHook()

      val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(config.shutdownTimeoutSeconds)
      def remainingNanos: Long = math.max(0, deadline - System.nanoTime())

      // Close the server channel first to stop accepting new connections
      channelFuture.foreach(_.close().await(remainingNanos, TimeUnit.NANOSECONDS))

      // Wait for active requests (including streaming responses) to complete
      if (connectionTracker.hasActiveRequests) {
        info(s"Waiting for ${connectionTracker.activeRequestCount} active requests to complete...")
        val completed = connectionTracker.awaitCompletion(remainingNanos, TimeUnit.NANOSECONDS)
        if (completed) {
          info("All active requests completed")
        } else {
          warn(s"Timeout waiting for active requests, ${connectionTracker.activeRequestCount} requests still pending")
        }
      }

      // Gracefully shutdown worker group first to complete in-flight requests
      val workerFuture = workerGroup.shutdownGracefully(
        TimeUnit.SECONDS.toNanos(config.shutdownQuietPeriodSeconds),
        remainingNanos,
        TimeUnit.NANOSECONDS
      )
      workerFuture.await(remainingNanos, TimeUnit.NANOSECONDS)

      // Then shutdown boss group
      val bossFuture = bossGroup.shutdownGracefully(
        TimeUnit.SECONDS.toNanos(config.shutdownQuietPeriodSeconds),
        remainingNanos,
        TimeUnit.NANOSECONDS
      )
      bossFuture.await(remainingNanos, TimeUnit.NANOSECONDS)

      // Close the HTTP logger
      httpLogger.close()

      info(s"${config.name} server stopped")
    }

  test("request tracking with simple HTTP request") {
    val router = RxRouter.of[SimpleApi]
    val config = Netty.server
      .withRouter(router)
      .noLogging

    config.designWithSyncClient.build[NettyServer, SyncClient] { (server, client) =>
      // Initially, the request count should be zero
      server.activeRequestCount shouldBe 0

      // Make a request
      val response = client.send(Http.GET("/hello"))
      response.status shouldBe HttpStatus.Ok_200

      // After the request completes, the count should return to 0.
      // Use `eventually` to handle the asynchronous nature of request completion tracking.
      eventually {
        server.activeRequestCount shouldBe 0
      }
    }
  }

    def sleepUnlessInterrupted(remainingNanos: Long): Boolean = {
      try {
        val sleepMillis = math.min(pollIntervalMs, math.max(1L, TimeUnit.NANOSECONDS.toMillis(remainingNanos)))
        Thread.sleep(sleepMillis)
        true
      } catch {
        case _: InterruptedException =>
          Thread.currentThread().interrupt()
          false
      }
    }

    @scala.annotation.tailrec
    def loop(): Boolean = {
      if (!hasActiveRequests) {
        true
      } else {
        val remainingNanos = deadlineNanos - System.nanoTime()
        if (remainingNanos <= 0) {
          warn(s"Timeout waiting for ${activeRequests.get()} active requests to complete")
          false
        } else if (!sleepUnlessInterrupted(remainingNanos)) {
          false
        } else {
          loop()
        }
      }
    }

            def completeSseStream(): Unit = {
              connectionTracker.requestCompleted()
              val f = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)
              f.addListener(ChannelFutureListener.CLOSE)
            }

            val c = RxRunner.run(resp.events) {
              case OnNext(e: ServerSentEvent) =>
                val event = e.toContent
                val buf   = Unpooled.copiedBuffer(event.getBytes("UTF-8"))
                ctx.writeAndFlush(new DefaultHttpContent(buf))
              case OnNext(_) =>
                // Ignore non-ServerSentEvent values
                ()
              case OnError(_) =>
                // Stream error - mark request as completed and close
                completeSseStream()
              case OnCompletion =>
                // Stream completed - mark request as completed and close
                completeSseStream()
            }