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

package rxjava;

import io.reactivex.rxjava3.core.Observable;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

public class SudokuVerifier {


	List<DigitField> digitFields;

	public SudokuVerifier(Stream<String> digitStream) {
		// Observable on the digit stream
		AtomicInteger row = new AtomicInteger(1);
		AtomicInteger column = new AtomicInteger(0);
		digitFields = Observable.fromStream(digitStream)
			.map(digit -> digitField(row, column, digit))
			.collect(Collectors.toList())
			.blockingGet();
	}

	public boolean isSolved() {
		// Filter for 9 rows, 9 columns, 9 squares
		return digitsInBlockFilters()
			.map(blockFilter -> blockFilter.distinct().count().blockingGet() == 9)
			.all(solved -> solved).blockingGet();
	}

	DigitField digitField(AtomicInteger row, AtomicInteger column, String digit) {
		if (column.incrementAndGet() > 9) {
			row.getAndIncrement();
			column.set(1);
		}
		return new DigitField(row.get(), column.get(), digit);
	}

	Observable<Observable<String>> digitsInBlockFilters() {
		Stream<Observable<String>> stream = Stream.of(
			IntStream.rangeClosed(1, 9).boxed().map(row -> digitsInRowFilter(row)),
			IntStream.rangeClosed(1, 9).boxed().map(column -> digitsInColumnFilter(column)),
			digitsInSquareFilters()
		).flatMap(s -> s);
		return Observable.fromStream(stream);
	}

	Observable<String> digitsInRowFilter(Integer row) {
		return Observable.fromStream(
			digitFields.stream()
				.filter(digitField -> digitField.row == row)
				.map(digitField -> digitField.digit));
	}

	Observable<String> digitsInColumnFilter(Integer column) {
		return Observable.fromStream(
			digitFields.stream()
				.filter(digitField -> digitField.column == column)
				.map(digitField -> digitField.digit));
	}

	Stream<Observable<String>> digitsInSquareFilters() {
		return IntStream.rangeClosed(1, 3).boxed().flatMap(
			squareRow -> IntStream.rangeClosed(1, 3).boxed().map(
				squareColumn -> digitsInSquareFilter(squareRow, squareColumn)));
	}

	Observable<String> digitsInSquareFilter(Integer squareRow, Integer squareColumn) {
		return Observable.fromStream(
			digitFields.stream()
				.filter(digitField -> digitField.isInSquare(squareRow, squareColumn))
				.map(digitField -> digitField.digit));
	}

	static class DigitField {
		int row;
		int column;
		String digit;

		public DigitField(int row, int column, String digit) {
			this.row = row;
			this.column = column;
			this.digit = digit;
		}

		public boolean isInSquare(Integer squareRow, Integer squareColumn) {
			return row >= squareRow * 3 - 2
				&& row <= squareRow * 3
				&& column >= squareColumn * 3 - 2
				&& column <= squareColumn * 3;
		}
	}
}