http4s Consul Middleware

Middleware for http4s to allow users to discover the host and port for an HTTP request using Consul service discovery.

The middleware rewrites URIs of the form consul://service-name/path to an HTTP URI, using Consul to discover an available instance of the service service-name. For example, if Consul knew about a service named httpd available at 127.0.0.1:80, a request made to consul://httpd/example/path would be rewritten and requested from http://127.0.0.1:80/example/path.

Usage

http4s middleware wrap an underlying Client[F], so we need to have such a client available. Consul's API uses "blocking queries" implemented via long polling to immediately notify listeners when its internal state changes. For this reason, we recommend constructing a second Client[F] with a much longer timeout.

import cats.effect._, cats.syntax.all._, cats.effect.std._
import fs2.Stream
import org.http4s._, org.http4s.Method.GET, org.http4s.client.Client, org.http4s.ember.client.EmberClientBuilder, org.http4s.syntax.all._, org.http4s.client.dsl.io._
import org.typelevel.log4cats.Logger, org.typelevel.log4cats.slf4j.Slf4jLogger
import scala.concurrent.duration._
import com.dwolla.http4s.consul._
import cats.effect.unsafe.implicits.global

val exampleConsulUri = uri"consul://httpd/"

def clientWithTimeout[F[_] : Async](timeout: FiniteDuration): Resource[F, Client[F]] =
  EmberClientBuilder.default[F].withTimeout(timeout).build

def longPollClient[F[_] : Async]: Resource[F, Client[F]] = clientWithTimeout(10.minutes)

def typicalClient[F[_] : Async]: Resource[F, Client[F]] = clientWithTimeout(20.seconds)

def consulServiceDiscoveryAlg[F[_] : Async : Logger : Random]: Resource[F, ConsulServiceDiscoveryAlg[F]] =
  longPollClient[F].map(ConsulServiceDiscoveryAlg(uri"http://localhost:8500", 1.minute, _))

def consulAwareClient[F[_] : Async : Logger : Random]: Resource[F, Client[F]] =
  (consulServiceDiscoveryAlg[F], typicalClient[F])
    .parMapN(ConsulMiddleware(_)(_))
    .flatten

Random.scalaUtilRandom[IO].flatMap { implicit random =>
  Slf4jLogger.create[IO].flatMap { implicit logger =>
    // in a real app you'd probably want a real Trace implementation
    import natchez.Trace.Implicits.noop

    Stream.resource(consulAwareClient[IO])
      .flatMap { client =>
        // make a GET call to consul://httpd/, every 2 seconds, until shut down
        //
        // if you change the state of the Consul cluster by registering
        // or de-registering services named "httpd", you should see the
        // requests going to different endpoints over time.
        
        Stream.repeatEval(client.successful(GET(exampleConsulUri))
          .flatMap {
            case true => Logger[IO].info("🔮 success")
            case false => Logger[IO].info("🔮 failure")
          })
          .metered(2.seconds)
      }
      .compile
      .drain
  }
}
  .unsafeRunSync()

Dwolla/http4s-consul-middleware

http4s Consul Middleware

Usage