IronMQ
The IronMQ connector provides an Akka stream source and sink to connect to the IronMQ queue.
IronMQ is a simple point-to-point queue, but it is possible to implement a fan-out semantic by configure the queue as push queue and set other queue as subscribers. More information about that could be found on IronMQ documentation
Project Info: Alpakka IronMQ | |
---|---|
Artifact | com.lightbend.akka
akka-stream-alpakka-ironmq
1.1.2
|
JDK versions | OpenJDK 8 |
Scala versions | 2.12.7, 2.11.12 |
JPMS module name | akka.stream.alpakka.ironmq |
License | |
Readiness level |
Since 0.8, 2017-05-05
|
Home page | https://doc.akka.io/docs/alpakka/current |
API documentation | |
Forums | |
Release notes | In the documentation |
Issues | Github issues |
Sources | https://github.com/akka/alpakka |
Artifacts
- sbt
libraryDependencies += "com.lightbend.akka" %% "akka-stream-alpakka-ironmq" % "1.1.2"
- Maven
<dependency> <groupId>com.lightbend.akka</groupId> <artifactId>akka-stream-alpakka-ironmq_2.12</artifactId> <version>1.1.2</version> </dependency>
- Gradle
dependencies { compile group: 'com.lightbend.akka', name: 'akka-stream-alpakka-ironmq_2.12', version: '1.1.2' }
The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.
- Direct dependencies
Organization Artifact Version License com.typesafe.akka akka-http_2.12 10.1.1 com.typesafe.akka akka-http_2.12 10.1.10 Apache-2.0 com.typesafe.akka akka-stream_2.12 2.5.12 com.typesafe.akka akka-stream_2.12 2.5.23 Apache License, Version 2.0 de.heikoseeberger akka-http-circe_2.12 1.21.0 Apache-2.0 org.scala-lang scala-library 2.12.7 BSD 3-Clause - Dependency tree
com.typesafe.akka akka-http_2.12 10.1.10 Apache-2.0 com.typesafe.akka akka-http-core_2.12 10.1.10 Apache-2.0 com.typesafe.akka akka-parsing_2.12 10.1.10 Apache-2.0 org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause com.typesafe.akka akka-stream_2.12 2.5.23 Apache License, Version 2.0 com.typesafe.akka akka-actor_2.12 2.5.23 Apache License, Version 2.0 com.typesafe config 1.3.3 Apache License, Version 2.0 org.scala-lang.modules scala-java8-compat_2.12 0.8.0 BSD 3-clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause com.typesafe.akka akka-protobuf_2.12 2.5.23 Apache License, Version 2.0 org.scala-lang scala-library 2.12.7 BSD 3-Clause com.typesafe ssl-config-core_2.12 0.3.7 Apache-2.0 com.typesafe config 1.3.3 Apache License, Version 2.0 org.scala-lang.modules scala-parser-combinators_2.12 1.1.1 BSD 3-clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.reactivestreams reactive-streams 1.0.2 CC0 org.scala-lang scala-library 2.12.7 BSD 3-Clause de.heikoseeberger akka-http-circe_2.12 1.21.0 Apache-2.0 com.typesafe.akka akka-http_2.12 10.1.10 Apache-2.0 com.typesafe.akka akka-http-core_2.12 10.1.10 Apache-2.0 com.typesafe.akka akka-parsing_2.12 10.1.10 Apache-2.0 org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause com.typesafe.akka akka-stream_2.12 2.5.23 Apache License, Version 2.0 com.typesafe.akka akka-actor_2.12 2.5.23 Apache License, Version 2.0 com.typesafe config 1.3.3 Apache License, Version 2.0 org.scala-lang.modules scala-java8-compat_2.12 0.8.0 BSD 3-clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause com.typesafe.akka akka-protobuf_2.12 2.5.23 Apache License, Version 2.0 org.scala-lang scala-library 2.12.7 BSD 3-Clause com.typesafe ssl-config-core_2.12 0.3.7 Apache-2.0 com.typesafe config 1.3.3 Apache License, Version 2.0 org.scala-lang.modules scala-parser-combinators_2.12 1.1.1 BSD 3-clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.reactivestreams reactive-streams 1.0.2 CC0 org.scala-lang scala-library 2.12.7 BSD 3-Clause io.circe circe-core_2.12 0.9.3 Apache 2.0 io.circe circe-numbers_2.12 0.9.3 Apache 2.0 org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel cats-core_2.12 1.0.1 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel cats-kernel_2.12 1.0.1 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel cats-macros_2.12 1.0.1 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel machinist_2.12 0.6.2 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-reflect 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel machinist_2.12 0.6.2 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-reflect 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause io.circe circe-jawn_2.12 0.9.3 Apache 2.0 io.circe circe-core_2.12 0.9.3 Apache 2.0 io.circe circe-numbers_2.12 0.9.3 Apache 2.0 org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel cats-core_2.12 1.0.1 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel cats-kernel_2.12 1.0.1 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel cats-macros_2.12 1.0.1 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel machinist_2.12 0.6.2 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-reflect 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.typelevel machinist_2.12 0.6.2 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-reflect 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.spire-math jawn-parser_2.12 0.11.1 MIT org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause org.scala-lang scala-library 2.12.7 BSD 3-Clause
Consumer
IronMQ can be used either in cloud or on-premise. Either way you need a authentication token and a project ID. These can be set in the application.conf
file.
alpakka.ironmq {
// The IronMg endpoint. It may vary due to availability zone and region.
endpoint = "https://mq-aws-eu-west-1-1.iron.io"
credentials {
// The IronMq project id
// project-id =
// The IronMq auth token
// token =
}
consumer {
// This is the max number of message to fetch from IronMq.
buffer-max-size = 100
// This is the threshold where fech other messages from IronMq
buffer-min-size = 25
// This is the time interval between each poll loop
fetch-interval = 250 milliseconds
// This is the amount of time the IronMq client will wait for a message to be available in the queue
poll-timeout = 0
// This is the amount of time a fetched message will be not available to other consumers
reservation-timeout = 30 seconds
}
}
The consumer is poll-based. It will poll every fetch-interval
milliseconds, waiting for poll-timeout
milliseconds to consume new messages and will push those downstream.
It supports both at-most-once and at-least-once semantics. In the first case the messages are deleted straight away after been fetched. In the latter case the messages piggy back a Committable
object that should be used to commit the message. Committing the message will cause the message to be deleted from the queue.
At most once
The consumer source is instantiated using the IronMqConsumer
IronMqConsumer
.
- Scala
-
import akka.stream.alpakka.ironmq.Message val source: Source[Message, NotUsed] = IronMqConsumer.atMostOnceSource(queueName, ironMqSettings) val receivedMessages: Future[immutable.Seq[Message]] = source .take(100) .runWith(Sink.seq)
- Java
-
import akka.stream.alpakka.ironmq.*; import akka.stream.alpakka.ironmq.javadsl.*; Source<Message, NotUsed> source = IronMqConsumer.atMostOnceSource(queueName, ironMqSettings); CompletionStage<List<Message>> receivedMessages = source.take(5).runWith(Sink.seq(), materializer);
At least once
To ensure at-least-once semantics, CommittableMessage
s need to be committed after successful processing which will delete the message from IronMQ.
- Scala
-
import akka.stream.alpakka.ironmq.scaladsl.CommittableMessage import akka.stream.alpakka.ironmq.Message val source: Source[CommittableMessage, NotUsed] = IronMqConsumer.atLeastOnceSource(queueName, ironMqSettings) val businessLogic: Flow[CommittableMessage, CommittableMessage, NotUsed] = Flow[CommittableMessage] // do something useful with the received messages val receivedMessages: Future[immutable.Seq[Message]] = source .take(100) .via(businessLogic) .mapAsync(1)(m => m.commit().map(_ => m.message)) .runWith(Sink.seq)
- Java
-
import akka.stream.alpakka.ironmq.*; import akka.stream.alpakka.ironmq.javadsl.*; Source<CommittableMessage, NotUsed> source = IronMqConsumer.atLeastOnceSource(queueName, ironMqSettings); Flow<CommittableMessage, CommittableMessage, NotUsed> businessLogic = Flow.of(CommittableMessage.class); // do something useful with the received messages CompletionStage<List<Message>> receivedMessages = source .take(5) .via(businessLogic) .mapAsync(1, m -> m.commit().thenApply(d -> m.message())) .runWith(Sink.seq(), materializer);
Producer
The producer is very trivial at this time, it does not provide any batching mechanism, but sends messages to IronMq as soon as they arrive to the stage.
The producer is instantiated using the IronMqProducer
IronMqProducer
. It provides methods to obtain either a Flow[PushMessage, Messages.Id, NotUsed]
Flow<PushMessage, Messages.Id, NotUsed>
or a Sink[PushMessage, NotUsed]
Sink<PushMessage, NotUsed>
.
Flow
The PushMessage
allows to specify the delay per individual message. The message expiration is set a queue level.
When using the Flow
the returned Messages.Ids
String
contains the ID of the pushed message, that can be used to manipulate the message. For each PushMessage
from the upstream you will have exactly one Message.Id
String
in downstream in the same order.
- Scala
-
import akka.stream.alpakka.ironmq.{Message, PushMessage} val messages: immutable.Seq[String] = (1 to messageCount).map(i => s"test-$i") val producedIds: Future[immutable.Seq[Message.Id]] = Source(messages) .map(PushMessage(_)) .via(IronMqProducer.flow(queueName, ironMqSettings)) .runWith(Sink.seq)
- Java
-
import akka.stream.alpakka.ironmq.*; import akka.stream.alpakka.ironmq.javadsl.*; CompletionStage<List<String>> producedIds = Source.from(messages) .map(PushMessage::create) .via(IronMqProducer.flow(queueName, ironMqSettings)) .runWith(Sink.seq(), materializer);
The producer also provides a committable aware Flow/Sink as Flow[(PushMessage, Committable), Message.Id, NotUsed]
Flow<CommittablePushMessage<Committable>, String, NotUsed>
. It can be used to consume a Flow from an IronMQ consumer or any other source that provides a commit mechanism.
- Scala
-
import akka.stream.alpakka.ironmq.{Message, PushMessage} import akka.stream.alpakka.ironmq.scaladsl.Committable val pushAndCommit: Flow[(PushMessage, Committable), Message.Id, NotUsed] = IronMqProducer.atLeastOnceFlow(targetQueue, ironMqSettings) val producedIds: Future[immutable.Seq[Message.Id]] = IronMqConsumer .atLeastOnceSource(sourceQueue, ironMqSettings) .take(messages.size) .map { committableMessage => (PushMessage(committableMessage.message.body), committableMessage) } .via(pushAndCommit) .runWith(Sink.seq)
- Java
-
import akka.stream.alpakka.ironmq.*; import akka.stream.alpakka.ironmq.javadsl.*; Flow<CommittablePushMessage<CommittableMessage>, String, NotUsed> pushAndCommit = IronMqProducer.atLeastOnceFlow(targetQueue, ironMqSettings); CompletionStage<List<String>> producedIds = IronMqConsumer.atLeastOnceSource(sourceQueue, ironMqSettings) .take(messages.size()) .map(CommittablePushMessage::create) .via(pushAndCommit) .runWith(Sink.seq(), materializer);
Sink
- Scala
-
import akka.stream.alpakka.ironmq.{Message, PushMessage} val messages: immutable.Seq[String] = (1 to messageCount).map(i => s"test-$i") val producedIds: Future[Done] = Source(messages) .map(PushMessage(_)) .runWith(IronMqProducer.sink(queueName, ironMqSettings))
- Java
-
import akka.stream.alpakka.ironmq.*; import akka.stream.alpakka.ironmq.javadsl.*; CompletionStage<Done> producedIds = Source.from(messages) .map(PushMessage::create) .runWith(IronMqProducer.sink(queueName, ironMqSettings), materializer);