@Override
    public Publisher<Exchange> from(String uri) {
        final String name = publishedUriToStream.computeIfAbsent(uri, camelUri -> {
            try {
                String uuid = context.getUuidGenerator().generateUuid();

                RouteBuilder.addRoutes(context, rb -> rb.from(camelUri).to("reactive-streams:" + uuid));

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

        return fromStream(name);
    }
@Test
    public void testFrom() throws Exception {
        context.start();

        Publisher<Exchange> timer = crs.from("timer:reactive?period=250&repeatCount=3&includeMetadata=true");

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

        Flux.from(timer)
                .map(exchange -> ExchangeHelper.getHeaderOrProperty(exchange, Exchange.TIMER_COUNTER, Integer.class))
                .doOnNext(res -> assertEquals(value.incrementAndGet(), res.intValue()))
                .doOnNext(res -> latch.countDown())
                .subscribe();

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