@Test ( timeout = 5000 ) public void testReceiveMessage() { source.open(); verify( delegate ).getJmsContext( consumerStep ); verify( delegate ).getDestination( consumerStep ); List<Object> sentMessage = source.observable().firstElement().blockingGet( Collections.emptyList() ); assertThat( sentMessage.size(), equalTo( 2 ) ); assertThat( sentMessage.get( 0 ), equalTo( "message" ) ); assertThat( sentMessage.get( 1 ), equalTo( "dest" ) ); }
@Override public void open() { consumer = jmsDelegate.getJmsContext( streamStep ) .createConsumer( jmsDelegate.getDestination( streamStep ) ); Observable .create( receiveLoop() ) .subscribeOn( io() ) .observeOn( io() ) .doOnNext( message -> acceptRows( singletonList( of( message, jmsDelegate.destinationName ) ) ) ) .doOnComplete( this::close ) .doOnError( this::error ) .publish() .connect(); }
JmsStreamSource extends BlockingQueueStreamSource<List<Object>> { @Override public void open() { consumer = jmsDelegate.getJmsContext( streamStep ) .createConsumer( jmsDelegate.getDestination( streamStep ) ); Observable .create( receiveLoop() ) .subscribeOn( io() ) .observeOn( io() ) .doOnNext( message -> acceptRows( singletonList( of( message, jmsDelegate.destinationName ) ) ) ) .doOnComplete( this::close ) .doOnError( this::error ) .publish() .connect(); } }
JmsStreamSource extends BlockingQueueStreamSource<List<Object>> { @Override public void open() { consumer = jmsDelegate.getJmsContext( streamStep ) .createConsumer( jmsDelegate.getDestination( streamStep ) ); Observable .create( receiveLoop() ) .subscribeOn( io() ) .observeOn( io() ) .doOnNext( message -> acceptRows( singletonList( of( message, jmsDelegate.destinationName ) ) ) ) .doOnComplete( this::close ) .doOnError( this::error ) .publish() .connect(); } JmsStreamSource( BaseStreamStep streamStep, JmsDelegate jmsDelegate, int receiverTimeout ); }
JmsStreamSource extends BlockingQueueStreamSource<List<Object>> { @Override public void open() { consumer = jmsDelegate.getJmsContext( streamStep ) .createConsumer( jmsDelegate.getDestination( streamStep ) ); Observable .create( receiveLoop() ) .subscribeOn( io() ) .observeOn( io() ) .doOnNext( message -> acceptRows( singletonList( of( message, jmsDelegate.destinationName ) ) ) ) .doOnComplete( this::close ) .doOnError( this::error ) .publish() .connect(); } JmsStreamSource( BaseStreamStep streamStep, JmsDelegate jmsDelegate, int receiverTimeout ); @Override void open(); @Override void close(); }
JmsStreamSource extends BlockingQueueStreamSource<List<Object>> { @Override public void open() { consumer = jmsDelegate.getJmsContext( streamStep ) .createConsumer( jmsDelegate.getDestination( streamStep ) ); Observable .create( receiveLoop() ) .subscribeOn( io() ) .observeOn( io() ) .doOnNext( message -> acceptRows( singletonList( of( message, jmsDelegate.destinationName ) ) ) ) .doOnComplete( this::close ) .doOnError( this::error ) .publish() .connect(); } JmsStreamSource( BaseStreamStep streamStep, JmsDelegate jmsDelegate, int receiverTimeout ); @Override void open(); @Override void close();  }