Complete SSE client Implementation for Scala using http4s.

SseClient.scala

import cats.effect.{IO, Timer}
import fs2.{Pull, Stream}
import org.http4s._
import org.http4s.ServerSentEvent.EventId
import org.http4s.client.Client
import org.http4s.headers.{`Cache-Control`, Accept}
import scala.concurrent.duration._
import SseClient._

final class SseClient private (
    httpClient: Client[IO],
    maxRetries: Int,
    initialRetryInterval: FiniteDuration
)(implicit timer: Timer[IO]) {

  def stream(uri: Uri): Stream[IO, ServerSentEvent] = autoReconnectStream(uri)

  private def autoReconnectStream(uri: Uri): Stream[IO, ServerSentEvent] = {
    def go(s: Stream[IO, ServerSentEvent], metadata: SseMetadata): Pull[IO, ServerSentEvent, Unit] =
      s.pull.uncons1
        .flatMap {
          case Some((event, tail)) =>
            val newMetadata = metadata.copy(
              eventId = event.id.fold(metadata.eventId)(id => if (id != EventId.reset) Some(id.value) else None),
              retry = event.retry.fold(metadata.retry)(r => Some(r.millis))
            )
            val metadataChanged = newMetadata != metadata
            Pull.output1(event) >> {
              // Add a new error handler only if the metadata changed.
              if (metadataChanged) goWithErrorHandled(tail, newMetadata) else go(tail, metadata)
            }
          case None =>
            reconnect(metadata, None)
        }

    def reconnect(metadata: SseMetadata, err: Option[Throwable]): Pull[IO, ServerSentEvent, Unit] = {
      val delay = metadata.retry.getOrElse(initialRetryInterval)
      val newMetadata = metadata.copy(attempts = metadata.attempts + 1)

      if (newMetadata.attempts > maxRetries) Pull.raiseError[IO](MaxRetriesReached(err))
      else goWithErrorHandled(newStream(uri, newMetadata.eventId).delayBy(delay), newMetadata)
    }

    def goWithErrorHandled(
        stream: Stream[IO, ServerSentEvent],
        metadata: SseMetadata
    ): Pull[IO, ServerSentEvent, Unit] =
      go(stream, metadata).handleErrorWith {
        case e: MaxRetriesReached => Pull.raiseError[IO](e)
        case other                => reconnect(metadata, Some(other))
      }

    goWithErrorHandled(newStream(uri, None), SseMetadata(None, None, 1)).stream
  }

  private def newStream(uri: Uri, eventId: Option[String]): Stream[IO, ServerSentEvent] =
    httpClient
      .stream(
        Request(
          uri = uri,
          headers = BaseHeaders ++ Headers(eventId.map(Header("Last-Event-ID", _)).toList)
        )
      )
      .flatMap(_.body)
      .through(ServerSentEvent.decoder[IO])
}

object SseClient {
  val DefaultRetryInterval: FiniteDuration = 5.seconds
  val DefaultMaxRetries: Int = 10

  def apply(
      httpClient: Client[IO],
      maxRetries: Int = DefaultMaxRetries,
      initialRetryInterval: FiniteDuration = DefaultRetryInterval
  )(implicit timer: Timer[IO]): SseClient =
    new SseClient(httpClient, maxRetries, initialRetryInterval)

  private case class SseMetadata(eventId: Option[String], retry: Option[FiniteDuration], attempts: Int)
  private case class MaxRetriesReached(underlying: Option[Throwable])
      extends Exception("Reached maximum number of retries", underlying.orNull)

  private val BaseHeaders: Headers = Headers.of(
    `Cache-Control`(CacheDirective.`no-cache`()),
    Accept(MediaType.`text/event-stream`)
  )
}

I did some changes in my fork to remove dependency on IO directly