### Question:
CamelSink implements Sink<T> { public CamelContext camelContext() { return camelContext; } CamelSink(String name, Config config); @Override String name(); @Override Completable dispatch(Message<T> message); CamelContext camelContext();  }### Answer:
@Test public void shouldWrapIntegersIntoCamelBodies(TestContext context) throws Exception { Async async = context.async(); CamelSink<Integer> sink = new CamelSink<>( null, new Config( new JsonObject().put("endpoint", "direct:test") ) ); CamelContext camelContext = sink.camelContext(); camelContext.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { from("direct:test").process(event -> { if (event.getIn().getBody(Integer.class) == 10) { context.assertEquals(event.getIn().getBody(Integer.class), 10); async.complete(); } }); } }); Source.from(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).to(sink); }