sudoku-verifier-rxjava/src/main/java/reactivestreams/SudokuVerifier.java

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(i));
		// Setup square checking subscribers
		IntStream.rangeClosed(1, 3).boxed()
			.forEach(squareRow -> {
				IntStream.rangeClosed(1, 3).boxed()
					.forEach(squareColumn -> squareChecker(squareRow, squareColumn));
			});
	}

	public boolean isSolved() {
		int subscriberCount = digitPublisher.getSubscribers().size();
		// Make publisher emit each digit
		digitStream.forEach(digitPublisher::submit);
		digitPublisher.close();
		// Wait for checking subscribers to finish
		while ((finished.get() < subscriberCount) && solved.get()) {
			justAMoment();
		}
		return solved.get();
	}

	void rowCheckerAndColumnChecker(Integer i) {
		subscribeBlockChecker(new RowChecker(i));
		subscribeBlockChecker(new ColumnChecker(i));
	}

	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
	}

	@Override
	public void onNext(Boolean solved) {
		if (! solved) {
			this.solved.compareAndSet(true, false);
		}
		finished.incrementAndGet();
	}

	@Override
	public void onError(Throwable throwable) {
		System.err.println(throwable.fillInStackTrace());
		finished.incrementAndGet();
	}

	@Override
	public void onComplete() {
		// not called
	}
}