@Override
    public Subscriber<Exchange> subscriber(String uri) {
        try {
            String uuid = context.getUuidGenerator().generateUuid();
            RouteBuilder.addRoutes(context, rb -> rb.from("reactive-streams:" + uuid).to(uri));

            return streamSubscriber(uuid);
        } catch (Exception e) {
            throw new IllegalStateException("Unable to create source reactive stream towards direct URI: " + uri, e);
        }
    }
@Test
    public void testSubscriber() throws Exception {
        context.start();
        context.addRoutes(new RouteBuilder() {
            @Override
            public void configure() {
                from("direct:reactor").to("mock:result");
            }
        });

        Flowable.just(1, 2, 3).subscribe(crs.subscriber("direct:reactor", Integer.class));

        MockEndpoint mock = getMockEndpoint("mock:result");
        mock.expectedMessageCount(3);
        mock.assertIsSatisfied();

        int idx = 1;
        for (Exchange ex : mock.getExchanges()) {
            assertEquals(Integer.valueOf(idx++), ex.getIn().getBody(Integer.class));
        }
    }