如何在camel-scala中建立从文件到websocket客户端的路由？

Question

我需要读取 /files 目录下的文件，然后在 websocket 客户端上显示该文件的内容。这就是我在 MyRouteBuilder.scala 中所做的：

import org.apache.camel.component.websocket.WebsocketComponent
import org.apache.camel.{LoggingLevel, CamelContext, Exchange}
import org.apache.camel.scala.dsl.builder.ScalaRouteBuilder

/**
 * A Camel Router using the Scala DSL
 */
class MyRouteBuilder(override val context : CamelContext) extends ScalaRouteBuilder(context) {

  // an example of a Processor method
  val myProcessorMethod = (exchange: Exchange) => {
    exchange.getIn.setBody("block test")
  }

  val ws = context.getComponent("websocket", classOf[WebsocketComponent]);
  ws.setPort(8444);
  ws.setHost("127.0.0.1")
  // we can serve static resources from the classpath: or file: system
  ws.setStaticResources("classpath:.");

  "file://files?noop=true" ==> {
    setBody(convertBodyTo(classOf[String]))
    to("websocket://127.0.0.1:8444/")
  }
}

但是，当我运行它时，它会给出以下堆栈跟踪：

堆栈跟踪

java.lang.IllegalArgumentException：无法将消息发送到单个连接；未设置连接键。 在 org.apache.camel.component.websocket.WebsocketProducer.process(WebsocketProducer.java:57) 在 org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61) 在 org.apache.camel.processor.SendProcessor.process(SendProcessor.java:141) 在 org.apache.camel.management.InstrumentationProcessor.process（InstrumentationProcessor.java:77） 在 org.apache.camel.processor.RedeliveryErrorHandler.process（RedeliveryErrorHandler.java:460） 在 org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190) 在 org.apache.camel.processor.Pipeline.process(Pipeline.java:121) 在 org.apache.camel.processor.Pipeline.process(Pipeline.java:83) 在 org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190) 在 org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:442) 在 org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:214) 在 org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:178) 在 org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:174) 在 org.apache.camel.impl.ScheduledPollConsumer.run（ScheduledPollConsumer.java:101） 在 java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 在 java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:745) [el-2) 线程 #0 - file://files] GenericFileOnCompletion WARN 回滚文件策略：org.apache.camel.component.file.strategy.GenericFileRenameProcessStrategy@72611f35 用于文件：GenericFile[audience.json] [el-2) 线程 #0 - file://files] DefaultErrorHandler 错误 (MessageId: ID-vaisakh-ubuntu-45785-1455531433908-1-11 on ExchangeId: ID-vaisakh-ubuntu-45785-1455531433908-1 -12)。传递尝试后耗尽：1 捕获：java.lang.IllegalArgumentException：无法将消息发送到单个连接；未设置连接键。

消息历史记录

RouteId ProcessorId 处理器已用（毫秒） [route1] [route1] [file://files?noop=true] [2] [route1] [convertBodyTo1] [convertBodyTo[java.lang.String]] [1] [route1] [setBody1] [setBody[{org.apache.camel.scala.ScalaExpression@f736069}]] [0] [route1] [to1] [websocket://127.0.0.1:8444/] [1]

交流

交换[ ID ID-vaisakh-ubuntu-45785-1455531433908-1-12 仅交换模式 标题 {breadcrumbId=ID-vaisakh-ubuntu-45785-1455531433908-1-11, CamelFileAbsolute=false, CamelFileAbsolutePath=/home/sagar/IdeaProjects/SampleIntegrationService/files/audience.json, CamelFileContentType=text/plain, CamelFileLastModified=1455526488000, CamelFileLength =150, CamelFileName=audience.json, CamelFileNameConsumed=audience.json, CamelFileNameOnly=audience.json, CamelFileParent=files, CamelFilePath=files/audience.json, CamelFileRelativePath=audience.json, CamelRedelivered=false, CamelRedeliveryCounter=0} BodyType org.apache.camel.scala.dsl.SRouteDefinition 正文 SRouteDefinition(Route(route1)[[From[file://files?noop=true]] -> [ConvertBodyTo[java.lang.String], SetBody[{org.apache.camel.scala.ScalaExpression@f736069}] , 到[websocket://127.0.0.1:8444/]]],com.mediaiqdigital.sampleIntegrationService.MyRouteBuilder@67a27caa) ]

Answer 1

通过在标题中设置连接键属性解决了这个问题。

 from(websocketUrl) ==> {

     setProperty("wsConnKey", header("websocket.connectionkey"))
     to("file://files?noop=true")
 }

 "file://files?noop=true" ==> {

     setBody(convertBodyTo(classOf[String]))
     to(WEBSOCKET_RESPONSE_ENDPOINT)
 }

 from(WEBSOCKET_REPONSE_ENDPOINT) ==> {

     process(new Processor() {
         void process(Exchange exchange) {
             Object connectionKey = exchange.getproperty("wsConnKey");
             Object exchangeBody = exchange.getIn().getBody();
             exchange.getOut().setHeader("wsConnKey", connectionKey);
             exchange.getOut().setBody(exchangeBody);
         }
     })
    to(websocketUrl)
 }