使用akka-camel和ActiveMQ请求回复

Question

更新：似乎更简单的测试用例不起作用：只是尝试通过进程内代理将消息从 ActiveMQ 生产者发送到 ActiveMQ 消费者。这是代码：

val brokerURL = "vm://localhost?broker.persistent=false"
val connectionFactory = new ActiveMQConnectionFactory(brokerURL)
val connection = connectionFactory.createConnection()
val session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)
val queue = session.createQueue("foo.bar")
val producer = session.createProducer(queue)
val consumer = session.createConsumer(queue)
val message = session.createTextMessage("marco")

producer.send(message)
val resp = consumer.receive(2000)
assert(resp != null)

我正在尝试使用akka-camel实现一个非常简单的请求-回复模式。这是我的（测试台）代码，它试图直接使用activeMQ发送消息并期望响应：

val brokerURL = "vm://localhost?broker.persistent=false"

// create in-process broker, session, queue, etc...
val connectionFactory = new ActiveMQConnectionFactory(brokerURL)
val connection = connectionFactory.createConnection()
val session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)
val queue = session.createQueue("myapp.somequeue")
val producer = session.createProducer(queue)
val tempDest = session.createTemporaryQueue()
val respConsumer = session.createConsumer(tempDest)
val message = session.createTextMessage("marco")
message.setJMSReplyTo(tempDest)
message.setJMSCorrelationID("myCorrelationID")

// create actor system with CamelExtension
val camel = CamelExtension(system)
val camelContext = camel.context
camelContext.addComponent("activemq", ActiveMQComponent.activeMQComponent(brokerURL))
val listener = system.actorOf(Props[Frontend])

// send a message, expect a response
producer.send(message)
val resp: TextMessage = respConsumer.receive(5000).asInstanceOf[TextMessage]
assert(resp.getText() == "polo")

我为消费者参与者尝试了两种不同的方法。第一个更简单，它尝试使用sender！响应：

class Frontend extends Actor with Consumer {
  def endpointUri = "activemq:myapp.somequeue"
  override def autoAck = false
  def receive = {
    case msg: CamelMessage => {
      println("received %s" format msg.bodyAs[String])
      sender ! "polo"
    }
  }
}

第二次尝试使用Camel模板回复：

class Frontend extends Actor with Consumer {
  def endpointUri = "activemq:myapp.somequeue"
  override def autoAck = false
  def receive = {
    case msg: CamelMessage => {
      println("received %s" format msg.bodyAs[String])
      val replyTo = msg.getHeaderAs("JMSReplyTo", classOf[ActiveMQTempQueue], camelContext)
      val correlationId = msg.getHeaderAs("JMSCorrelationID", classOf[String], camelContext)
      camel.template.sendBodyAndHeader("activemq:"+replyTo.getQueueName(), "polo", "JMSCorrelationID", correlationId)
    }
  }
}

我确实看到了我的actor的接收方法的println（）输出，所以ActiveMQ消息正在进入actor，但是我在测试平台中的respConsumer.receive（）调用上超时。我已经尝试了很多在回复中指定和不指定标题的组合。我还尝试启用和禁用自动确认。

提前谢谢。

Answer 1

事实证明，我需要在JMS代码中调用connection.start（）。