the description for this gist

post.scala

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()
        }
      }