pierangeloc

A working setup for local kafka in docker-compose, using kafkacat as a CLI client

Setup Docker compose

Add these sections to your docker compose to bring kafka, zookeeper and required volumes up

  zookeeper:
    image: bitnami/zookeeper:latest
    ports:

pierangeloc

//test it with `runMain com.thenewmotion.myr.TestSttpLogging2 http://httpstat.us/400` from within SBT
object TestSttpLogging2 extends ZIOAppDefault {

  override val bootstrap: ZLayer[ZIOAppArgs, Any, Environment] =
    Runtime.removeDefaultLoggers >>> SLF4J.slf4j

  import sttp.client3.quick._
  // curl --request GET \
  //  --url httpstat.us/200 \

pierangeloc

val currentDir= os.pwd

println(os.list(currentDir).toList.map(p => (os.isDir(p), p.relativeTo(currentDir))).mkString("\n"))

def allSubDirectoriesNamed(name: String, p: os.Path): List[os.Path] = {
  if (os.isDir(p) && p.last == name) List(p)
  else if (os.isDir(p)) {
    for {
      child <- os.list(p).toList

pierangeloc

On docker-compose and healthchecks

I'm writing these notes because I spent a significant amount of time on a situation related with how docker and docker-compose work, specifically in integration tests. I've learned a few things about how docker that might be useful to remember at the next occasion.

If you use docker-compose for your integration tests, you might end up in a situation where you have erratic tests because, while tests are run, some containers are not yet doing what they are supposed to do.

For example, you might have a docker-compose.yml that sets up kafka, zookeeper and postgres and then you want to run your application that connects to these systems. Likely your build pipeline will follow these steps:

pierangeloc

import MyZIO.{Test, ZIO}

import scala.concurrent.Future
import scala.util.{Failure, Success}

/**
 * Following this talk https://www.youtube.com/watch?v=wsTIcHxJMeQ
 */
object MyZIO {

pierangeloc

package api

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, StatusCodes}
import akka.stream.ActorMaterializer

import scala.concurrent.Future
import scala.concurrent.duration._
import akka.http.scaladsl.server.Directives._

pierangeloc

import zio.console.Console
import zio.{App, ZIO}
import zio.{console, random}

object Main extends App {
  override def run(args: List[String]): ZIO[zio.ZEnv, Nothing, Int] =
    prg.forever

  val prg = for {
    _ <- console.putStrLn("Hier zijn twee nummers, jij moet de som geven")

pierangeloc

import io.estatico.newtype.ops._
import io.estatico.newtype.macros.newtype
import eu.timepit.refined.api.Refined
import doobie.implicits._
import doobie.postgres.implicits._
import doobie.refined.implicits._

  /**
    * Derive a Put (so it can be interpolated in doobie sql expressions) for any newtype backed by a type that is
    * supported by doobie (e.g. any wrapper of String, UUID, Boolean etc)

pierangeloc

trait TimedExecution {

  implicit val materializer: Materializer

  def delayed[A, B](f: A => B)(delay: FiniteDuration): A => Future[B] = a => Source.tick(delay, delay, a).take(1).map(f).runWith(Sink.head)

  def delayedValue[A](a: => A)(delay: FiniteDuration): Future[A] = delayed[Unit, A](_ => a)(delay)(())

}

pierangeloc

def loadResources(onBehalfOf: Class[_])(path: String): List[Path] = {
    val url: URL = onBehalfOf.getClassLoader.getResource(path)
    if (url.toURI.getScheme.contains("jar")) {
      val jar: URL = onBehalfOf.getProtectionDomain.getCodeSource.getLocation
      val jarFile: Path = Paths.get(jar.toString.substring("file:".length))
      val fs: FileSystem = FileSystems.newFileSystem(jarFile, null)
      val directoryStream: DirectoryStream[Path] = Files.newDirectoryStream(fs.getPath(path))
      directoryStream.asScala.toList
 } else {