@Test ( timeout = 5000 ) public void handlesJmsRuntimeException() { when( consumer.receive( 0 ) ).thenThrow( new JMSRuntimeException( "exception" ) ); source.open(); verify( delegate ).getJmsContext( consumerStep ); verify( delegate ).getDestination( consumerStep ); try { source.observable().firstElement().blockingGet( Collections.emptyList() ); fail( "Expected exception " ); } catch ( Exception e ) { assertTrue( e instanceof JMSRuntimeException ); } }
@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();  }