ParallelBatchFluxBuilder, adding builder visitor.

Author: marco-brandizi

src/main/java/uk/ac/ebi/utils/opt/runcontrol/ReactorUtils.java

@@ -128,40 +128,108 @@ public ParallelBatchFluxBuilder<T, B> withBatchSupplier ( Supplier<? extends Col
 			return this;
 		}
 
-		public ParallelFlux<B> build ()
+		
+		public int getParallelism ()
+		{
+			return parallelism;
+		}
+
+		public void setParallelism ( int parallelism )
+		{
+			this.parallelism = parallelism;
+		}
+
+		public int getParallelismPreFetch ()
+		{
+			return parallelismPreFetch;
+		}
+
+		public Scheduler getScheduler ()
+		{
+			return scheduler;
+		}
+
+		public int getBatchSize ()
+		{
+			return batchSize;
+		}
+
+		public Supplier<B> getBatchSupplier ()
+		{
+			return batchSupplier;
+		}
+		
+		/**
+		 * @param visitor if non-null, I'll call it with myself before creating the result. This can be
+		 * used to inspect a builder during the build process, to set info from defaults (eg, from
+		 * {@link #getParallelism()}).
+		 *
+		 */
+		public ParallelFlux<B> build ( Consumer<ParallelBatchFluxBuilder<? super T, ? super B>> visitor )
 		{
 			@SuppressWarnings ( "unchecked" )
 			Flux<B> result = this.batchSupplier == null 
 				? (Flux<B>) flux.buffer ( batchSize ) : flux.buffer ( batchSize, batchSupplier );
+
+			if ( visitor != null ) visitor.accept ( this );
 
 			return result
 			.parallel ( parallelism, parallelismPreFetch )
 			.runOn ( scheduler );
-		}		
+		}
+		
+		public ParallelFlux<B> build ()
+		{
+			return build ( null );
+		}
+		
 	} // class ParallelBatchFluxBuilder
 
 
 	/**
 	 * Just uses {@link ParallelBatchFluxBuilder} with its defaults. 
 	 */
-	public static <T> ParallelFlux<List<T>> parallelBatchFlux ( Flux<? extends T> flux ) {
-		return new ParallelBatchFluxBuilder<T, List<T>> ( flux ).build ();
+	public static <T> ParallelFlux<List<T>> parallelBatchFlux (
+		Flux<? extends T> flux, Consumer<ParallelBatchFluxBuilder<? super T, ? super List<T>>> visitor
+	)
+	{
+		return new ParallelBatchFluxBuilder<T, List<T>> ( flux )
+		.build ( visitor );
 	}
 
+	public static <T> ParallelFlux<List<T>> parallelBatchFlux ( Flux<? extends T> flux ) {
+		return parallelBatchFlux ( flux, null );
+	}
+	
+	
 	/**
 	 * Just uses {@link ParallelBatchFluxBuilder} with its defaults. 
 	 */
+	public static <T> ParallelFlux<List<T>> parallelBatchFlux (
+		Stream<? extends T> stream, Consumer<ParallelBatchFluxBuilder<? super T, ? super List<T>>> visitor
+	) 
+	{
+		return new ParallelBatchFluxBuilder<T, List<T>> ( stream ).build ( visitor );
+	}
+
 	public static <T> ParallelFlux<List<T>> parallelBatchFlux ( Stream<? extends T> stream  ) {
-		return new ParallelBatchFluxBuilder<T, List<T>> ( stream ).build ();
+		return parallelBatchFlux ( stream, null );
 	}
 
+	
 	/**
 	 * Just uses {@link ParallelBatchFluxBuilder} with its defaults. 
 	 */
-	public static <T> ParallelFlux<List<T>> parallelBatchFlux ( Collection<? extends T> collection  ) {
-		return new ParallelBatchFluxBuilder<T, List<T>> ( collection ).build ();
+	public static <T> ParallelFlux<List<T>> parallelBatchFlux (
+		Collection<? extends T> collection, Consumer<ParallelBatchFluxBuilder<? super T, ? super List<T>>> visitor
+	) 
+	{
+		return new ParallelBatchFluxBuilder<T, List<T>> ( collection ).build ( visitor );
 	}
 
+	public static <T> ParallelFlux<List<T>> parallelBatchFlux ( Collection<? extends T> collection  ) {
+		return parallelBatchFlux ( collection );
+	}
 
 	/**
 	 * Uses {@link ParallelBatchFluxBuilder} to process a source of batches.
@@ -191,19 +259,15 @@ public static <T> void batchProcessing (
 	/**
 	 * Variant of {@link #batchProcessing(Flux, Consumer)}
 	 */
-	public static <T> void batchProcessing (
-		Stream<T> stream, Consumer<List<T>> task		
-	)
+	public static <T> void batchProcessing ( Stream<T> stream, Consumer<List<T>> task	)
 	{
 		batchProcessing ( parallelBatchFlux ( stream ), task );
 	}
 
 	/**
 	 * Variant of {@link #batchProcessing(Flux, Consumer)}
 	 */
-	public static <T> void batchProcessing (
-		Collection<T> collection, Consumer<List<T>> task		
-	)
+	public static <T> void batchProcessing ( Collection<T> collection, Consumer<List<T>> task	)
 	{
 		batchProcessing ( parallelBatchFlux ( collection ), task );
 	}

src/test/java/uk/ac/ebi/utils/opt/runcontrol/ReactorUtilsTest.java

@@ -1,6 +1,7 @@
 package uk.ac.ebi.utils.opt.runcontrol;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 import java.util.Comparator;
 import java.util.HashSet;
@@ -10,6 +11,8 @@
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableInt;
 import org.junit.Test;
 
 import reactor.core.publisher.ParallelFlux;
@@ -66,4 +69,30 @@ public void testBatchProcessing ()
 		// Usual Gauss formula for Sum (1..n)
 		assertEquals ( "Result isn't as expected!", max * (max - 1) / 2, sum.get () );
 	}
+	
+	
+	@Test
+	public void testBatchProcessingWithVisitor ()
+	{
+		int max = 1000;
+		
+		Stream<Integer> strm = IntStream.range ( 0, max )
+		.mapToObj ( Integer::valueOf );
+		
+		
+		AtomicInteger sum = new AtomicInteger ();
+		
+		MutableInt parallelism = new MutableInt ( 0 );
+		
+		ReactorUtils.parallelBatchFlux ( 
+			strm,
+			builder -> parallelism.setValue ( builder.getParallelism () )
+		)
+		.doOnNext ( b -> sum.addAndGet ( b.stream ().mapToInt ( Integer::intValue ).sum () ) )
+		.sequential ()
+		.blockLast ();
+		
+		assertEquals ( "Result isn't as expected!", max * (max - 1) / 2, sum.get () );
+		assertTrue ( "parallelism wasn't retrieved from the builder!", parallelism.getValue () > 0 );
+	}	
 }