---
# System prepended metadata

title: apache Camel
tags: [camel]

---

:::warning
:dromedary_camel: 基於企業整合模式EIP(enterprise integration patterns)設計的開源框架，使用component的方式接入第三方系統，便於快速簡單整合不同系統做處理及生產data
-> 直接使用URLs來串接不同的傳輸或訊息模型，像是HTTP、kafka、cxf
-> 包含有消息路由及轉換等特定領域語言(DSL)
:::

## :japanese_castle: 架構組成

- **CamelContext**，Camel的工作環境，一個application通常會有一個CamelContext對象，像是有：
        1. SpringCamelContext與Spring框架縫合，透過createRegistrt創建applicationContextRegistry instance，再從其中的applicationContext的getBean取得instance
        2. DefaultCamelContext運行所需的基本資訊，Service啟動時作為參數傳入，實現共享上下文的效果
- **Service**，大多數在Camel中的元素，都是一個獨立的service，像是Component、endpoint、routes等，透過CamelContext銜接，擁有獨立的生命週期，調用start()開始，stop()終止
    - CamelTemplate，對CamelContext作包裝，提供發送消息到endpoint的方法
        - **生產者模板**：producerTemplate、**fluentProducerTemplate(流式串接的寫法)**
            - 同步執行
                1. onlyIn模式: send(),sendBody(),sendBodyAndHeader()
                2. inOut模式: request(),requestBody(),requestBodyAndHeader()
            - 非同步執行 (返回future)
                1. onlyIn模式: asyncSend(),asyncSendBody()
                2. inOut模式: asyncRequestBody(),asyncRequestBodyAndHeader(),asyncRequestBodyAndHeaders()
                3. asyncCallback(),asyncCallbackSendBody(),asyncCallbackRequestBody(): 包含callback
        - **消費者模板**
            - consumerTemplate
                - 事件驅動模式 **event-driven consumer**: 常見的模式，像是client-server架構及網路服務，也會涉及到EIP模型的非同步receiver，監聽特定的messaging channel，像是TCP/IP port或Twitter handle, Amazon SQS queue或web socket等，等client送到message後，consumer啟動並開始處理
                - 輪詢及排程輪詢模式 **polling/scheduled polling consumer**; 和事件驅動相反，這種模式會主動的拉取特定source的message，像是FTP server，這種polling consumer在EIP模型稱為同步receiver，因為在完成現在的message之前不會再poll新的message；但一般來說是用scheduled polling consumer，像是File, FTP, and email components都是這樣處理，non-scheduled的polling consumer只在特地呼叫時才會使用
    - **endpoint**: 接受或發送數據的組件，指引channel要送到哪的model，支持多種協議，如jms, http, ftp, file
        - camel能傳送或接收message給endpoint
            - consumer (from)
            - producer (to)
            - or both
        - 作為建立producer跟consumers的factory
        - 使用URI描述目標系統的通訊 e.g. file:data/inbox?delay=5000
            - schema: 指示哪個camel component處理這種type的endpoint，像是範例會使用FileComponent
            - contextPath: 根據此context，指示該component建立endpoint
            - Options: 其它選項，像delay=5000表示每5秒poll一次
        - 實現org.apache.camel.Component interface建立
        - DIRECT component: `direct:someName[?options]`，當producer發送message exchange時，提供直接、同步啟動consumer的方法，此endpoint能使連接到同樣層級context的既存route
    - **processor**: 處理具體業務邏輯，接收來自Endpoint、Routing或別的processor的消息，需實作Process interface，用==exchange==型別傳遞，處理時截取出內含的message
        - 串接各endpoint
        - routing
        - 轉換、調整、檢核、追蹤、interception、mediation、紀錄...
        ```java
        public class testProcessor implements Processor{
            @Override
                public void process(Exchange exchange) throws Exception{
                    // 1. getIn(): 擷取傳入的exchange內的訊息
                    Message message = exchange.getIn(); 
                    String body = message.getBody.toString();
                    // ===
                    // 執行數據轉換的邏輯
                    // ===
                    if(exchange.getPattern() == ExchangePattern.Inout){
                        // a. 自建立out message並將in message帶入
                        // copy headers from IN to OUT to propagate them
                        exchange.getOut().setHeaders(exchange.getIn().getHeaders());
                        // getOut(): 擷取傳出的exchange內的訊息, 若無則建立一個空Message
                        Message.outMessage = excahnge.getOut();
                        outMessage.setBody(body + "|| other out")
                        // copy attachements from IN to OUT to propagate them
                       exchange.getOut().setAttachments(exchange.getIn().getAttachments());
                        
                        // b. 或是直接更改in message, camel偵測無out message的話會直接拿in做成out message
                        String body = exchange.getIn().getBody(String.class);
                        // change the existing message to say Hello
                        exchange.getIn().setBody("Hello " + body);
                    }
                }
        }
        ```
    - **exchange**: 訊息本體, 貫穿整條路徑執行的控制endpoint、處理器、表達式及路由條件判斷，內包含的資訊有：
        - exception
        - exchangeID
        - FromEndpoint
        - Properties: key/value結構配置共享的參數
        - Pattern(ExchangePattern)交換器工作模式: 為enum，描述消息的傳遞方式，包含InOnly, RobustInOnly, InOut, InOptionalOut, OutOnly, RobustOutOnly, OutIn, OutOptionalIn
        - Message IN/OUT: 當endpoint、processor之間的message在exchange中傳遞，exchange會自動將上一個輸出作為下一個輸入使用、包含MessageID、Attachment、Header、Body
        - 方法
            - Message exchange.getIn(): 擷取傳入的exchange內的訊息
            - Message exchange.getOut():擷取傳出的exchange內的訊息, 若無則建立一個空Message
            - Message exchange.hasOut(): 檢查是否有outMessage
            - Message ExchangePattern: enum, 分有單向的InOnly event message, 或是雙向的InOut Request Reply message
        - message: exchange的message部分，包含有body、headers、messageId、timestamp等，messageId預設等同於exchange.getExchangeId
    - **route**: 路由規則，指示processor之間的路由跳轉，支持多種規則如下:
        - content based router: 單一邏輯切轉至不同系統
        - message filter: 篩選掉不需要的message
        - dynamic router: 能透過條件表達式動態決定下個系統方向，完成後返回再次重新判斷直到沒有新的路徑為止
        - recipient router: 轉導message至靜態或動態清單的特定收件者
        - splitter: 拆解message，發送各element去不同的system
        - aggregator: 集結多個message再發送
        - recequencer: 將失序的message重排回正確順序
        - composed message processor: 在多元素且不同processor的情況，維持整體message的流向
        ---
        - routing code (主體設計為轉導，支援判斷及處理)
            - from(): route起始點，始於作為input source的endpoint
            - to(): 送message到靜態路由
            - toD(): 送message到動態路由
            - try..catch..finally: java姊妹款, 使用時則error handler不生效，如onException()
                - doTry()
                - doCatch()
                - doFinally()
                - end()
            ```java
            from("direct:test").routeId("myroute")
                .doTry().
                    doTry().
                        throwException(new IllegalArgumentException("Forced by me"))
                    .doCatch(Exception.class)
                        .log("docatch 1")
                        .throwException(new IllegalArgumentException("Second forced by me"))
                // 注意巢狀結構要確實endDoTry(), 否則內部exception不會被拋出
                     .endDoTry() // end this doCatch block
                 .endDoTry() // end the inner doTry
                .doCatch(Exception.class)
                    .log("docatch 2")
                .end();
            ```
            - setProperty(): 設定Exchange的property，以key/value的Map形式存放
            ```java
            from("direct:a")
                // 可以拿現有的exchange(bar) 設定到另一個exchange(foo)
                .setProperty("foo", exchangeProperty("bar"))
                .to("direct:b");
            ```
            - setHeader(): 設定message的header，可以使用固定(constant)或變動header name(搭配Simple()語法使用)，跟body不同，可以有多個欄位
            ```java
            from("direct:a")
                .setHeader("myHeader", constant("test"))
                .to("direct:b")
                .setHeader("randomNumber", simple("${random(1,100)}"))
                .to("direct:c");
            ```
            - setBody(): EIP patterns的message translator，轉換傳遞的內容(message)，從路徑或從excahange都能設定


            ```java
            from("direct:cheese")
                .setBody(simple("Hello ${body}"))
                .to("log:hello");
            ```

            ```java
            from("file:messages/foo").to("jms:queue:foo");
            from("direct:start").toD("${header.foo}"); // 抓message的header資訊動態導路
            from("direct:start").toD("language:xpath:/order/@uri"); // toD使用不同語言
            ```
            - choice(): 開啟分支路徑，可以巢狀結構，但不建議做的太複雜(it does not do it well)
                - when()
                - otherwise()
            ```java
            // 有3種情況: 
            // 1. 只有一個主要的.when()/.otherwise()區塊，可以不顯式關閉，最後用.end()關閉choice()
            // 若有巢狀結構，需顯式關閉(離開stack)，否則不能繼續往下跑，
            // 2. when()裡面包含多個choice，巢狀結構: 內層需顯式.endChoice()
            // 3. when()裡面包含複雜巢狀區塊，像是loadBalance()，內層.endChoice()
            // stack概念，所以要關閉when()/otherwise()區塊才能接著處理
            from("direct:start")
                .choice()
                    .when(body().contains("Camel"))
                        .loadBalance().roundRobin()
                            .to("mock:foo")
                            .to("mock:bar")
                            .choice()
                                .when(body().contains("people"))
                                    .to("mock:test")
                            .endchoice()
                        .endChoice()
                    .otherwise()
                        .to("mock:result");
                .end()
                .to("direct:b")
            ```
            - filter(expression 條件式篩選值)
            ```java
            from("direct:a")
                .filter(simple("${header.foo} == 'bar'"))
                    .to("direct:bar")
                .end()
                .to("direct:b")
            ```
            - onException(): 特定exception類型要執行的動作，
    - component(endpointFactory)
        - 當CamelContext調用getEndpoint()時，根據Endpoint的URI前綴(schema)來尋找對應的Component再調用其方法
        - 提供一致的endpoint interface
        - 作為其他系統的connector
    - 其它components
        - simple(): 從架構層面抽出language作為Expression及predicate的工廠，免除再導入其它dependencies或語言的學習(如Groovy)，simple是其中一個簡易的語言，內建許多變數像是exchange、body、header等，一般用來創建邏輯表達或一般表達式，在camel route使用少量scripting時適用，若是更為複雜的使用情境，則建議用Groovy, MVEL或OGNL 
        ```java
        simple("${header.title} contains 'Camel' && ${header.type'} == 'gold'")
            
        // 若header存在foo這個key, 則true, message會被轉導到下個路徑; 否則message is dropped
        from("seda:orders")
            .filter().simple("${header.foo}")
                .to("seda:fooOrders");
        ```

- 解決問題
    - 不同系統/數據來源的整合應用，如共享或交換資訊
    - 優化現有結構，使系統易於拓展或維護
    - 使系統獨立不互相干擾
- camel的message system
    - message endpoint
    - message channel
    - message
    - message route
    - message translator
    - pipes and filters
---
### :closed_book: 應用場景
1. 消息匯聚: 匯聚不同server的數據，如ActiveMQ、RabbitMQ...；
2. 消息分發: 將消息從生產者轉發給接收者，有squential跟parallel 2種分發方式
```java
from("amqp:queue:order")
    .to("uri:validateBean", "uri:handleBean", "uri:emailBean");

from("amqp:queue:order")
    .multicast()
        .to("uri:validateBean", "uri:handleBean", "uri:emailBean");
```
3. 消息轉換
```java
from("amqp:queue:order")
    .process(new XmlToJsonProcessor())
        .to("bean:orderHandler");
```
4. 規則引擎
```xml
<route>
    <from uri="amqp:queue:order"/>
    <multicast>
        <to uri="uri:validateBean"/>
        <to uri="uri:handleBean"/>
        <to uri="uri:emailBean"/>
    </multicast>
</route>

```
```java
from("amqp:queue:order")
    .filter(header("foo")
    .isEqualTo("bar"))
    .choice()
    .when(xpath("/person/city = &#39;London&#39;"))
      .to("file:target/messages/uk")
    .otherwise()
      .to("file:target/messages/others");
```



---
:mag: Spring 實作
---
```mermaid
graph TD;
創建CamelContext-->設定endpoints跟components;
設定endpoints跟components-->加上route規則;
加上route規則-->start啟動消息發送接收及處理;
start啟動消息發送接收及處理 --> stop關閉所有的endpoint跟執行緒
```
### 1. 引進套件
```java
    <!-- Dependency for camel -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-http4</artifactId>
    </dependency>
```
### 2. 利用camelContext建立producer template去拋message
```java
@Service
public class ProcessServiceImpl implements ProcessService {
    @Autowired
    private CamelContext camelContext;

    FluentProducerTemplate producer = camelContext.createFluentProducerTemplate();
    
    try {
        // 拋出message之後, 取得回覆內容
        Exchange exchange = producer.to("direct:callEaiGW").withBody(request).send();
        response = exchange.getOut().getBody(CubResponse.class);
    } catch (Exception e){
        logger.info("[Log] Exception message -> {}\n", e.getMessage(), e);
    } finally {
        try {
            // 完成後, 顯式將此service停用
            producer.stop();
        } catch(Exception){
            logger.info("Close producer fail!", e);
        }
    }
}  
```
### 3. route extends RouteBuilder 建立consumer處理路徑
```java
// 此為Java DSL方法，也可以用xml或yml定義路徑
// 這邊選用direct component
public class EAIRoute extends RouteBuilder {
    @Value("${eai.gw.url}")
    private String gwUrl;

    private String urlForGW;

    @Override
    public void configure() throws Exception {
        //Error Handle
        onException(Exception.class).process("ErrorHandleProcessor").handled(true);

        StringBuilder sb = new StringBuilder();
        sb.append(gwUrl).append("?bridgeEndpoint=true&clientConnectionManager=#connectionManager");
        urlForGW = sb.toString();

        // 由from接取producer傳來的message, 依序to到別的地方
       from("direct:callEaiGW").to("SetBodyProcessor").to(urlForGW).to("GetResultProcessor");
    }
}
```
### 4. **processor implements Processor定義各系統的執行內容(&錯誤處理)**
```java
@Component("SetBodyProcessor")
public class SetBodyProcessor implements Processor {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Utils utils;

    @Override
    public void process(Exchange exchange) throws Exception {
        try {
            Message message = exchange.getIn();
            CubRequest<?> body = message.getBody(CubRequest.class);
            MwHeader mw = body.getMwHeader();

            exchange.setProperty("Header", message.getHeaders());
            exchange.setProperty("Request", body);

            String json = utils.transferObjToJson(body);
            logger.info("[Log] Send Request -> {}, {}, {}", mw.getMsgid(), mw.getTxnseq(), json);

            message.setHeader(Exchange.CONTENT_TYPE, "application/json;charset=UTF-8");
            message.setBody(json);
        } catch (Exception e) {
            logger.info(e.getMessage(), e);
        }
    }
}
```

**REF**
1. [Apache Camel的认识和应用详解](https://blog.51cto.com/u_15459458/5158242)
2. [Apache Camel user manual](https://camel.apache.org/components/3.20.x/eips/choice-eip.html)
3. [Apache Camel](https://camel.apache.org/)
4. [Using .when(), .end(), .endChoice() in Apache Camel with examples](https://bvn13.medium.com/using-when-end-endchoice-in-apache-camel-with-examples-5e6e4f4f7f16)
5. [Camel end vs endChoice - not the usual query](https://stackoverflow.com/questions/32382322/camel-end-vs-endchoice-not-the-usual-query)
6. [Package org.apache.camel](https://www.javadoc.io/doc/org.apache.camel/camel-api/latest/org/apache/camel/package-summary.html)
