small refinements, still flaky

src/main/java/reactivestreams/DigitBlockChecker.java

@@ -12,12 +12,14 @@ public abstract class DigitBlockChecker extends SubmissionPublisher<Boolean> imp
 
 	int currentRow;
 	int currentColumn;
+	Flow.Subscription subscription;
 	Collection<String> digitsProcessed;
 
 	@Override
 	public void onSubscribe(Flow.Subscription subscription) {
 		currentRow = 1;
 		currentColumn = 0;
+		this.subscription = subscription;
 		digitsProcessed = new HashSet<>();
 		subscription.request(9*9);
 	}
@@ -32,12 +34,12 @@ public abstract class DigitBlockChecker extends SubmissionPublisher<Boolean> imp
 		if (currentDigitApplies(currentRow, currentColumn)) {
 			if (digitsProcessed.contains(digit)) {
 				submit(false);
-				close();
+				subscription.cancel();
 			}
 			digitsProcessed.add(digit);
 			if (digitsProcessed.size() == 9) {
 				submit(true);
-				close();
+				subscription.cancel();
 			}
 		}
 	}
@@ -47,7 +49,7 @@ public abstract class DigitBlockChecker extends SubmissionPublisher<Boolean> imp
 	@Override
 	public void onError(Throwable throwable) {
 		System.err.println(throwable.fillInStackTrace());
-		close();
+		closeExceptionally(throwable);
 	}
 
 	@Override

src/main/java/reactivestreams/SudokuVerifier.java

@@ -20,12 +20,12 @@ public class SudokuVerifier implements Flow.Subscriber<Boolean> {
 		digitPublisher = new SubmissionPublisher<>();
 		// Setup row and column checking subscribers
 		IntStream.rangeClosed(1, 9).boxed()
-			.forEach(i -> rowCheckerAndColumnChecker(digitPublisher, i));
+			.forEach(i -> rowCheckerAndColumnChecker(i));
 		// Setup square checking subscribers
 		IntStream.rangeClosed(1, 3).boxed()
 			.forEach(squareRow -> {
 				IntStream.rangeClosed(1, 3).boxed()
-					.forEach(squareColumn -> subscribeBlockChecker(digitPublisher, new SquareChecker(squareRow, squareColumn)));
+					.forEach(squareColumn -> squareChecker(squareRow, squareColumn));
 			});
 	}
 
@@ -34,26 +34,36 @@ public class SudokuVerifier implements Flow.Subscriber<Boolean> {
 		digitStream.forEach(digitPublisher::submit);
 		digitPublisher.close();
 		// Wait for checking subscribers to finish
-		while (finished.get() < digitPublisher.getSubscribers().size() && solved.get()) {
+		int subscriberCount = digitPublisher.getSubscribers().size();
-			try {
+		while (true) {
-				Thread.sleep(5);
+			if (!(finished.get() < subscriberCount && solved.get())) break;
-			} catch (InterruptedException ex) {
+			justAMoment();
-				System.out.println(ex.fillInStackTrace());
-			}
 		}
 		return solved.get();
 	}
 
-	private void rowCheckerAndColumnChecker(SubmissionPublisher<String> digitPublisher, Integer i) {
+	void rowCheckerAndColumnChecker(Integer i) {
-		subscribeBlockChecker(digitPublisher, new RowChecker(i));
+		subscribeBlockChecker(new RowChecker(i));
-		subscribeBlockChecker(digitPublisher, new ColumnChecker(i));
+		subscribeBlockChecker(new ColumnChecker(i));
 	}
 
-	private void subscribeBlockChecker(SubmissionPublisher<String> digitPublisher, DigitBlockChecker blockChecker) {
+	void squareChecker(Integer squareRow, Integer squareColumn) {
+		subscribeBlockChecker(new SquareChecker(squareRow, squareColumn));
+	}
+
+	void subscribeBlockChecker(DigitBlockChecker blockChecker) {
 		digitPublisher.subscribe(blockChecker);
 		blockChecker.subscribe(this);
 	}
 
+	void justAMoment() {
+		try {
+			Thread.sleep(5);
+		} catch (InterruptedException ex) {
+			System.out.println(ex.fillInStackTrace());
+		}
+	}
+
 	@Override
 	public void onSubscribe(Flow.Subscription subscription) {
 		subscription.request(1); // 1 Boolean result from the block checker
@@ -69,8 +79,8 @@ public class SudokuVerifier implements Flow.Subscriber<Boolean> {
 	@Override
 	public void onError(Throwable throwable) {
 		System.err.println(throwable.fillInStackTrace());
-		solved.compareAndSet(true, false);
 		finished.incrementAndGet();
+		System.exit(1);
 	}
 
 	@Override