标签: apache-camel

我一直试图让一个双向的ssl/https代理与Camel一起工作.我已经能够使用双向ssl设置Jetty组件,现在我正在尝试使用Http4组件来完成代理的客户端.

当我将码头流量路由到日志组件时,一切都很顺利,2路ssl信任链也没问题.当我抛出Http4组件时,它会爆发一个未经过身份验证的对等异常.我使用的是Camel 2.7.0

这是我到目前为止所拥有的

public static void main(String[] args) throws Exception {
    CamelContext context = new DefaultCamelContext();

    JettyHttpComponent jetty = context.getComponent("jetty", JettyHttpComponent.class);

    SslSelectChannelConnector sslConnector = new SslSelectChannelConnector();
    sslConnector.setPort(9443);
    sslConnector.setKeystore("/home/brian/jboss.keystore");
    sslConnector.setKeyPassword("changeit");
    sslConnector.setTruststore("/home/brian/jboss.truststore");
    sslConnector.setTrustPassword("changeit");
    sslConnector.setPassword("changeit");
    sslConnector.setNeedClientAuth(true);

    Map<Integer, SslSelectChannelConnector> connectors = new HashMap<Integer, SslSelectChannelConnector>();
    connectors.put(9443, sslConnector);

    jetty.setSslSocketConnectors(connectors);

    final Endpoint jettyEndpoint = jetty.createEndpoint("jetty:https://localhost:9443/service");

    KeyStore keystore = KeyStore.getInstance("PKCS12");
    keystore.load(new FileInputStream(new File("/home/brian/User2.p12")), "Password1234!".toCharArray());
    X509KeyManager keyManager = new CTSKeyManager(keystore, "user2", "Password1234!".toCharArray());
    KeyManager[] keyManagers = new KeyManager[] { keyManager };

    X509TrustManager trustManager = new EasyTrustManager();
    TrustManager[] trustManagers = …

我正在尝试构建一个具有多个camel路由的应用程序,这些路由在内部重用了许多常用路由.因此,我试图在几个不同的Route Builder类中隔离路由,然后在需要的地方连接路由.

例如,所有与发送电子邮件有关的路由都进入EmailRouteBuilder类,所有处理特定JMS队列的路由都进入MyQueueRouteBuilder类.我认为这应该没问题,因为Camel不区分类,只查找路由定义.

另外,我还将几个异常处理路由分组到一个单独的ExceptionHandlingRouteBuilder中.

我也通过在Spring中定义camel context来连接所有不同的类,如此 -

<camelContext id="camelContext" xmlns="http://camel.apache.org/schema/spring">
    <propertyPlaceholder id="properties" location="classpath:${env}/autoimport.properties"/>
    <!-- Common Routes -->
    <routeBuilder ref="emailRouteBuilder" />
    <routeBuilder ref="myQueueRouteBuilder" />
  <routeBuilder ref="httpRouteBuilder" />
    <routeBuilder ref="exceptionsRouteBuilder" />
    <routeBuilder ref="customer1RouteBuilder" />
    <routeBuilder ref="customer2RouteBuilder" />

</camelContext>

我的例外路由器包含许多例外条款,如 -

onException(ConnectException.class)
            .routeId("connectExceptionEP")
            .handled(true)
            .log("Caught Exception: ")
            .to("direct:gracefulExit");

..
..
..

但是,看起来在另一个类中定义的异常存在问题,或者就此而言,在主路由定义中单独定义.

我通过查找正在引导的路由(通过routeId)并在检查何时抛出异常时在日志中验证了这一点.

另外,为了进一步确认,我采用了http Connect异常处理路由并将其直接放在httpRouteBuilder中并且... ..!,异常处理现在可以解决这个异常.

我在这里错过了一些东西,以便在其自己的类中很好地定义所有异常.？

我正在使用Apache Camel 2.9.0,但我在2.8.3中也验证了相同的行为.

谢谢,阿南德

我是Apache Camel和模拟测试的新手,所以在这里......

我有一个没有XSD架构的XML,我没有影响.此XML的子元素包含我想要绑定到我的业务pojo的数据.这个POJO(WeatherCurrent)已经注释了JPA,我正在考虑添加JAXB注释,因此可以将拆分的XML映射到我的POJO.

由于这个XML有一个根元素,我只想要它的子元素(metData),我有一个问题如何注释我的POJO因为我不能使用@XmlRootElement.

这部分描述如下:http://camel.apache.org/splitter.html 使用Tokenizer语言章节在Streaming大XML有效负载上.我的POJO就像那个例子中的order xml元素.我只需要metData xml元素中的一些元素来映射到我的POJO字段.

在http://camel.apache.org/jaxb.html上还有一章Partial marshalling/unmarshalling,但没有JAVA DSL示例(必须),也没有如何注释pojo以使用XML片段.

到目前为止,我有这个测试代码:

import java.io.File;

import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.converter.jaxb.JaxbDataFormat;
import org.apache.camel.spi.DataFormat;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.junit.Test;

public class WeatherCurrentTest extends CamelTestSupport {

    @EndpointInject(uri = "file:src/test/resources")
    private ProducerTemplate inbox;

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                DataFormat jaxbDataFormat = new JaxbDataFormat("com.mycompany.model.entities.weather");// WARNING two packages …

我们使用Apache Camel作为编排引擎.通常,以下情形:

客户端发送HTTP请求< - > CAMEL代码< - >外部服务器

当我们的客户端向我们的CAMEL代码发送HTTP请求时,球开始滚动.Camel代码将通过REST HTTP调用触发外部服务器.最终,Camel代码会将回复发送回客户端.

在将响应发送回客户端之前的最后一个操作,Camel代码向外部服务器发送HTTP GET.因此首先设置TCP连接,然后发送数据.一段时间后(这可能需要5到10秒),外部服务器回复200 OK.

问题:收到200 OK后,Camel不会向外部服务器发送TCP FIN.结果,TCP连接保持打开状态...(外部服务器然后在超时200秒后关闭TCP连接,但这意味着TCP资源在200秒内丢失).

所以,在TCP级别,它是这样的:

Camel <---------->外部服务器

   TCP SYN  -->
   <-- TCP SYN,ACK
   TCP ACK  -->

   HTTP GET -->
   <-- 200 OK
   TCP ACK  -->

   <200 seconds later>
   <-- TCP FIN,ACK
   TCP ACK  -->

知道如何让Camel收到200 OK之后关闭TCP连接吗？

注意:我尝试添加"Connection:close"标题,但是Camel没有添加标题？!它似乎忽略了它......

这是添加标头的代码:

exchange.getOut().setHeader("Connection","Close");

我在使用Eclipse IDE的Spring框架中使用Camel 2.9.1.

最近我开始使用camel,我发现它可能解决了我的很多集成层需求.

我创建了一个java客户端应用程序(不在任何容器中运行),我定义了两个路由:

route1:将文件从传入的folder1移动到folder2

route2:将文件内容从folderx移动到mq队列.

我启动我的应用程序,这些路由正在他们的工作轮询这些文件夹并相应地路由消息.

任何人都可以解释我的路线如何运作.camel(context)是否为每个路由创建一个线程.究竟发生了什么？

注意:我在骆驼网站上找不到这方面的简单说明.

我开始使用Camel,我在编写测试时遇到了一些问题.我的用例与cfx代理示例完全相同.除了我不需要"RealWebservice".现在我正在尝试使用注释方法编写单元测试(不是示例中包含的集成测试):

@RunWith(CamelSpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "classpath:application-context.xml" })
@MockEndpointsAndSkip
public class RoutesTest {

@Autowired
CamelContext camelContext;

@EndpointInject(uri = "mock:cxf:bean:cxfEndpoint", context = "camelContext")
MockEndpoint cxfEndpoint;

@EndpointInject(uri = "mock:log:input", context = "camelContext")
MockEndpoint logInputEndpoint;

@EndpointInject(uri = "mock:http:realhostname:8211/service", context = "camelContext")
MockEndpoint realEndpoint;

@EndpointInject(uri = "mock:cxf:bean:cxfEndpoint")
ProducerTemplate producer;

@Test
public void testLeleuxMifidRoute() throws InterruptedException {
    String body = "<blah/>";

    cxfEndpoint.expectedBodiesReceived(body);
    logInputEndpoint.expectedBodiesReceived(body);
    realEndpoint.expectedBodiesReceived(body);

    producer.sendBody(body);

    MockEndpoint.assertIsSatisfied(camelContext);
}
}

cxfEndpoint接收消息但其他端点不接收消息.

路径看起来像这样(当我运行它并使用SoapUI发送消息时,它很有效,显然我在这个例子中混淆了ips和beannames):

<endpoint id="callRealWebService" uri="http://realhostname:8211/service?throwExceptionOnFailure=true" /> 
<route>
  <from uri="cxf:bean:cxfEndpoint?dataFormat=MESSAGE"/>
  <to uri="log:input?showStreams=true"/>
  <to ref="callRealWebService"/> …

我在使用Apache CAMEL实现Websphere MQ(WMQ)连接器时遇到了很多困难,该连接器可以无异常地处理MQ确认传递(CoD)报告,也不会以不需要的响应数据报的形式出现副作用.最后,我按照我想要的方式工作,如果您习惯于编写本机MQ客户端,那么这是一种非常标准和通用的方式.我在同一主题的帖子中记录了该方法,但我发现解决方案臃肿而复杂,并且非常感谢任何建议或示例,以使实现更清晰,更优雅.

我理解这个问题的根源在于MQ设计请求 - 回复消息交换模式(MEP)的方式,而不是JMS规范的方式,而不是其JMS组件中的请求 - 应答MEP的CAMEL实现.三种不同的哲学!

1. WMQ具有MessageType标头(请参阅MQMD字段和常量),其中请求值为1,回复为2,数据报为8(单向MEP).此外,值4用于以CoD(交付配置),PAN(肯定确认)和NAN(否定确认)的形式标记报告消息,其中 - 在消息流方面 - 也进行额外的回复信息.可以使用另一个名为"Report"的标题字段为Request消息,回复或数据报请求CoD,PAN和NAN ack,其中可以组合所有报告变体的标志.其他标头字段'ReplyToQ'和'ReplyToQMgr'指定原始发件人期望报告和回复的队列和队列管理器,以及固定的24字节'CorrelId'字段 - 可选 - 可以帮助将报告和回复与原始数据报或请求消息相关联.为了使其更复杂,实际上可以使用相同的原始消息ID和没有CorrelID发回回复和报告,或者在CorrelId中提供原始消息ID,或者在原始请求或数据报中已经指定时返回CorrelId值.IBM 通过WMQ提供JMS API,允许通过WMQ将纯JMS交换作为传输(在额外消息头名称MQRFH2的帮助下),或将本机MQ消息映射到JMS消息,反之亦然.
2. 另一方面,JMS规范提供了一个可选的'JMSReplyTo'头字段和一个'JMSCorrelationID',但是确实将MEP语义保留给客户端应用程序; 即在规格中陈述:" 答复可能是可选的;由客户自行决定. "
3. CAMEL以XML或Java DSL中的"路由"和内部Exchange对象模型为特征,目的是支持EIP模式,其中包括Request-Reply模式.然后CAMEL在其JMS组件中假定如果设置了JMSReplyTo字段,则这必然是期望回复的请求,导致Exchange的Out部分(或者如果Out为空,则部分修改)返回到JMSReplyTo中定义的队列. .

我愿意通过远程Websphere队列管理器支持本地MQ消息与交付确认(CoD)报告的交换,这样除了事务和持久性(即没有丢失,没有重复)之外,我还可以跟踪消息何时被消费并在发生延误时提出警报.

入境问题:

默认情况下,Websphere队列管理器在队列中的消息消耗完成时生成CoD报告.因此,在没有任何特定设置的情况下,当CAMEL端点消耗消息时,远程MQ客户端发送带有CoD标志的数据报(以及当时强制的ReplyToQ)将从队列管理器获得第一个回复作为MQ报告,然后是第二个(意外的)回复消息由CAMEL显式返回并包含CAMEL路由末尾的Exchange对象中剩余的内容,因为CAMEL假定存在JMSReplyTo字段的请求 - 应答EIP(从MQ ReplyToQ和ReplyToQMgr映射)被要求支持CoD回流).

出境问题:

如果没有特定设置,CAMEL默认也会在出站连接上采用请求 - 回复EIP/MEP.然后,CAMEL JMS/MQ端点将等待1个响应.当OUTbound消息是MQ上的JMS(因此具有MQRFH2头)时,这可以正常工作.当强制普通的vanilla MQ,即删除下面的MQRFH2头时,我无法使端点侦听器与相关的传入MQ报告匹配,尽管跟踪值看起来都是正确的(强制执行24个char相关ID以便截断更长的CorrelId值或空填充通过MQ无法对相关过滤器进行地理化处理).有没有人能够解决这个问题？

详细信息:虽然IBM JMS API接受传递特定的JMS属性值WMQ_MESSAGE_BODY = {1 | 0}/WMQ_TARGET_CLIENT = {1 | 0}来控制生成的消息中是否存在JMS头MQRFH2,但这些选项通过CAMEL变得不可用.必须使用CamelJmsDestinationName标头(如CAMEL JMS …

Camel中的wireTap和多播的默认行为似乎相似.这两者都有助于以不同的方式处理相同的消息.那么wireTap和组播之间的主要区别是什么？

刚开始,我是apache camel的新手.我有一个spring boot(MVC)+ mongodb项目已经运行正常,当我尝试将它与apache camel集成时,我在控制台上遇到异常.根据我对异常弹簧启动的理解,@Component当我@ComponentScan(basePackages="packagePath")在项目中的其他任何地方使用时,我试图说我不能在camel Route文件上使用默认注释,因为它是一个MVC架构,我有Controller,Service&Repository,我不能忽视,请帮我解决这个问题.

控制台例外

启动ApplicationContext时出错.要显示条件报告,请在启用"debug"的情况下重新运行应用程序.2018-03-15 17:17:55.426 ERROR 744 --- [main] osboot.SpringApplication
:应用程序运行失败

org.springframework.beans.factory.BeanDefinitionStoreException:无法处理配置类的导入候选[com.era.conf.ApplicationConfiguration]; 嵌套异常是java.lang.IllegalStateException:由于未找到org/springframework/boot/bind/RelaxedPropertyResolver,无法评估org.apache.camel.spring.boot.health.HealthCheckRoutesAutoConfiguration上的条件.确保您自己的配置不依赖于该类.(如果你把一个@ComponentScan在默认包错误EG)这也有可能发生,如果你是一个@ComponentScanning包springframework的在org.springframework.context.annotation.ConfigurationClassParser.processImports(ConfigurationClassParser.java:616)〜[弹簧context-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org.springframework.context.annotation.ConfigurationClassParser.processDeferredImportSelectors(ConfigurationClassParser.java:548)〜[spring-context-5.0.4.RELEASE.jar:5.0 .4.RELEASE]在org.springframework.context的org.springframework.context.annotation.ConfigurationClassParser.parse(ConfigurationClassParser.java:184)〜[spring-context-5.0.4.RELEASE.jar:5.0.4.RELEASE] .annotation.ConfigurationClassPostProcessor.processConfigBeanDefinitions(ConfigurationClassPostProcessor.java:316)〜[弹簧上下文5.0.4.RELEASE.jar:5.0.4.RELEASE]在org.springframework.context.annotation.ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(ConfigurationClassPostProcessor.java: 233)〜[s 普林-上下文5.0.4.RELEASE.jar:5.0.4.RELEASE]在org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(PostProcessorRegistrationDelegate.java:273)〜[弹簧上下文5.0.4.RELEASE.jar :5.0.4.RELEASE]在org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(PostProcessorRegistrationDelegate.java:93)〜[弹簧上下文5.0.4.RELEASE.jar:5.0.4.RELEASE]在org.springframework .context.support.AbstractApplicationContext.invokeBeanFactoryPostProcessors(AbstractApplicationContext.java:693)〜[弹簧上下文5.0.4.RELEASE.jar:5.0.4.RELEASE]在org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext. java:531)〜[spring-context-5.0.4.RELEASE.jar:5.0.4.RELEASE] org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:140)~ [spring org.springfra中的-boot-2.0.0.RELEASE.jar:2.0.0.RELEASE] orwork.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:388)中的mework.boot.SpringApplication.refresh(SpringApplication.java:752)[spring-boot-2.0.0.RELEASE.jar:2.0.0.RELEASE] org.springframework.boot.SpringApplication.run上的[spring-boot-2.0.0.RELEASE.jar:2.0.0.RELEASE](SpringApplication.java:327)[spring-boot-2.0.0.RELEASE.jar:2.0 .O.RELEASE] org.springframework.boot.SpringApplication.run(SpringApplication.java:1246)[spring-boot-2.0.0.RELEASE.jar:2.0.0.RELEASE] org.springframework.boot.SpringApplication.运行(SpringApplication.java:1234)[spring-boot-2.0.0.RELEASE.jar:2.0.0.RELEASE],位于com.era.conf.ApplicationConfiguration.main(ApplicationConfiguration.java:12)[classes /:na]引起:java.lang.IllegalStateException:由于未找到org/springframework/boot/bind/RelaxedPropertyResolver,无法评估org.apache.camel.spring.boot.health.HealthCheckRoutesAutoConfiguration上的条件.确保您自己的配置不依赖于该类.如果你是@ComponentScanning springframework包(例如,如果你错误地将@ComponentScan放在默认包中),org.springframework.boot.autoconfigure.condition.SpringBootCondition.matches(SpringBootCondition.java:55)〜[[]在org.springframework.context.annotation.ConditionEvaluator.shouldSkip(ConditionEvaluator.java:109)〜[spring-context-5.0.4.RELEASE]中的spring-boot-autoconfigure-2.0.0.RELEASE.jar:2.0.0.RELEASE .jar:5.0.4.RELEASE]在org.springframework.context.annotation.ConfigurationClassParser.processConfigurationClass(ConfigurationClassParser.java:217)〜[spring-context-5.0.4.RELEASE.jar:5.0.4.RELEASE] at org .springframework.context.annotation.ConfigurationClassParser.processImports(ConfigurationClassParser.java:606)〜[spring-context-5.0.4.RELEASE.jar:5.0.4.RELEASE] ...省略了15个常见帧引起的:java.lang .NoClassDefFoundError:org.apache.camel.spring.boot.util.HierarchicalPropertiesEvalu中的org/springframework/boot/bind/RelaxedPropertyResolver ator.isEnabled(HierarchicalPropertiesEvaluator.java:49)〜[camel-spring-boot-2.20.2.jar:2.20.2] at org.apache.camel.spring.boot.util.HierarchicalPropertiesEvaluator.evaluate(HierarchicalPropertiesEvaluator.java:42 )〜[camel-spring-boot-2.20.2.jar:2.20.2] org.apache.camel.spring.boot.util.GroupCondition.getMatchOutcome(GroupCondition.java:40)〜[camel-spring-boot- 2.20.2.jar:2.20.2] org.springframework.boot.autoconfigure.condition.SpringBootCondition.matches(SpringBootCondition.java:47)〜[spring-boot-autoconfigure-2.0.0.RELEASE.jar:2.0.0抛出java.lang.ClassNotFoundException:.RELEASE] ... 18个共同帧省略了致org.springframework.boot.bind.RelaxedPropertyResolver在java.net.URLClassLoader.findClass(URLClassLoader.java:381)〜[NA:1.8.0_102 ]在java.lang.ClassLoader.loadClass(ClassLoader.java:424)〜[na:1.8.0_102] at sun.misc.Launcher $ AppClassLoader.loadClass(Launcher.java:331)〜[na:1.8.0_102] at at java.lang.ClassLoader.loadClass(ClassLoader.java:357)〜[na:1.8.0_102] ... 22常见 帧被省略

POM

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-spring-boot-starter</artifactId>
            <version>2.20.2</version> …

我有以下情况:

• 有固定数量的组.
• 有一个传入消息的TCP流.每条消息都与一个组完全相关.

我启动Camel路线如下:

public class MyMessage implements Runnable {
    public void run() {
        // omitted here
    }
}

from("netty:tcp://localhost:7777?textline=true&sync=false")
   ... // omitted here: parse message to pojo MyMessage, set header "group-identifier"
   .to(seda:process);

此Camel路由使用TCP流,解析每个传入消息的有效负载并将其转换为MyMessagepojo,并group-identifier在交换机上设置与消息对应的标头...

现在我想消费seda:process如下:

• 属于同一组的消息不能同时执行.
• 属于不同组的消息可以同时执行.
• 每条消息都应该通过调用来执行run().我想为此提供/定义一个ExecutorService,所以我可以控制线程数.

我可以在这里应用哪些企业集成模式？我如何将这些概念映射到Camel？

我了解到ActiveMQ具有消息组的概念(http://activemq.apache.org/message-groups.html).这可能提供一种方法来确保同一组中的两条消息永远不会同时执行.虽然,我不确定仅为此引入ActiveMQ并不是一种矫枉过正.这可以通过'核心'Camel/Java来实现吗？