reactivestreams version, still flaky

src/main/java/reactivestreams/ColumnChecker.java

@@ -0,0 +1,16 @@
+package reactivestreams;
+
+public class ColumnChecker extends DigitBlockChecker {
+
+	int column;
+
+	public ColumnChecker(int column) {
+		this.column = column;
+	}
+
+	@Override
+	protected boolean currentDigitApplies(int currentRow, int currentColumn) {
+		return currentColumn == column;
+	}
+
+}

src/main/java/reactivestreams/DigitBlockChecker.java

@@ -0,0 +1,58 @@
+package reactivestreams;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.concurrent.Flow;
+import java.util.concurrent.SubmissionPublisher;
+
+/**
+ * Processor consuming a flow of digit Strings and emitting a single Boolean which is true if applying 9 digits are unique.
+ */
+public abstract class DigitBlockChecker extends SubmissionPublisher<Boolean> implements Flow.Processor<String, Boolean> {
+
+	int currentRow;
+	int currentColumn;
+	Collection<String> digitsProcessed;
+
+	@Override
+	public void onSubscribe(Flow.Subscription subscription) {
+		currentRow = 1;
+		currentColumn = 0;
+		digitsProcessed = new HashSet<>();
+		subscription.request(9*9);
+	}
+
+	@Override
+	public void onNext(String digit) {
+		currentColumn++;
+		if (currentColumn > 9) {
+			currentRow++;
+			currentColumn = 1;
+		}
+		if (currentDigitApplies(currentRow, currentColumn)) {
+			if (digitsProcessed.contains(digit)) {
+				submit(false);
+				close();
+			}
+			digitsProcessed.add(digit);
+			if (digitsProcessed.size() == 9) {
+				submit(true);
+				close();
+			}
+		}
+	}
+
+	protected abstract boolean currentDigitApplies(int currentRow, int currentColumn);
+
+	@Override
+	public void onError(Throwable throwable) {
+		System.err.println(throwable.fillInStackTrace());
+		close();
+	}
+
+	@Override
+	public void onComplete() {
+		close();
+	}
+
+}

src/main/java/reactivestreams/RowChecker.java

@@ -0,0 +1,16 @@
+package reactivestreams;
+
+public class RowChecker extends DigitBlockChecker {
+
+	int row;
+
+	public RowChecker(int row) {
+		this.row = row;
+	}
+
+	@Override
+	protected boolean currentDigitApplies(int currentRow, int currentColumn) {
+		return currentRow == row;
+	}
+
+}

src/main/java/reactivestreams/Solution.java

@@ -0,0 +1,24 @@
+package reactivestreams;
+
+import java.util.Scanner;
+import java.util.Spliterator;
+import java.util.concurrent.SubmissionPublisher;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+/**
+ * stream.Solution involving a Sukdoku stream.Board, a stream.SudokuVerifier, plus some ReactiveStreams processing.
+**/
+class Solution {
+	
+	public static void main( String[] args ) {
+		// Setup stream on System.in
+		Scanner in = new Scanner( System.in );
+		Spliterator spliterator = ((Iterable) () -> in).spliterator();
+		Stream<String> digitStream = StreamSupport.stream(spliterator, false);
+		// Setup SudokuVerifier
+		SudokuVerifier verifier = new SudokuVerifier(digitStream);
+		// Verify and print out result
+		System.out.println(verifier.isSolved());
+	}
+}

src/main/java/reactivestreams/SquareChecker.java

@@ -0,0 +1,30 @@
+package reactivestreams;
+
+public class SquareChecker extends DigitBlockChecker {
+
+	int squareRow;
+	int squareColumn;
+
+	public SquareChecker(int squareRow, int squareColumn) {
+		this.squareRow = squareRow;
+		this.squareColumn = squareColumn;
+	}
+
+	@Override
+	protected boolean currentDigitApplies(int currentRow, int currentColumn) {
+		return currentRowApplies(currentRow) && currentColumnApplies(currentColumn);
+	}
+
+	boolean currentRowApplies(int currentRow) {
+		int fromRow = squareRow * 3 - 2;
+		int toRow = fromRow + 2;
+		return currentRow >= fromRow && currentRow <= toRow;
+	}
+
+	boolean currentColumnApplies(int currentColumn) {
+		int fromColumn = squareColumn * 3 - 2;
+		int toColumn = fromColumn + 2;
+		return currentColumn >= fromColumn && currentColumn <= toColumn;
+	}
+
+}

src/main/java/reactivestreams/SudokuVerifier.java

@@ -0,0 +1,80 @@
+package reactivestreams;
+
+import java.util.concurrent.Flow;
+import java.util.concurrent.SubmissionPublisher;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+public class SudokuVerifier implements Flow.Subscriber<Boolean> {
+
+	Stream<String> digitStream;
+	SubmissionPublisher<String> digitPublisher;
+	AtomicInteger finished = new AtomicInteger(0);
+	AtomicBoolean solved = new AtomicBoolean(true);
+
+	public SudokuVerifier(Stream<String> digitStream) {
+		this.digitStream = digitStream;
+		// Setup publisher to emit the digitStream
+		digitPublisher = new SubmissionPublisher<>();
+		// Setup row and column checking subscribers
+		IntStream.rangeClosed(1, 9).boxed()
+			.forEach(i -> rowCheckerAndColumnChecker(digitPublisher, 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)));
+			});
+	}
+
+	public boolean isSolved() {
+		// Make publisher emit each digit
+		digitStream.forEach(digitPublisher::submit);
+		digitPublisher.close();
+		// Wait for checking subscribers to finish
+		while (finished.get() < digitPublisher.getSubscribers().size() && solved.get()) {
+			try {
+				Thread.sleep(5);
+			} catch (InterruptedException ex) {
+				System.out.println(ex.fillInStackTrace());
+			}
+		}
+		return solved.get();
+	}
+
+	private void rowCheckerAndColumnChecker(SubmissionPublisher<String> digitPublisher, Integer i) {
+		subscribeBlockChecker(digitPublisher, new RowChecker(i));
+		subscribeBlockChecker(digitPublisher, new ColumnChecker(i));
+	}
+
+	private void subscribeBlockChecker(SubmissionPublisher<String> digitPublisher, DigitBlockChecker blockChecker) {
+		digitPublisher.subscribe(blockChecker);
+		blockChecker.subscribe(this);
+	}
+
+	@Override
+	public void onSubscribe(Flow.Subscription subscription) {
+		subscription.request(1); // 1 Boolean result from the block checker
+	}
+
+	@Override
+	public void onNext(Boolean solved) {
+		if (! solved) {
+			this.solved.compareAndSet(true, false);
+		}
+	}
+
+	@Override
+	public void onError(Throwable throwable) {
+		System.err.println(throwable.fillInStackTrace());
+		solved.compareAndSet(true, false);
+		finished.incrementAndGet();
+	}
+
+	@Override
+	public void onComplete() {
+		finished.incrementAndGet();
+	}
+}