@Override
    public Publisher<Exchange> toStream(String name, Object data) {
        return doRequest(
                name,
                ReactiveStreamsHelper.convertToExchange(context, data));
    }
@Test
    public void testToStream() throws Exception {
        context.addRoutes(new RouteBuilder() {
            public void configure() {
                from("reactive-streams:reactive").setBody().constant("123");
            }
        });

        context.start();

        Publisher<Exchange> publisher = crs.toStream("reactive", new DefaultExchange(context));
        Exchange res = Flowable.fromPublisher(publisher).blockingFirst();

        assertNotNull(res);

        String content = res.getIn().getBody(String.class);

        assertNotNull(content);
        assertEquals("123", content);
    }