RxJavaのGrails 3でObservableとしてRabbitMQキューを使用する

Question

Grails 3コントローラまたはサービスでRabbitMqをConsumerがObservableであるかのように消費する方法はありますか？ 私の実際のコードは、その

def consumer 
rx.stream { Subscriber subscriber -> 
    consumer = new DefaultConsumer(channel) { 
     @Override 
     void handleDelivery(String consumerTag, Envelope envelope, 
          AMQP.BasicProperties properties, byte[] body) throws IOException { 

      String message = new String(body, "UTF-8")      
      subscriber.onNext(rx.render(message)) 

     } 
    } 
    channel.basicConsume(queueName, true, consumer) 
} 

のようなものですが、私は本当に実装のこの種好きではないと私はRxのパラダイムに付着すると良いだろうと思います。

Answer 1

コールバックベースのモデルではなく、RabbitMQ側でプルベースのモデルを使用するようにコードを変更できますか？はいの場合は、SyncOnSubscribeを使用して簡単にメッセージを取得できます。