ZIO SQS is a ZIO-powered client for AWS SQS. It is built on top of the AWS SDK for Java 2.0 via the automatically generated wrappers from zio-aws.
ZIO SQS enables us to produce and consume elements to/from the Amazon SQS service. It is integrated with ZIO Streams, so we can produce and consume elements in a streaming fashion, element by element or micro-batching.
In order to use this library, we need to add the following line in our build.sbt file:
libraryDependencies += "dev.zio" %% "zio-sqs" % "0.10.0"In this example we produce a stream of events to the MyQueue and then consume them from that queue (at-most-once delivery semantics):
import zio._
import zio.aws.core.config.AwsConfig
import zio.aws.netty.NettyHttpClient
import zio.aws.sqs.Sqs
import zio.sqs.producer.{Producer, ProducerEvent}
import zio.sqs.serialization.Serializer
import zio.sqs.{SqsStream, SqsStreamSettings, Utils}
import zio.stream.ZStream
object ProducerConsumerExample extends ZIOAppDefault {
val queueName = "MyQueue"
val stream: ZStream[Any, Nothing, ProducerEvent[String]] =
ZStream.iterate(0)(_ + 1).map(_.toString).map(ProducerEvent(_))
val program: ZIO[Sqs, Throwable, Unit] = for {
_ <- Utils.createQueue(queueName)
queueUrl <- Utils.getQueueUrl(queueName)
producer = Producer.make(queueUrl, Serializer.serializeString)
_ <- ZIO.scoped(producer.flatMap(_.sendStream(stream).runDrain))
_ <- SqsStream(
queueUrl,
SqsStreamSettings.default
.withAutoDelete(true) // at-most-once delivery semantics
.withStopWhenQueueEmpty(true)
.withWaitTimeSeconds(3)
).foreach(msg => Console.printLine(msg.body))
} yield ()
def run =
program.provide(
Sqs.live,
AwsConfig.default,
NettyHttpClient.default
)
}Check out the examples folder in zio-sqs/src/test/scala/examples for additional examples that cover at-least-once and at-most-once delivery semantics.
Learn more on the ZIO SQS homepage!
For the general guidelines, see ZIO contributor's guide.
See the Code of Conduct
Come chat with us on 