Skip to content

Instantly share code, notes, and snippets.

/post.scala

Created Jun 29, 2017
Embed
What would you like to do?
the description for this gist
def createSimple(
connection: Connection,
exchange: String
): Consumer[OutboundMessage, Unit] =
Consumer.fromObserver { s =>
new Observer[OutboundMessage] {
val ch = connection.createChannel()
val properties = new AMQP.BasicProperties()
def publish(m: OutboundMessage) =
ch.basicPublish(exchange, m.routingKey, properties, m.body)
def onNext(m: OutboundMessage): Future[Ack] =
Future {
blocking(publish(m))
Continue
}(s)
def onError(ex: Throwable): Unit = abort()
def onComplete(): Unit = abort()
private def abort(): Unit = if (ch.isOpen) ch.abort()
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.