detect-stream-timeout.scala

  private def detectStall[F[_]: Concurrent: Timer, A]: Pipe[F, A, A] = {
    val currentTime = Sync[F].delay(OffsetDateTime.now())
    (stream: Stream[F, A]) =>
      Stream.eval(currentTime).flatMap { startTime =>
        Stream.eval(Ref[F].of(startTime)).flatMap { lastChunkSeen =>
          val watchdog = Stream
            .fixedRate(30.seconds)
            .evalMap(_ => (currentTime, lastChunkSeen.get).tupled)
            .ensure(new RuntimeException("pipeline appears to have stalled"))({
              case (now, lastChunk) =>
                now.minusSeconds(30).isAfter(lastChunk)
            })
          stream
            .evalTapChunk(_ => currentTime.flatMap(lastChunkSeen.set))
            .concurrently(watchdog)
        }
      }
  }