Vert.x Camel 网桥 (vertx-camel-bridge)

Apache Camel（http://camel.apache.org）是 Java 生态中久负盛名的开源企业集成模式（EIP）框架，致力于让异构系统间的互联集成更加优雅轻便。 Vert.x Camel 网桥建立起了 Vert.x 应用与 Camel 端点（Endpoint）之间的互通立交桥：

Vert.x 应用可以向 Camel 端点发送消息。
Vert.x 应用可以接收来自 Camel 端点的消息。

该网桥依托 Vert.x Event Bus，将特定的事件总线地址与 Camel 的 Endpoint 建立双向路由映射。

本组件不支持多语言跨端（Non-Polyglot），由于其深度依赖 Camel 原生 Java 类库，仅适用于 Java 开发环境。

引入依赖

要在项目中使用 Vert.x Camel 网桥，请在构建描述文件中添加以下依赖：

Maven（在 pom.xml 中）：

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-camel-bridge</artifactId>
  <version>5.0.12</version>
</dependency>

Gradle（在 build.gradle 文件中）：

compile 'io.vertx:vertx-camel-bridge:5.0.12'

网桥参数配置与启动

网桥在使用前必须完成映射配置并显式启动：

CamelContext camel = new DefaultCamelContext();
CamelBridge.create(vertx,
    new CamelBridgeOptions(camel)
        .addInboundMapping(InboundMapping.fromCamel("direct:stuff").toVertx("eventbus-address"))
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel("stream:out"))
).start();

网桥需要绑定一个 CamelContext 运行时上下文，并据此解析各类端点。请注意 start 方法是完全异步的，你可以利用返回的 Future 监听网桥启动就绪事件。

入站映射 (Inbound Mapping)

入站映射将一个 Camel 端点关联到一个 Event Bus 地址。在该 Camel 端点上接收到的消息将被自动转换为 Event Bus 消息并投递。

Endpoint endpoint = camel.getEndpoint("direct:foo");

CamelBridge.create(vertx,
    new CamelBridgeOptions(camel)
        .addInboundMapping(InboundMapping.fromCamel("direct:stuff").toVertx("eventbus-address"))
        .addInboundMapping(InboundMapping.fromCamel(endpoint).toVertx("eventbus-address"))
        .addInboundMapping(InboundMapping.fromCamel(endpoint).toVertx("eventbus-address")
            .withoutHeadersCopy())
        .addInboundMapping(InboundMapping.fromCamel(endpoint).toVertx("eventbus-address")
            .usePublish())
        .addInboundMapping(InboundMapping.fromCamel(endpoint).toVertx("eventbus-address")
            .withBodyType(String.class))
);

上述代码演示了入站映射的多种配置方式：

既可以通过 Camel 端点 URI 字符串配置，也可以直接传入 Endpoint 对象
可以禁用 Header 头信息的复制（默认会将 Camel Message Headers 自动复制到 Event Bus 消息中）
可以配置使用 publish 广播替代默认的点对点 send，将消息分发给所有消费者
可以显式配置 Event Bus 消息体的目标类型。若未指定，直接沿用 Camel 原始载荷；若指定，系统将在 Camel Context 中查找并调用对应的类型转换器（Converter）。

提示：org.fusesource.hawtbuf.Buffer 会被网桥自动透明转换为 Vert.x 原生的 Buffer。

如果使用 send（而非 publish），且 Camel Exchange 期望获得回包（In Out 请求-响应模式），Vert.x 代码将等待对发送消息的 Reply 应答。应答抵达后会自动传播回 Camel Exchange：

Endpoint endpoint = camel.getEndpoint("direct:stuff");

CamelBridge bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
    .addInboundMapping(new InboundMapping().setAddress("test-reply").setEndpoint(endpoint)));

vertx.eventBus().consumer("with-reply", message -> {
  message.reply("你最近怎么样？");
});

camel.start();
bridge.start();

ProducerTemplate template = camel.createProducerTemplate();
Future<Object> future = template.asyncRequestBody(endpoint, "hello");
String response = template.extractFutureBody(future, String.class);

你也可以通过 setTimeout 配置响应等待超时时间。

出站映射 (Outbound Mapping)

出站映射将一个 Event Bus 地址关联到一个 Camel 端点。发布或发送至该 Event Bus 地址的消息将被自动转换为 Camel 消息并转发至对应的 Camel 端点。

Endpoint endpoint = camel.getEndpoint("stream:out");

CamelBridge.create(vertx,
    new CamelBridgeOptions(camel)
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel("stream:out"))
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel(endpoint))
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel(endpoint)
            .withoutHeadersCopy())
        .addOutboundMapping(OutboundMapping.fromVertx("eventbus-address").toCamel(endpoint))
);

你可以将出站映射直接挂接至 Camel Route 路由：

camel.addRoutes(new RouteBuilder() {
  @Override
  public void configure() throws Exception {
    from("direct:start")
        .transform(constant("OK"));
  }
});

CamelBridge bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
    .addOutboundMapping(OutboundMapping.fromVertx("test").toCamel("direct:start")));

camel.start();
bridge.start();

vertx.eventBus().request("test", "hello").onComplete(reply -> {
  // 来自 Camel Route 的应答（此处为 "OK"）
});

若在 Event Bus 发送消息时注册了 Reply Handler 应答回调，网桥会自动将 Camel Exchange 配置为请求-响应模式。响应内容将填入 Reply Body 中；若 Camel Route 处理失败，Event Bus 将收到失败回执：

camel.addRoutes(new RouteBuilder() {
  @Override
  public void configure() throws Exception {
    from("direct:my-route")
        .to("http://localhost:8080");
  }
});

CamelBridge bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
    .addOutboundMapping(OutboundMapping.fromVertx("camel-route").toCamel("direct:my-route")));

camel.start();
bridge.start();

vertx.eventBus().request("camel-route", "hello").onComplete(reply -> {
  if (reply.succeeded()) {
    Object theResponse = reply.result().body();
  } else {
    Throwable theCause = reply.cause();
  }
});

若 Camel 路由中包含耗时的阻塞处理，必须将 blocking 属性设为 true。这能确保阻塞逻辑派发至 Worker 线程执行，避免阻塞 Event Loop 响应式事件循环：

camel.addRoutes(new RouteBuilder() {
  @Override
  public void configure() throws Exception {
    from("direct:my-route")
      .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
          // 执行耗时阻塞任务...
        }
      })
      .to("http://localhost:8080");
  }
});

CamelBridge bridge = CamelBridge.create(vertx, new CamelBridgeOptions(camel)
  .addOutboundMapping(OutboundMapping.fromVertx("camel-route").toCamel("direct:my-route").setBlocking(true)));

camel.start();
bridge.start();

vertx.eventBus().request("camel-route", "hello").onComplete(reply -> {
  if (reply.succeeded()) {
    Object theResponse = reply.result().body();
  } else {
    Throwable theCause = reply.cause();
  }
});

默认情况下阻塞任务提交给系统的通用 Worker 线程池，可以通过 setWorkerExecutor 定制指定的专用执行器。

停止网桥

在应用退出时，务必调用 stop 方法销毁网桥。stop 是异步方法，可通过返回的 Future 监听完全销毁事件。

传输自定义业务对象

若在 Event Bus 与 Camel 之间流转自定义 Java 对象，需在 Event Bus 上注册对应的编解码器（Codec）：

vertx.eventBus().registerDefaultCodec(Person.class, codec);

TCP Eventbus Bridge

Zipkin