ZIO-JMS

ZIO-JMS adapts JMS API to ZIO streams and makes it working more conveniently and seamlessly integrates ZIO

---

Add library to your project

• Add Maven Central Repository to your build script
• Add dependency on the library "io.github.dobrynya" %% "zio-jms" % "0.2". zio-jms is available for Scala 2 & 3

Receive messages

For receiving messages it needs to create a message consumer using utility methods in JmsConsumer object

import zio.{ZIO, Chunk}
import javax.jms.{Connection, Message, JMSException}
import io.github.dobrynya.zio.jms._

val received: ZIO[Connection, JMSException, Chunk[Message]] = 
    JmsConsumer.consume(Queue("test-queue"))
                 .take(5)
                 .collect(onlyText)
                 .runCollect

You can process a stream of messages transactionally like follows

import zio.{ZIO, Chunk, UIO}
import javax.jms.{Connection, Message, JMSException}
import io.github.dobrynya.zio.jms._

def messageProcessor(message: Message): UIO[Unit] = ??? 

val received: ZIO[Connection, JMSException, Unit] = 
    JmsConsumer.consumeTx(Queue("test-queue"))
                 .take(5)
                 .tap(transactionalMessage => messageProcessor(transactionalMessage.message) <* transactionalMessage.commit)
                 .runDrain

Another ability to process input messages without using ZIO streams is more concise. A message is being acknowledged after successful processing

import zio._

import javax.jms.{Connection, JMSException, Message}
import io.github.dobrynya.zio.jms._
import zio.Console.printLine

def someMessageProcessor(message: Message): ZIO[Console, Exception, Unit] = 
    printLine(s"Received message $message")

val processing: ZIO[Console with Connection, Exception, Unit] = 
    JmsConsumer.consumeWith(Topic("test-topic"), someMessageProcessor)

In case of possible failures during processing a message I recommend using a transactional consumer which commits a message when it is processed successfully and rolls back when it fails

import zio.{ZIO, IO}
import javax.jms.{Connection, Message, JMSException}
import io.github.dobrynya.zio.jms._

def someMessageProcessor(message: Message): IO[String, Unit] = 
   IO.fail(s"Error occurred during processing a message $message")

val processing: ZIO[Connection, Any, Unit] = 
   JmsConsumer.consumeTxWith(Topic("test-topic"), someMessageProcessor)

Send messages

For sending messages it needs to create sinks providing a destination and a message encoder as follows

import zio.stream.ZStream
import io.github.dobrynya.zio.jms._

val messages = (1 to 100).map(i => s"Message $i")

ZStream.fromIterable(messages).run(JmsProducer.sink(Queue("test-queue"), textMessageEncoder))

Connection

The last thing is to provide a connection like follows

import zio._
import javax.jms.{Connection, ConnectionFactory, JMSException}
import io.github.dobrynya.zio.jms._

def connectionFactory: ConnectionFactory = ???

val connectionLayer: Layer[JMSException, Connection] =
  ZLayer.fromManaged(connection(connectionFactory))

val consuming = JmsConsumer
  .consume(Queue("test-queue"))
  .runDrain
  .provideSomeLayer(connectionLayer)

Request - Reply integration pattern

From a client's perspective

import io.github.dobrynya.zio.jms._
import zio.stream._

val request = Queue("request")
val response = Queue("response")

val messages: List[String] = ??? 

ZStream.fromIterable(messages)
    .run(JmsProducer.requestSink(request, response, textMessageEncoder))

From a server's perspective

import io.github.dobrynya.zio.jms._
import zio.stream._

val request = Queue("request")
JmsConsumer.consumeAndReplyWith(request,
  (message, session) => textMessageEncoder(onlyText(message).toUpperCase, session).asSome)

Here is a responder handling input messages and optionally responding to them to a destination specified in JMSReplyTo header. It automatically copy JMSCorellationID header before sending.