@Override
    public Publisher<Exchange> to(String uri, Object data) {
        String streamName = requestedUriToStream.computeIfAbsent(uri, camelUri -> {
            try {
                String uuid = context.getUuidGenerator().generateUuid();
                RouteBuilder.addRoutes(context, rb -> rb.from("reactive-streams:" + uuid).to(camelUri));

                return uuid;
            } catch (Exception e) {
                throw new IllegalStateException("Unable to create requested reactive stream from direct URI: " + uri, e);
            }
        });

        return toStream(streamName, data);
    }
@Test
    public void testToWithExchange() throws Exception {
        context.start();

        AtomicInteger value = new AtomicInteger();
        CountDownLatch latch = new CountDownLatch(1);

        Flowable.just(1, 2, 3).flatMap(e -> crs.to("bean:hello", e)).map(Exchange::getMessage).map(e -> e.getBody(String.class))
                .doOnNext(res -> assertEquals("Hello " + value.incrementAndGet(), res)).doOnNext(res -> latch.countDown())
                .subscribe();

        assertTrue(latch.await(2, TimeUnit.SECONDS));
    }