### Question:
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();  }### Answer:
@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" ) ); }
@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 ); } }