rxjava solution - check only row 1

build.gradle

@@ -16,6 +16,7 @@ allprojects {
     }
     
     dependencies {
+        compile 'io.reactivex.rxjava3:rxjava:3.0.1'
         testImplementation 'org.junit.jupiter:junit-jupiter-api:5.6.0'
         testImplementation 'org.junit.jupiter:junit-jupiter-params:5.6.0'
         testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.6.0'

src/main/java/reactivestreams/Solution.java

@@ -2,12 +2,11 @@ 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.
+ * stream.Solution involving a Sukdoku Board, a SudokuVerifier, plus some ReactiveStreams processing.
 **/
 class Solution {
 	

src/main/java/rxjava/Solution.java

@@ -0,0 +1,23 @@
+package rxjava;
+
+import java.util.Scanner;
+import java.util.Spliterator;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+/**
+ * stream.Solution involving a Sukdoku Board, a SudokuVerifier, plus some RxJava 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/rxjava/SudokuVerifier.java

@@ -0,0 +1,55 @@
+package rxjava;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.Single;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+public class SudokuVerifier {
+
+	@NonNull Observable<String> digitObservable;
+
+	public SudokuVerifier(Stream<String> digitStream) {
+		// Observable on the digit stream
+		digitObservable = Observable.fromStream(digitStream);
+	}
+
+	public boolean isSolved() {
+		AtomicInteger row = new AtomicInteger(1);
+		AtomicInteger column = new AtomicInteger(0);
+		@NonNull Observable<DigitField> digitFieldObservable = digitObservable
+			.map(digit -> digitField(row, column, digit));
+		@NonNull Observable<DigitField> filter = digitFieldObservable
+			.filter(digitField -> digitField.row == 1);
+		return isSolved(filter).blockingGet();
+	}
+
+	@NonNull Single<Boolean> isSolved(@NonNull Observable<DigitField> filter) {
+		return filter
+			.map(digitField -> digitField.digit)
+			.distinct().count()
+			.map(count -> count == 9);
+	}
+
+	DigitField digitField(AtomicInteger row, AtomicInteger column, String digit) {
+		if (column.getAndIncrement() > 9) {
+			row.getAndIncrement();
+			column.set(1);
+		}
+		return new DigitField(row.get(), column.get(), 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;
+		}
+	}
+}

src/main/java/stream/Solution.java

@@ -7,7 +7,7 @@ import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 /**
- * stream.Solution involving a Sukdoku stream.Board, a stream.SudokuVerifier, plus some Stream processing.
+ * stream.Solution involving a Sukdoku Board, a SudokuVerifier, plus some Stream processing.
 **/
 class Solution {
 	

src/test/java/rxjava/SolutionTest.java

@@ -0,0 +1,92 @@
+package rxjava;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.converter.ArgumentConversionException;
+import org.junit.jupiter.params.converter.ConvertWith;
+import org.junit.jupiter.params.converter.SimpleArgumentConverter;
+import org.junit.jupiter.params.provider.CsvFileSource;
+
+import java.io.*;
+import java.util.stream.Collectors;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class SolutionTest {
+	
+	static class SudokuConverter extends SimpleArgumentConverter {
+		@Override
+		protected Object convert( Object source, Class<?> targetType ) throws ArgumentConversionException {
+			assertEquals(String.class, targetType, "Can only convert to String");
+			BufferedReader in = new BufferedReader( new InputStreamReader( SolutionTest.class.getResourceAsStream( String.valueOf( source ))));
+			return in.lines().collect( Collectors.joining( System.lineSeparator() ));
+		}
+	}
+	
+	static long totalDuration;
+	
+	@BeforeAll
+	static void beforeAll() {
+		totalDuration = 0;
+	}
+	
+	@AfterAll
+	static void afterAll() {
+		System.out.println( String.format( "All solutions took %01d.%03d secs",  totalDuration / 1000,  totalDuration % 1000 ));
+	}
+	
+	@ParameterizedTest
+	@CsvFileSource( resources = "/testdata.csv" )
+	void main( @ConvertWith( SudokuConverter.class ) final String input, final String expected ) throws IOException {
+		// keep original streams
+		InputStream oldIn = System.in;
+		PrintStream oldOut = System.out;
+		PrintStream oldErr = System.err;
+		ByteArrayOutputStream bos =  new ByteArrayOutputStream();
+		
+		try {
+			redirectStreams(
+					new ByteArrayInputStream( input.getBytes( UTF_8 )),
+					new PrintStream( bos, true, UTF_8 ),
+					new PrintStream( new ByteArrayOutputStream(), true, UTF_8 )
+			);
+			
+			// start time tracking
+			long start = System.currentTimeMillis();
+			
+			Solution.main( new String[0] );
+			
+			// stop time tracking
+			long duration = System.currentTimeMillis() - start;
+			totalDuration += duration;
+			
+			// restore streams
+			redirectStreams( oldIn, oldOut, oldErr );
+			
+			System.out.println( String.format( "Solution took %01d.%03d secs",  duration / 1000,  duration % 1000 ));
+			
+			try (BufferedReader chk = new BufferedReader( new InputStreamReader( new ByteArrayInputStream( bos.toByteArray() ), UTF_8 ))) {
+				String[] expectedLines = expected.split( "\\s*[|]\\s*" );
+				
+				int lineCount = 0;
+				String line;
+				for ( ; (line = chk.readLine()) != null; lineCount++ ) {
+					assertEquals( expectedLines[ lineCount ], line );
+				}
+				assertEquals( expectedLines.length, lineCount );
+			}
+		}
+		finally {
+			// restore streams
+			redirectStreams( oldIn, oldOut, oldErr );
+		}
+	}
+	
+	static void redirectStreams( final InputStream input, final PrintStream output, final PrintStream error ) {
+		System.setIn( input );
+		System.setOut( output );
+		System.setErr( error );
+	}
+}