@Override
    public Publisher<Exchange> to(String uri, Object data) {
        String streamName = requestedUriToStream.computeIfAbsent(uri, camelUri -> {
            try {
                String uuid = context.getUuidGenerator().generateUuid();
                context.addRoutes(new RouteBuilder() {
                    @Override
                    public void configure() throws Exception {
                        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 testToFunction() throws Exception {
        context.start();

        Set<String> values = Collections.synchronizedSet(new TreeSet<>());
        CountDownLatch latch = new CountDownLatch(3);
        Function<Object, Publisher<String>> fun = crs.to("bean:hello", String.class);

        Flux.just(1, 2, 3)
                .flatMap(fun)
                .doOnNext(values::add)
                .doOnNext(res -> latch.countDown())
                .subscribe();

        assertTrue(latch.await(2, TimeUnit.SECONDS));
        assertEquals(new TreeSet<>(Arrays.asList("Hello 1", "Hello 2", "Hello 3")), values);
    }