### Clone ZIO Quickstarts Kafka Project

Source: https://zio.dev/guides/tutorials/producing-consuming-data-from-kafka-topics

Clone the ZIO Quickstarts Kafka project to access the code examples.

```bash
git clone https://github.com/zio/zio-quickstarts.git
cd zio-quickstarts/zio-quickstart-kafka
```

--------------------------------

### Add ZIO Config Modules for Examples

Source: https://zio.dev/zio-config

Include these lines in your `build.sbt` to use the modules required for the examples.

```scala
libraryDependencies += "dev.zio" %% "zio-config"          % "<version>"
libraryDependencies += "dev.zio" %% "zio-config-magnolia" % "<version>"
libraryDependencies += "dev.zio" %% "zio-config-typesafe" % "<version>"
libraryDependencies += "dev.zio" %% "zio-config-refined"  % "<version>"
```

--------------------------------

### Basic Logging with ZIO.log

Source: https://zio.dev/reference/observability/logging

Use ZIO.log for general logging messages. This example demonstrates logging application start, user input, and a greeting.

```scala
import zio._
val app =  for {
    _ <- ZIO.log("Application started!")
    name <- Console.readLine("Please enter your name: ")
    _ <- ZIO.log(s"User entered its name: $name")
    _ <- Console.printLine(s"Hello, $name")
} yield ()
```

--------------------------------

### Full Example of Acquire Release with File Handling

Source: https://zio.dev/reference/core/zio

This example demonstrates acquiring a FileInputStream, converting its content to a string, and ensuring the stream is closed. It utilizes `ZIO.attempt` for fallible operations and `ZIO.acquireReleaseWith` for resource management.

```scala
import zio._
import java.io.{ File, FileInputStream }
import java.nio.charset.StandardCharsets
object Main extends ZIOAppDefault {
  // run my acquire release
  def run = myAcquireRelease
  def closeStream(is: FileInputStream) =
    ZIO.succeed(is.close())
  def convertBytes(is: FileInputStream, len: Long) =
    ZIO.attempt {
      val buffer = new Array[Byte](len.toInt)
      is.read(buffer)
      println(new String(buffer, StandardCharsets.UTF_8))
    }
  // myAcquireRelease is just a value. Won't execute anything here until interpreted
  val myAcquireRelease: Task[Unit] = for {
    file   <- ZIO.attempt(new File("/tmp/hello"))
    len    = file.length
    string <- ZIO.acquireReleaseWith(ZIO.attempt(new FileInputStream(file)))(closeStream)(convertBytes(_, len))
  } yield string
}
```

--------------------------------

### Full Example with Imports and Multiple Types (Scala 2)

Source: https://zio.dev/zio-json

A comprehensive example demonstrating imports, case classes, sealed traits, and JSON parsing/encoding in Scala 2.

```scala
import zio.json._

sealed trait Fruit extends Product with Serializable
case class Banana(curvature: Double) extends Fruit
case class Apple(poison: Boolean)    extends Fruit

object Fruit {
  implicit val decoder: JsonDecoder[Fruit] =    DeriveJsonDecoder.gen[Fruit]
  implicit val encoder: JsonEncoder[Fruit] =    DeriveJsonEncoder.gen[Fruit]
}

val json1         = """{ \"Banana\":{ \"curvature\":0.5 }}"""
val json2         = """{ \"Apple\": { \"poison\": false }}"""
val malformedJson = """{ \"Banana\":{ \"curvature\": true }}"""

json1.fromJson[Fruit]
json2.fromJson[Fruit]
malformedJson.fromJson[Fruit]
List(Apple(false), Banana(0.4)).toJsonPretty
```

--------------------------------

### Full Example with Imports and Multiple Types (Scala 3)

Source: https://zio.dev/zio-json

A comprehensive example demonstrating imports, enums, and JSON parsing/encoding in Scala 3.

```scala
import zio.json.*

enum Fruit derives JsonCodec {
  case Banana(curvature: Double)
  case Apple(poison: Boolean)
}
export Fruit.*

val json1         = """{ \"Banana\":{ \"curvature\":0.5 }}"""
val json2         = """{ \"Apple\": { \"poison\": false }}"""
val malformedJson = """{ \"Banana\":{ \"curvature\": true }}"""

json1.fromJson[Fruit]
json2.fromJson[Fruit]
malformedJson.fromJson[Fruit]
List(Apple(false), Banana(0.4)).toJsonPretty
```

--------------------------------

### Full Example: In-memory Key-Value Store with @accessible

Source: https://zio.dev/reference/service-pattern/accessor-methods

A complete example demonstrating an in-memory key-value store using the @accessible macro for generating service accessors and ZIOAppDefault for running the application.

```scala
@accessible
trait KeyValueStore {
  def set(key: String, value: Int): Task[Int]

  def get(key: String): Task[Int]
}

case class InmemoryKeyValueStore(map: Ref[Map[String, Int]])
  extends KeyValueStore {
  override def set(key: String, value: Int): Task[Int] = 
    map.update(_.updated(key, value)).map(_ => value)

  override def get(key: String): Task[Int] = 
    map.get.map(_.get(key)).someOrFailException
}

object InmemoryKeyValueStore {
  val layer: ULayer[KeyValueStore] = 
    ZLayer {
      for {
        map <- Ref.make(Map[String, Int]())
      } yield InmemoryKeyValueStore(map)
    }
}

object MainApp extends ZIOAppDefault {
  val myApp = 
    for {
      _ <- KeyValueStore.set("key", 5)
      key <- KeyValueStore.get("key")
    } yield key

  def run = myApp.provide(InmemoryKeyValueStore.layer).debug
}
```

--------------------------------

### Running DynamoDB Local

Source: https://zio.dev/zio-dynamodb

Instructions for starting and stopping the DynamoDB Local Docker container. This is necessary for running integration tests and examples that use an in-memory DynamoDB instance.

```bash
docker compose -f docker/docker-compose.yml up -d
```

```bash
docker compose -f docker/docker-compose.yml down
```

--------------------------------

### Complete ZIO Kafka Streaming Example

Source: https://zio.dev/guides/tutorials/producing-consuming-data-from-kafka-topics

This example demonstrates a full streaming application using ZIO Kafka and ZIO Streams. It sets up a producer to repeatedly send timestamps and a consumer to read and print these messages, committing offsets asynchronously. Ensure Kafka is running on localhost:9092 and the topic 'streaming-hello' exists.

```scala
import org.apache.kafka.clients.producer.ProducerRecord
import zio._
import zio.kafka.consumer._
import zio.kafka.producer.{Producer, ProducerSettings}
import zio.kafka.serde._
import zio.stream.ZStream

object StreamingKafkaApp extends ZIOAppDefault {

  private val BOOSTRAP_SERVERS = List("localhost:9092")
  private val KAFKA_TOPIC      = "streaming-hello"

  private val producerSettings = ProducerSettings(BOOSTRAP_SERVERS)
  private val consumerSettings =
    ConsumerSettings(BOOSTRAP_SERVERS).withGroupId("streaming-kafka-app")

  def run: ZIO[Any, Throwable, Unit] = {

    val p: ZIO[Any, Throwable, Unit] = ZIO.scoped {
      for {
        producer <- Producer.make(producerSettings)
        _ <- ZStream
              .repeatZIO(Clock.currentDateTime)
              .schedule(Schedule.spaced(1.second))
              .map {
                time =>
                  new ProducerRecord(
                    KAFKA_TOPIC,
                    time.getMinute,
                    s"$time -- Hello, World!"
                  )
              }
              .via(producer.produceAll(Serde.int, Serde.string))
              .runDrain
      } yield ()
    }

    val c: ZIO[Any, Throwable, Unit] = ZIO.scoped {
      for {
        consumer <- Consumer.make(consumerSettings)
        _ <- consumer
              .plainStream(
                Subscription.topics(KAFKA_TOPIC),
                Serde.int,
                Serde.string
              )
              // Do not use `tap` in throughput sensitive applications because it
              // destroys the chunking structure and leads to lower performance.
              // See the previous section for more info.
              .tap(r => Console.printLine("Consumed: " + r.value))
              .map(_.offset)
              .aggregateAsync(Consumer.offsetBatches)
              .mapZIO(_.commit)
              .runDrain
      } yield ()
    }

    p <&> c
  }
}
```

--------------------------------

### Starting Root and Child Spans

Source: https://zio.dev/zio-telemetry/opentracing

Illustrates how to start a new root span with baggage and a child span with tags and logs using ZIO Telemetry's OpenTracing aspects. Ensure `tracing.aspects._` is imported.

```scala
ZIO.serviceWithZIO[OpenTracing] {
    tracing =>
      import tracing.aspects._

      // start a new root span and set some baggage item
      val zio1 = tracing.setBaggage("foo", "bar") @@ root("root span")

      // start a child of the current span, set a tag and log a message
      val zio2 =     (for {
        _ <- tracing.tag("http.status_code", 200)
        _ <- tracing.log("doing some serious work here!")
      } yield ()) @@ span("child span")
      
      zio1 *> zio2
}
```

--------------------------------

### Basic ZIO Process Example

Source: https://zio.dev/zio-process

Demonstrates calling external commands, streaming their output, piping commands, and redirecting output to a file. Requires importing zio._, zio.process.Command, and java.io.File.

```scala
import zio._
import zio.process.Command
import java.io.File

object ZIOProcessExample extends ZIOAppDefault {

  val myApp = for {
    fiber <- Command("dmesg", "--follow").linesStream
      .foreach(Console.printLine(_))
      .fork

    cpuModel <- (Command("cat", "/proc/cpuinfo") |
                 Command("grep", "model name") |
                 Command("head", "-n", "1") |
                 Command("cut", "-d", ":", "-f", "2")).string

    _ <- Console.printLine(s"CPU Model: $cpuModel")

    _ <- (Command("pg_dump", "my_database") > new File("dump.sql")).exitCode

    _ <- fiber.join
  } yield ()

  override def run = myApp
}
```

--------------------------------

### Example of Logging within a ZIO Span

Source: https://zio.dev/guides/tutorials/enable-logging-in-a-zio-application

This example demonstrates creating a span named "get-profile-picture" around a workflow that fetches user information and downloads a profile image. Logs inside this span will be annotated with the span's name and duration.

```scala
case class User(id: String, name: String, profileImage: String)

def getProfilePicture(username: String) =
  ZIO.logSpan("get-profile-picture") {
    for {
      _    <- ZIO.log(s"Getting information of $username from the UserService")
      user <- ZIO.succeed(User("1", "john", "john.png"))
      _    <- ZIO.log(s"Downloading profile image ${user.profileImage}")
      img  <- ZIO.succeed(Array[Byte](1, 2, 3))
      _    <- ZIO.log("Profile image downloaded")
    } yield img
  }
```

--------------------------------

### Configure Logback Logging

Source: https://zio.dev/guides/tutorials/create-custom-logger-for-a-zio-application

Create a logback.xml file in the resources directory to configure the Logback logger. This example sets up a console appender with a specific pattern.

```xml
<?xml version="1.0" encoding="UTF-8"?>

<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>
    <root level="info">
        <appender-ref ref="STDOUT"/>
    </root>
</configuration>
```

--------------------------------

### Enable SLF4J Logging in ZIO Application

Source: https://zio.dev/guides/tutorials/create-custom-logger-for-a-zio-application

Use the SLF4J.slf4j layer to integrate SLF4J logging. This example demonstrates starting the application with colored log output.

```scala
object MainApp extends ZIOAppDefault {
  override val bootstrap = SLF4J.slf4j(LogFormat.colored)

  def run = ZIO.log("Application started!")
}
```

--------------------------------

### Example: Using fromHubScoped with Promise

Source: https://zio.dev/reference/concurrency/hub

Demonstrates using `fromHubScoped` with a `Promise` to sequence subscription and publishing. The `Promise` ensures the subscription is active before messages are sent to the hub.

```scala
for {
  promise <- Promise.make[Nothing, Unit]
  hub     <- Hub.bounded[String](2)
  scoped  = ZStream.fromHubScoped(hub).tap(_ => promise.succeed(()))
  stream   = ZStream.unwrapScoped(scoped)
  fiber   <- stream.take(2).runCollect.fork
  _       <- promise.await
  _       <- hub.publish("Hello")
  _       <- hub.publish("World")
  _       <- fiber.join
} yield ()
```

--------------------------------

### Assemble and Run ZIO Application

Source: https://zio.dev/reference/service-pattern/service-pattern

An example of a ZIO application that uses the DocRepo service. It demonstrates how to provide multiple service layers (DocRepo, InmemoryBlobStorage, InmemoryMetadataRepo) to the application's run method.

```scala
object MainApp extends ZIOAppDefault {
  val app = 
    for {
      docRepo <- ZIO.service[DocRepo]
      id      <- docRepo.save(
                    Doc(
                      "title",
                      "description",
                      "en",
                      "text/plain",
                      "content".getBytes()
                    )
                 )
      doc     <- docRepo.get(id)
      _       <- Console.printLine(
                   s"""
                     |Downloaded the document with $id id:
                     |  title: ${doc.title}
                     |  description: ${doc.description}
                     |  language: ${doc.language}
                     |  format: ${doc.format}
                     |""".stripMargin
                 )
      _       <- docRepo.delete(id)
      _       <- Console.printLine(s"Deleted the document with $id id")
    } yield ()

  def run = 
    app.provide(
      DocRepo.live,
      InmemoryBlobStorage.layer,
      InmemoryMetadataRepo.layer
    )
}
```

--------------------------------

### ZIO AWS Example: Describe Applications and EC2 Instances

Source: https://zio.dev/zio-aws

This example demonstrates how to use ZIO AWS to describe Elastic Beanstalk applications and their associated EC2 instances. It requires EC2 and ElasticBeanstalk services to be available.

```scala
import zio.aws.core.AwsError
import zio.aws.core.config.AwsConfig
import zio.aws.ec2.Ec2
import zio.aws.ec2.model._
import zio.aws.ec2.model.primitives._
import zio.aws.elasticbeanstalk.ElasticBeanstalk
import zio.aws.elasticbeanstalk.model._
import zio.aws.elasticbeanstalk.model.primitives._
import zio.aws.netty.NettyHttpClient
import zio._
import zio.stream._

object ZIOAWSExample extends ZIOAppDefault {

  val program: ZIO[Ec2 & ElasticBeanstalk, AwsError, Unit] =
    for {
      appsResult <- ElasticBeanstalk.describeApplications(
        DescribeApplicationsRequest(applicationNames = Some(List(ApplicationName("my-service"))))
      )
      app <- appsResult.getApplications.map(_.headOption)
      _ <- app match {
        case Some(appDescription) =>
          for {
            applicationName <- appDescription.getApplicationName
            _ <- Console
              .printLine(
                s"Got application description for $applicationName"
              )
              .ignore
            envStream = ElasticBeanstalk.describeEnvironments(
              DescribeEnvironmentsRequest(applicationName =
                Some(applicationName)
              )
            )
            _ <- envStream.run(ZSink.foreach {
              env =>
                env.getEnvironmentName.flatMap {
                  environmentName =>
                    (for {
                      environmentId <- env.getEnvironmentId
                      _ <- Console
                        .printLine(
                          s"Getting the EB resources of $environmentName"
                        )
                        .ignore
                      resourcesResult <-
                        ElasticBeanstalk.describeEnvironmentResources(
                          DescribeEnvironmentResourcesRequest(environmentId =
                            Some(environmentId)
                          )
                        )
                      resources <- resourcesResult.getEnvironmentResources
                      _ <- Console
                        .printLine(
                          s"Getting the EC2 instances in $environmentName"
                        )
                        .ignore
                      instances <- resources.getInstances
                      instanceIds <- ZIO.foreach(instances)(_.getId)
                      _ <- Console
                        .printLine(
                          s"Instance IDs are ${instanceIds.mkString(", ")}"
                        )
                        .ignore
                      reservationsStream = Ec2.describeInstances(
                        DescribeInstancesRequest(instanceIds = Some(instanceIds.map(id => zio.aws.ec2.model.primitives.InstanceId(ResourceId.unwrap(id)))))
                      )
                      _ <- reservationsStream.run(ZSink.foreach {
                        reservation =>
                          reservation.getInstances
                            .flatMap {
                              instances =>
                                ZIO.foreach(instances) {
                                  instance =>
                                    for {
                                      id <- instance.getInstanceId
                                      typ <- instance.getInstanceType
                                      launchTime <- instance.getLaunchTime
                                      _ <- Console.printLine(s"  instance $id:").ignore
                                      _ <- Console.printLine(s"    type: $typ").ignore
                                      _ <- Console
                                        .printLine(
                                          s"    launched at: $launchTime"
                                        )
                                        .ignore
                                    } yield () 
                                }
                            }
                      })
                    } yield ()) 
                }
            })
          } yield () 
        case None =>
          Console.printLine("Application not found").ignore
      }
    } yield ()

  override def run = program.provide(NettyHttpClient.layer, AwsConfig.default >>> Ec2.live, AwsConfig.default >>> ElasticBeanstalk.live)
}
```

--------------------------------

### Run Kafka Cluster with Docker Compose

Source: https://zio.dev/guides/tutorials/producing-consuming-data-from-kafka-topics

Start the Kafka cluster using the docker-compose.yml file.

```bash
docker compose up -d
```

--------------------------------

### Add ZIO SBT Plugins to project.sbt

Source: https://zio.dev/zio-sbt

Add these lines to your `plugin.sbt` file to install ZIO SBT plugins.

```scala
addSbtPlugin("dev.zio" %% "zio-sbt-website" % "2.2.0")
```

--------------------------------

### ZStream Examples

Source: https://zio.dev/reference/stream/zstream/index

Demonstrates various types of ZStream, including empty, single-element, finite, and infinite streams, showcasing their creation and type signatures.

```scala
val emptyStream         : ZStream[Any, Nothing, Nothing]   = ZStream.empty
val oneIntValueStream   : ZStream[Any, Nothing, Int]       = ZStream.succeed(4)
val oneListValueStream  : ZStream[Any, Nothing, List[Int]] = ZStream.succeed(List(1, 2, 3))
val finiteIntStream     : ZStream[Any, Nothing, Int]       = ZStream.range(1, 10)
val infiniteIntStream   : ZStream[Any, Nothing, Int]       = ZStream.iterate(1)(_ + 1)
```

--------------------------------

### Example: Running Stream into Hub with Take

Source: https://zio.dev/reference/concurrency/hub

Demonstrates using `runIntoHub` to send stream elements to a Hub of `Take` values. It includes unwrapping `Take` values using `flattenTake` for stream consumption.

```scala
for {
  promise <- Promise.make[Nothing, Unit]
  hub     <- Hub.bounded[Take[Nothing, String]](2)
  scoped  = ZStream.fromHubScoped(hub).tap(_ => promise.succeed(()))
  stream   = ZStream.unwrapScoped(scoped).flattenTake
  fiber   <- stream.take(2).runCollect.fork
  _       <- promise.await
  _       <- ZStream("Hello", "World").runIntoHub(hub)
  _       <- fiber.join
} yield ()
```

--------------------------------

### Add ZIO SBT GitHub Query Plugin to plugins.sbt

Source: https://zio.dev/zio-sbt

Add this line to your `plugins.sbt` file to install the ZIO SBT GitHub Query plugin.

```scala
addSbtPlugin("dev.zio" % "zio-sbt-gh-query" % "0.5.0")
```

--------------------------------

### DocRepo Service Implementation

Source: https://zio.dev/reference/service-pattern

Implements the DocRepo service interface with basic methods. This serves as a starting point before adding dependencies.

```scala
final class DocRepoLive() extends DocRepo {  override def get(id: String): ZIO[Any, Throwable, Doc] = ???  override def save(document: Doc): ZIO[Any, Throwable, String] = ???  override def delete(id: String): ZIO[Any, Throwable, Unit] = ???  override def findByTitle(title: String): ZIO[Any, Throwable, List[Doc]] = ???}
```

--------------------------------

### Creating a ZPipeline from a custom channel

Source: https://zio.dev/reference/stream/zpipeline

Illustrates building a stateful pipeline from a ZChannel for transformations that cannot be expressed with simple map functions. This example creates a pipeline that pairs each element with its predecessor.

```APIDOC
## ZPipeline.fromChannel with ZChannel.readWithCause

### Description
Builds pipelines directly from `ZChannel` for stateful transformations. This example creates a pipeline that pairs each element with its predecessor.

### Method Signature
```scala
def pairwise[A]: ZPipeline[Any, Nothing, A, (A, A)]
```

### Channel Logic (`pairwiseGo`)
Handles three cases: processing incoming chunks, handling errors, and stream completion.

- **Case 1 (in):** Process incoming chunk, emit outputs, recurse to read next chunk.
- **Case 2 (err):** Handle errors from upstream (usually propagate with `refailCause`).
- **Case 3 (done):** Handle stream completion (finalize any pending state).

### Code Example
```scala
def pairwise[A]: ZPipeline[Any, Nothing, A, (A, A)] = 
  ZPipeline.fromChannel(pairwiseGo[A](None))

def pairwiseGo[A](prev: Option[A]): ZChannel[Any, ZNothing, Chunk[A], Any, ZNothing, Chunk[(A, A)], Any] = 
  ZChannel.readWithCause(
    (in: Chunk[A]) => {
      val buf  = Chunk.newBuilder[(A, A)]
      var last = prev
      in.foreach { a =>
        last.foreach(p => buf += ((p, a)))
        last = Some(a)
      }
      val out = buf.result()
      (if (out.nonEmpty) ZChannel.write(out) else ZChannel.unit) *> pairwiseGo(last)
    },
    (err: Cause[ZNothing]) => ZChannel.refailCause(err),
    (_: Any) => ZChannel.unit
  )
```
```

--------------------------------

### Editor with Constructor-Based Dependency Injection

Source: https://zio.dev/reference/di

This example demonstrates constructor-based dependency injection where the Editor service receives its dependencies (Formatter and Compiler) through its constructor. This decouples the Editor from the creation of its dependencies, improving testability.

```scala
import zio._

class Editor(formatter: Formatter, compiler: Compiler) {
  def formatAndCompile(code: String): UIO[String] = ???
}
```

--------------------------------

### Traditional Dependency Passing Example

Source: https://zio.dev/reference/contextual/index

Illustrates a cumbersome approach to dependency management where multiple services are passed as explicit parameters to functions.

```scala
def foo(
   s1: Service1,
   s2: Service2,
   s3: Service3
)(arg1: String, arg2: String, arg3: Int): Task[Int] = ???

def bar(
  s1: Service1,
  s12: Service12,
  s18: Service18, 
  sn: ServiceN
)(arg1: Int, arg2: String, arg3: Double, arg4: Int): Task[Unit]

def myApp(s1: Service1, s2: Service2, ..., sn: ServiceN): Task[Unit] = 
  for {
    a <- foo(s1, s2, s3)("arg1", "arg2", 4) 
    _ <- bar(s1, s12, s18, sn)(7, "arg2", 1.2, a)
      ...
  } yield ()
```

--------------------------------

### Basic ZIO Ensuring Example

Source: https://zio.dev/reference/core/zio

Demonstrates the basic usage of `ensuring` to execute a finalizer after an effect terminates. The finalizer is guaranteed to run even if the effect fails.

```scala
import zio._
val finalizer =  ZIO.succeed(println("Finalizing!"))
val finalized: IO[String, Unit] =  ZIO.fail("Failed!").ensuring(finalizer)
```

--------------------------------

### Configure Reload4j Logging

Source: https://zio.dev/guides/tutorials/create-custom-logger-for-a-zio-application

Create a log4j.properties file in the resources directory to configure the Reload4j logger. This example sets the root logger to Info level and directs output to the console.

```properties
log4j.rootLogger = Info, consoleAppender
log4j.appender.consoleAppender=org.apache.log4j.ConsoleAppender
log4j.appender.consoleAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.consoleAppender.layout.ConversionPattern=[%p] %d %c %M - %m%n
```

--------------------------------

### Basic OpenTracing Usage with Jaeger

Source: https://zio.dev/zio-telemetry/opentracing

Demonstrates starting a root span, setting tags, managing baggage, and logging within a ZIO application using JaegerTracer. Requires importing `tracing.aspects._`.

```scala
import zio.telemetry.opentracing.OpenTracing
import zio.telemetry.opentracing.example.JaegerTracer
import zio._
import io.opentracing.tag.Tags

val app =  ZIO.serviceWithZIO[OpenTracing] {
    tracing =>
      import tracing.aspects._
      (for {
        _ <- tracing.tag(Tags.SPAN_KIND.getKey, Tags.SPAN_KIND_CLIENT)
        _ <- tracing.tag(Tags.HTTP_METHOD.getKey, "GET")
        _ <- tracing.setBaggageItem("proxy-baggage-item-key", "proxy-baggage-item-value")
        message <- Console.readline
        _ <- tracing.log("Message has been read")
      } yield message) @@ root("/app")
  }.provide(OpenTracing.live, JaegerTracer.live("my-app"))
```

--------------------------------

### Basic File Line Counting (Potential Leak)

Source: https://zio.dev/reference/resource

This example demonstrates a basic file line counting function. Without proper error handling, it can lead to resource leaks if an exception occurs before the file is closed.

```scala
def lines(file: String): Task[Long] = ZIO.attempt {  def countLines(br: BufferedReader): Long = br.lines().count()  val bufferedReader = new BufferedReader(    new InputStreamReader(new FileInputStream("file.txt")),    2048  )  val count = countLines(bufferedReader)  bufferedReader.close()  count}
```

--------------------------------

### ZIO DynamoDB JSON Example

Source: https://zio.dev/zio-dynamodb

This example demonstrates how to use the ZIO DynamoDB JSON module to put and get a Person object. Ensure you have the necessary AWS configurations and DynamoDB table set up. The example uses ZIO's effect system for asynchronous operations.

```scala
import zio.aws.core.config
import zio.aws.{ dynamodb, netty }
import zio.dynamodb.DynamoDBQuery.{ get, put }
import zio.dynamodb.{ DynamoDBExecutor }
import zio.schema.{ DeriveSchema, Schema }
import zio.ZIOAppDefault
import zio.dynamodb.ProjectionExpression

object Main extends ZIOAppDefault {

  final case class Person(id: Int, firstName: String)
  object Person {
    implicit lazy val schema: Schema.CaseClass2[Int, String, Person] = DeriveSchema.gen[Person]
    val (id, firstName) = ProjectionExpression.accessors[Person]
  }

  val examplePerson = Person(1, "avi")

  private val program = for {
    _      <- put("personTable", examplePerson).execute
    person <- get("personTable")(Person.id.partitionKey === 1).execute
    _      <- zio.Console.printLine(s"hello $person")
  } yield ()

  override def run = 
    program.provide(
      netty.NettyHttpClient.default,
      config.AwsConfig.default, // uses real AWS dynamodb
      dynamodb.DynamoDb.live,
      DynamoDBExecutor.live
    )
}
```

--------------------------------

### Service with Dependencies in Scala

Source: https://zio.dev/reference/di

This example shows a Scala class 'Editor' that creates instances of its dependencies, 'Formatter' and 'Compiler', internally. This demonstrates a common scenario before applying dependency injection.

```scala
class Editor { val formatter = new Formatter val compiler = new Compiler  def formatAndCompile(code: String): UIO[String] =  formatter.format(code).flatMap(compiler.compile)}
```

--------------------------------

### Run the Application with sbt

Source: https://zio.dev/guides/tutorials/producing-consuming-data-from-kafka-topics

After cloning the project and setting up dependencies, run the application using sbt.

```bash
sbt run
```

--------------------------------

### Get the current value of a Ref

Source: https://zio.dev/reference/concurrency/ref

The `get` method retrieves the current value of the Ref. It's an effectful operation that can be chained with other effects using `flatMap` or for-comprehensions.

```scala
Ref.make("initial")   .flatMap(_.get)   .flatMap(current => Console.printLine(s"current value of ref: $current"))
```

```scala
for {
  ref   <- Ref.make("initial")
  value <- ref.get
} yield assert(value == "initial")
```

--------------------------------

### Build Project with SBT

Source: https://zio.dev/contributor-guidelines

Run a comprehensive build command with `sbt build`, which formats code, compiles, and runs tests. If already in an SBT session, simply type `build`.

```bash
sbt build
```

--------------------------------

### Unsafe State Management with get and set

Source: https://zio.dev/reference/concurrency/ref

Illustrates a non-concurrent-safe way to manage state by composing `get` and `set` operations. This can lead to race conditions in concurrent environments.

```scala
object UnsafeCountRequests extends ZIOAppDefault {
  def request(counter: Ref[Int]) = for {
    current <- counter.get
    _ <- counter.set(current + 1)
  } yield ()
  private val initial = 0
  private val myApp =
    for {
      ref <- Ref.make(initial)
      _ <- request(ref) zipPar request(ref)
      rn <- ref.get
      _ <- Console.printLine(s"total requests performed: $rn")
    } yield ()
  def run = myApp
}
```

--------------------------------

### ZIO Application with Service Dependencies

Source: https://zio.dev/reference/service-pattern

An example of a ZIO application that uses services like DocRepo, BlobStorage, and MetadataRepo. The application logic is defined within a for-comprehension, and the required services are provided using `app.provide`. This demonstrates how to assemble the application by layering different service implementations.

```scala
import zio._
import java.io.IOException

object MainApp extends ZIOAppDefault {

  val app = 
    for {
      docRepo <- ZIO.service[DocRepo]
      id      <- docRepo.save(
                    Doc(
                      "title",
                      "description",
                      "en",
                      "text/plain",
                      "content".getBytes()
                    )
                 )
      doc     <- docRepo.get(id)
      _       <- Console.printLine(
                   s"""
                     |Downloaded the document with $id id:
                     |  title: ${doc.title}
                     |  description: ${doc.description}
                     |  language: ${doc.language}
                     |  format: ${doc.format}
                     |""".stripMargin
                 )
        _       <- docRepo.delete(id)
      _       <- Console.printLine(s"Deleted the document with $id id")
    } yield ()

  def run = 
    app.provide(
      DocRepo.live,
      InmemoryBlobStorage.layer,
      InmemoryMetadataRepo.layer
    )
}
```

--------------------------------

### Unsafe Concurrent Logging with update and get

Source: https://zio.dev/reference/concurrency/ref

Demonstrates an unsafe pattern for logging request numbers concurrently by composing `update` and `get`. This can lead to non-deterministic output due to race conditions.

```scala
def request(counter: Ref[Int]) = {
  for {
    _  <- counter.update(_ + 1)
    rn <- counter.get
    _  <- Console.printLine(s"request number received: $rn")
  } yield ()
}
```

--------------------------------

### Rewrite Transfer Example with Scope

Source: https://zio.dev/reference/resource

This example demonstrates rewriting a transfer operation using ZIO.scoped to manage input and output streams. The resource is acquired within the ZIO.scoped block and used for copying data.

```scala
def transfer(from: String, to: String): ZIO[Any, Throwable, Unit] = {
  val resource = for {
    from <- ZIO.acquireRelease(is(from))(close)
    to   <- ZIO.acquireRelease(os(to))(close)
  } yield (from, to)
  ZIO.scoped {
    resource.flatMap { case (in, out) =>
      copy(in, out)
    }
  }
}
```

--------------------------------

### Implement DocRepo Service (Basic)

Source: https://zio.dev/reference/service-pattern/service-pattern

A basic implementation of the DocRepo service. Actual logic for each method is left as an unimplemented placeholder (???).

```scala
final class DocRepoLive() extends DocRepo {
  override def get(id: String): ZIO[Any, Throwable, Doc] = ???

  override def save(document: Doc): ZIO[Any, Throwable, String] = ???

  override def delete(id: String): ZIO[Any, Throwable, Unit] = ???

  override def findByTitle(title: String): ZIO[Any, Throwable, List[Doc]] = ???
}
```

--------------------------------

### Example of a Complex Log Format String

Source: https://zio.dev/zio-logging/formatting-log-records

This example demonstrates a comprehensive log format string that includes timestamp, level, fiber ID, message, cause, and highlights specific parts with custom formatting and labels.

```text
%timestamp %level [%fiberId] %name:%line %message %cause%highlight{%timestamp{yyyy-MM-dd'T'HH:mm:ssZ} %fixed{7}{%level} [%fiberId] %name:%line %message %cause}%label{timestamp}{%fixed{32}{%timestamp}} %label{level}{%level} %label{thread}{%fiberId} %label{message}{%message} %label{cause}{%cause}
```

--------------------------------

### Sync GitHub Data with ZIO SBT GitHub Query

Source: https://zio.dev/zio-sbt

Fetch data from GitHub and build/update the search database. Use `--force` for a full rebuild.

```bash
sbt gh-sync
```

```bash
sbt "gh-sync --force"
```

--------------------------------

### Get Fresh Layers for Service Isolation

Source: https://zio.dev/reference/di/examples

Demonstrates using `ZLayer#fresh` to ensure that each service gets its own isolated instance of a shared dependency, preventing potential conflicts. This is crucial when a dependency, like a cache, should not be shared between different parts of the application.

```scala
object MainApp extends ZIOAppDefault {

  val layers: ZLayer[Any, Throwable, UserRepo & DocumentRepo] = 
    ((InmemoryCache.layer.fresh ++ DatabaseLive.layer) >>> UserRepoLive.layer) ++ 
      ((InmemoryCache.layer.fresh ++ BlobStorageLive.layer) >>> DocumentRepoLive.layer)

  def run = myApp.provideLayer(layers)
}

// Output:
// initialized: zio.examples.InmemoryCache@13c9672b
// initialized: zio.examples.InmemoryCache@26d79027
```

--------------------------------

### Basic Automatic Derivation with HOCON

Source: https://zio.dev/zio-config/automatic-derivation-of-config

Demonstrates loading configuration using automatic derivation with different HOCON string inputs. Ensure necessary imports for ZIO Config and Magnolia are present.

```scala
import zio._
import zio.config._, import zio.config.typesafe._
import zio.config.magnolia._

import X._

// Defining different possibility of HOCON source
val aHoconSource =  ConfigProvider    .fromHoconString("x = A")

val bHoconSource =  ConfigProvider    .fromHoconString("x = B")

val cHoconSource =  ConfigProvider    .fromHoconString("x = C")

val dHoconSource =  ConfigProvider    .fromHoconString(
      s"""         |
         | x {
         |   DetailsWrapped {
         |    detail  {
         |      firstName : ff
         |      lastName  : ll
         |      region {
         |        city   : syd
         |        suburb : strath
         |     }
         |   }
         |}
         |}
         |""".stripMargin    )

// Let's try automatic derivation
aHoconSource.load(deriveConfig[MyConfig])
// res0: Right(MyConfig(A))
bHoconSource.load(deriveConfig[MyConfig])
// res0: Right(MyConfig(B))
cHoconSource.load(deriveConfig[MyConfig])
// res0: Right(MyConfig(C))
dHoconSource.load(deriveConfig[MyConfig])
// res0: Right(MyConfig(DetailsWrapped(Detail("ff", "ll", Region("strath", "syd")))))
```

--------------------------------

### ZPipeline.dropWhile

Source: https://zio.dev/reference/stream/zpipeline

Creates a pipeline that starts consuming and emitting values only after a predicate is no longer met.

```APIDOC
## ZPipeline.dropWhile

### Description
Creates a pipeline that starts consuming values only when one fails the given predicate.

### Example
```scala
ZStream(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
  .via(ZPipeline.dropWhile((x: Int) => x <= 5))
// Output: 6, 7, 8, 9, 10
```

### Note
There is also `dropWhileZIO` which takes an effectful predicate `p: I => ZIO[R, E, Boolean]`.
```

--------------------------------

### Imperative Loop Example

Source: https://zio.dev/reference/control-flow

Demonstrates a traditional imperative `while` loop in Scala, which relies on mutable state.

```scala
object MainApp extends scala.App {
  def printNumbers(from: Int, to: Int): Unit = {
    var i = from
    while (i <= to) {
      println(s"$i")
      i = i + 1
    }
  }
  printNumbers(1, 3)
}
```

--------------------------------

### Manual Pipelining with Fibers and Queues

Source: https://zio.dev/reference/stream/index

Illustrates manual data pipelining using ZIO fibers and bounded queues. This approach is complex and requires careful handling of interruptions, resources, and shutdown.

```scala
def writeToInput(q: Queue[Int]): Task[Unit]                            = ZIO.succeed(???)
def processBetweenQueues(from: Queue[Int], to: Queue[Int]): Task[Unit] = ZIO.succeed(???)
def printElements(q: Queue[Int]): Task[Unit]                           = ZIO.succeed(???)

for {
  input  <- Queue.bounded[Int](16)
  middle <- Queue.bounded[Int](16)
  output <- Queue.bounded[Int](16)
  _      <- writeToInput(input).fork
  _      <- processBetweenQueues(input, middle).fork
  _      <- processBetweenQueues(middle, output).fork
  _      <- printElements(output).fork
} yield ()
```

--------------------------------

### Define BlobStorage Service Trait

Source: https://zio.dev/reference/service-pattern/accessor-methods

Defines the interface for a BlobStorage service with get and put operations.

```scala
trait BlobStorage {
  def get(id: String): ZIO[Any, Throwable, Array[Byte]]

  def put(content: Array[Byte]): ZIO[Any, Throwable, String]
}
```

--------------------------------

### Running a ZIO Effect with ZIOAppDefault

Source: https://zio.dev/reference/core/zioapp

Use ZIOAppDefault to define the main entry point for your ZIO application. The `run` method orchestrates the execution of ZIO effects.

```scala
object MyApp extends ZIOAppDefault {
  def run = for {
    _ <- Console.printLine("Hello! What is your name?")
    n <- Console.readLine
    _ <- Console.printLine("Hello, " + n + ", good to meet you!")
  } yield ()
}
```

--------------------------------

### ZIO Environment-Based Dependency Access

Source: https://zio.dev/reference/contextual/index

Demonstrates how ZIO's environment simplifies dependency management by allowing functions to access services via ZIO.service.

```scala
def foo(arg1: String, arg2: String, arg3: Int): ZIO[Service1 & Service2 & Service3, Throwable, Int] = 
  for {
    s1 <- ZIO.service[Service1]
    s2 <- ZIO.service[Service2] 
      ...
  } yield ()

def bar(arg1: Int, arg2: String, arg3: Double, arg4: Int): ZIO[Service1 & Service12 & Service18 & ServiceN, Throwable, Unit] =
  for {
    s1  <- ZIO.service[Service1] 
    s12 <- ZIO.service[Service12]
      ...
  } yield ()
```

--------------------------------

### Documenting Configuration with `describe` Annotation

Source: https://zio.dev/zio-config/automatic-derivation-of-config

Illustrates how to add documentation to automatically derived configurations using the `describe` annotation. This annotation can be applied at the class level or field level to provide descriptions that are equivalent to manual configuration.

```scala
import zio.config.magnolia.describe

@describe("This config is about aws")
case class Aws(region: String, dburl: DbUrl)
case class DbUrl(value: String)

// Equivalent manual configuration:
// string("region").zip(string("dburl").to[DbUrl]).to[Aws] ?? "This config is about aws"
```

```scala
case class Aws(@describe("AWS region") region: String, dburl: DbUrl)

// Equivalent manual configuration:
// (string("region") ?? "AWS region" zip string("dburl").to[DbUrl]).to[Aws] ?? "This config is about aws"
```

--------------------------------

### Broadcasting a Message to Multiple Subscribers

Source: https://zio.dev/reference/concurrency/hub

An example demonstrating how to use a Hub to broadcast a message to multiple subscribers.

```APIDOC
## Example: Broadcasting to Multiple Subscribers

### Description
This example shows how to create a bounded Hub, subscribe two consumers to it, publish a message, and have both consumers receive the message.

### Code
```scala
Hub.bounded[String](2).flatMap {
  hub =>
    ZIO.scoped {
      hub.subscribe.zip(hub.subscribe).flatMap {
        case (left, right) =>
          for {
            _ <- hub.publish("Hello from a hub!")
            _ <- left.take.flatMap(Console.printLine(_))
            _ <- right.take.flatMap(Console.printLine(_))
          } yield ()
      }
    }
}
```
```

--------------------------------

### update

Source: https://zio.dev/reference/concurrency/ref

Atomically updates the state of a Ref with a pure function. This operation is concurrent-safe and should be preferred over a get followed by a set.

```APIDOC
## update

### Description
Atomically updates the state of `Ref` with a given pure function.

### Signature
```scala
def update(f: A => A): IO[E, Unit]
```

### Example
```scala
import zio.Ref
import zio.IO

val counterInitial = 0
for {
  counterRef <- Ref.make(counterInitial)
  _          <- counterRef.update(_ + 1)
  value <- counterRef.get
} yield assert(value == 1)
```

### Caution
`update` is not the composition of `get` and `set`. This composition is not concurrent-safe. Whenever we need to update our state, we should use the `update` operation which modifies its `Ref` atomically.
```

--------------------------------

### Launch sbt with Increased Heap Size

Source: https://zio.dev/contributor-guidelines

Launch the sbt build tool with an increased maximum heap size to potentially avoid performance issues during compilation or dependency resolution.

```bash
sbt -J-Xmx8g
```

--------------------------------

### Run Tests with SBT

Source: https://zio.dev/contributor-guidelines

Execute all project tests using the `sbt test` command. If already in an SBT session, simply type `test`.

```bash
sbt test
```

--------------------------------

### Example Logs with Correlation ID Annotation

Source: https://zio.dev/guides/tutorials/enable-logging-in-a-zio-application

Sample log output showing that log lines are annotated with the `correlation-id` from the request header.

```scala
timestamp=2022-06-03T10:13:18.334468Z level=INFO thread=#zio-fiber-32 message="POST /users -d {\"name\": \"John\", \"age\": 42}" register-user=1ms location=dev.zio.quickstart.users.UserApp.apply.applyOrElse file=UserApp.scala line=22 correlation-id=f798d2f2-abf2-46ff-b3f4-ae1888256706
timestamp=2022-06-03T10:13:18.335034Z level=INFO thread=#zio-fiber-32 message="User registered: ec02143a-8030-4c70-a110-a497617c5c72" register-user=2ms location=dev.zio.quickstart.users.UserApp.apply.applyOrElse file=UserApp.scala line=35 correlation-id=f798d2f2-abf2-46ff-b3f4-ae1888256706
```

--------------------------------

### DocRepo Live Implementation with Dependencies

Source: https://zio.dev/reference/service-pattern

Provides a concrete implementation of the DocRepo service, utilizing MetadataRepo and BlobStorage services for its operations.

```scala
final class DocRepoLive(
  metadataRepo: MetadataRepo,
  blobStorage: BlobStorage
) extends DocRepo {
  override def get(id: String): ZIO[Any, Throwable, Doc] = 
    (metadataRepo.get(id) <&> blobStorage.get(id)).map {
      case (metadata, content) =>
        Doc(
          title = metadata.title,
          description = metadata.description,
          language = metadata.language,
          format = metadata.format,
          content = content
        )
    }
      override def save(document: Doc): ZIO[Any, Throwable, String] = 
    for {
      id       <- blobStorage.put(document.content)
      metadata = Metadata(
        title = document.title,
        description = document.description,
        language = document.language,
        format = document.format
      )
      _        <- metadataRepo.put(id, metadata)
    } yield id
  override def delete(id: String): ZIO[Any, Throwable, Unit] = blobStorage.delete(id) &> metadataRepo.delete(id).unit
  override def findByTitle(title: String): ZIO[Any, Throwable, List[Doc]] = 
    for {
      metadatas <- metadataRepo.findByTitle(title)
      content   <- ZIO.foreachPar(metadatas) { (id, metadata) => 
                     blobStorage
                       .get(id)
                       .map {
                         content =>
                           val doc = Doc(
                             title = metadata.title,
                             description = metadata.description,
                             language = metadata.language,
                             format = metadata.format,
                             content = content
                           )
                                             id -> doc
                       }
                   }
    } yield content.values.toList
}
```

--------------------------------

### Define BlobStorage Service Trait

Source: https://zio.dev/reference/service-pattern/service-pattern

Defines the interface for the BlobStorage service, outlining methods for getting, putting, and deleting binary data.

```scala
trait BlobStorage {
  def get(id: String): ZIO[Any, Throwable, Array[Byte]]

  def put(content: Array[Byte]): ZIO[Any, Throwable, String]

  def delete(id: String): ZIO[Any, Throwable, Unit]
}
```

--------------------------------

### Define MetadataRepo Service Trait

Source: https://zio.dev/reference/service-pattern/service-pattern

Defines the interface for the MetadataRepo service, outlining methods for getting, putting, deleting, and finding metadata.

```scala
trait MetadataRepo {
  def get(id: String): ZIO[Any, Throwable, Metadata]

  def put(id: String, metadata: Metadata): ZIO[Any, Throwable, Unit]

  def delete(id: String): ZIO[Any, Throwable, Unit]

  def findByTitle(title: String): ZIO[Any, Throwable, Map[String, Metadata]]
}
```

--------------------------------

### Define DocRepo Service Trait

Source: https://zio.dev/reference/service-pattern/service-pattern

Defines the interface for the DocRepo service, outlining methods for getting, saving, deleting, and finding documents.

```scala
trait DocRepo {
  def get(id: String): ZIO[Any, Throwable, Doc]

  def save(document: Doc): ZIO[Any, Throwable, String]

  def delete(id: String): ZIO[Any, Throwable, Unit]

  def findByTitle(title: String): ZIO[Any, Throwable, List[Doc]]
}
```