public CloseableIterator<P> getMany(final Map<String, Object> query, final PipelineContext context, final boolean streaming) {
        if(!streaming) {
            final List<A> collector = new LinkedList<>();
            try(CloseableIterator<A> result = source.getMany(acquiredType, query, context)) {
                if(result == null) {
                    return null;
                }

                while(result.hasNext()) {
                    collector.add(result.next());
                }
            } catch(final Exception e) {
                LOGGER.error("Error closing results iterator!", e);
                throw new RuntimeException(e);
            }

            for(final SinkHandler<A, ?> sink : preTransform) {
                sink.putMany(collector, context);
            }

            final List<P> transformed = new LinkedList<>();
            for(final A item : collector) {
                transformed.add(transform.transform(item, context));
            }

            for(final SinkHandler<P, ?> sink : postTransform) {
                sink.putMany(transformed, context);
            }

            return CloseableIterators.from(transformed);
        } else {
            final CloseableIterator<A> result = source.getMany(acquiredType, query, context);
            if(result == null) {
                return null;
            }
            return new TransformingIterator(result, context);
        }
    }
public CloseableIterator < P > getMany ( final Map < String , Object > query , final PipelineContext context , final boolean streaming ) { if ( ! streaming ) { final List < A > collector = new LinkedList < > ( ) ; try ( CloseableIterator < A > result = source . getMany ( acquiredType , query , context ) ) { if ( result == null ) { return null ; } while ( result . hasNext ( ) ) { collector . add ( result . next ( ) ) ; } } catch ( final Exception e ) { LOGGER . error ( "Error closing results iterator!" , e ) ; throw new RuntimeException ( e ) ; } for ( final SinkHandler < A , ? > sink : preTransform ) { sink . putMany ( collector , context ) ; } final List < P > transformed = new LinkedList < > ( ) ; for ( final A item : collector ) { transformed . add ( transform . transform ( item , context ) ) ; } for ( final SinkHandler < P , ? > sink : postTransform ) { sink . putMany ( transformed , context ) ; } return CloseableIterators . from ( transformed ) ; } else { final CloseableIterator < A > result = source . getMany ( acquiredType , query , context ) ; if ( result == null ) { return null ; } return new TransformingIterator ( result , context ) ; } }
Gets multiple data elements from the underlying {@link com.merakianalytics.datapipelines.sources.DataSource}, provides them to the appropriate
{@link com.merakianalytics.datapipelines.SinkHandler}s, converts them and returns them

@param query
a query specifying the details of what data should fulfill this request
@param context
information about the context of the request such as what {@link com.merakianalytics.datapipelines.DataPipeline} called this method
@param streaming
whether to stream the results. If streaming is enabled, results from the {@link com.merakianalytics.datapipelines.sources.DataSource} will be
converted and provided to the {@link com.merakianalytics.datapipelines.sinks.DataSink}s one-by-one as they are requested from the
{@link com.merakianalytics.datapipelines.iterators.CloseableIterator} instead of converting and storing them all at once.
@return a {@link com.merakianalytics.datapipelines.iterators.CloseableIterator} of the request type if the query had a result, or null