Mark：Camel SQL Route

Posted 2020-07-26

 1 <?xml version="1.0" encoding="UTF-8"?>
 2 <beans xmlns="http://www.springframework.org/schema/beans"
 3        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 4        xmlns:cxf="http://camel.apache.org/schema/cxf"
 5        xmlns:context="http://www.springframework.org/schema/context"
 6        xsi:schemaLocation="
 7        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
 8        http://camel.apache.org/schema/cxf http://camel.apache.org/schema/cxf/camel-cxf.xsd
 9        http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
10        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
11 
12     <context:property-placeholder/>
13 
14     <bean id="integrationMessageProcessor" class="com.exigen.dgig.message.IntegrationMessageProcessor"/>
15 
16     <bean id="flowExtractor" class="com.exigen.dgig.message.FlowIdExtractor"/>
17 
18     <bean id="routeStopper" class="com.exigen.dgig.message.RouteStopper"/>
19 
20     <bean id="shutdownStrategy" class="org.apache.camel.impl.DefaultShutdownStrategy">
21         <property name="timeout" value="1"/>
22     </bean>
23 
24     <cxf:cxfEndpoint id="publishEventClient"
25                      address="${PublishEventUrl}"
26                      serviceClass="com.desjardins.assurancedommages.sm840_integrationevenements.services.eventpublisher.v1.EventPublisherPortType"
27                      loggingFeatureEnabled="true"/>
28 
29     <camelContext id="dgigIntegrationLayer" xmlns="http://camel.apache.org/schema/spring">
30         <route id="policyIntegrationMessageRoute" shutdownRunningTask="CompleteCurrentTaskOnly">
31             <from uri="sql:SELECT * FROM PolicyIntegrationMessage WHERE messageId IS NULL ORDER BY id?dataSource=#dataSource&amp;delay=1s"/>
32             <log message="Found new message transactionId:${body[transactionId]} in DB" loggingLevel="INFO"/>
33             <log message="${body[messageBody]}" loggingLevel="TRACE"/>
34             <onException>
35                 <exception>java.lang.Throwable</exception>
36                 <redeliveryPolicy maximumRedeliveries="10" redeliveryDelay="1000" logRetryAttempted="true"
37                                   retryAttemptedLogLevel="WARN"/>
38                 <process ref="routeStopper"/>
39                 <log message="Route stopped" loggingLevel="ERROR"/>
40             </onException>
41             <setHeader headerName="dbMessageId">
42                 <simple>${body[id]}</simple>
43             </setHeader>
44             <process ref="integrationMessageProcessor"/>
45             <to uri="cxf:bean:publishEventClient"/>
46             <log message="Message sent to DGIG NEXT layer. Received location: ${headers.Location}" loggingLevel="INFO"/>
47             <setHeader headerName="messageId">
48                 <ref>flowExtractor</ref>
49             </setHeader>
50             <to uri="sql:UPDATE PolicyIntegrationMessage SET messageId=:#messageId WHERE id=:#dbMessageId?dataSource=#dataSource"/>
51         </route>
52         <route id="billingIntegrationMessageRoute" shutdownRunningTask="CompleteCurrentTaskOnly">
53             <from uri="sql:SELECT * FROM BillingIntegrationMessage WHERE messageId IS NULL ORDER BY id?dataSource=#dataSource&amp;delay=1s"/>
54             <log message="Found new message transactionId:${body[transactionId]} in DB" loggingLevel="INFO"/>
55             <log message="${body[messageBody]}" loggingLevel="TRACE"/>
56             <onException>
57                 <exception>java.lang.Throwable</exception>
58                 <redeliveryPolicy maximumRedeliveries="10" redeliveryDelay="1000" logRetryAttempted="true"
59                                   retryAttemptedLogLevel="WARN"/>
60                 <process ref="routeStopper"/>
61                 <log message="Route stopped" loggingLevel="ERROR"/>
62             </onException>
63             <setHeader headerName="dbMessageId">
64                 <simple>${body[id]}</simple>
65             </setHeader>
66             <process ref="integrationMessageProcessor"/>
67             <to uri="cxf:bean:publishEventClient"/>
68             <log message="Message sent to DGIG NEXT layer. Received location: ${headers.Location}" loggingLevel="INFO"/>
69             <setHeader headerName="messageId">
70                 <ref>flowExtractor</ref>
71             </setHeader>
72             <to uri="sql:UPDATE BillingIntegrationMessage SET messageId=:#messageId WHERE id=:#dbMessageId?dataSource=#dataSource"/>
73         </route>
74     </camelContext>
75 </beans>

View Code