class JmsStreamSource extends BlockingQueueStreamSource<List<Object>> {

  @Override public void open() {
    consumer = jmsDelegate.getJmsContext( streamStep )
      .createConsumer( jmsDelegate.getDestination( streamStep ) );
    Observable
      .create( receiveLoop() )  // jms loop
      .subscribeOn( io() )   // subscribe and observe on new io threads
      .observeOn( io() )
      .doOnNext( message -> acceptRows( singletonList( of( message, jmsDelegate.destinationName ) ) ) )
      .doOnComplete( this::close )
      .doOnError( this::error )
      .publish() // publish/connect will "start" the receive loop
      .connect();
  }

    JmsStreamSource( BaseStreamStep streamStep, JmsDelegate jmsDelegate, int receiverTimeout );

}