Allowed affected row count to be modelled (#747)

example/app-model/src/main/java/com/mastercard/test/flow/example/app/model/Deferred.java

@@ -109,7 +109,7 @@ public Deferred() {
 												.call( e -> e
 														.to( Actors.DB )
 														.request( dbInsert( deferredId, text ) )
-														.response( new Result( "1" ) ) )
+														.response( new Result().set( Result.ROW_COUNT, 1 ) ) )
 												.response( httpRes( null ) ) )
 										.response( textRes( deferredId ) ) )
 								.response( coreRes(
@@ -195,7 +195,7 @@ public Deferred() {
 										.to( Actors.DB )
 										.tags( add( "put" ) )
 										.request( dbInsert( deferredId, countText ) )
-										.response( new Result() ) )
+										.response( new Result().set( Result.ROW_COUNT, 1 ) ) )
 								.response( httpRes( null ) ) )
 						.response( httpRes( null ) ) )
 				.residue( new DBItems()
@@ -233,7 +233,7 @@ public Deferred() {
 						.call( b -> b.to( Actors.DB )
 								.tags( add( "delete" ) )
 								.request( dbDelete( deferredId ) )
-								.response( new Result() ) )
+								.response( new Result().set( Result.ROW_COUNT, 1 ) ) )
 						.response( textRes( countText ) ) )
 				.update( UI,
 						rq( METHOD, "POST" ),

example/app-model/src/test/java/com/mastercard/test/flow/example/app/model/ExampleSystemTest.java

@@ -56,7 +56,7 @@ void messageHashes() {
 
 		mh.expect( ExampleSystem.MODEL,
 				"ALL MESSAGES",
-				"720AC1156AFAE947D7B32D94483B5B16 0122 21.3 KiB",
+				"091D205B75B0D8044227E52CCC7B70D6 0122 21.4 KiB",
 				"REQUESTS --> USER",
 				"00000000000000000000000000000000 0000 0 B",
 				"RESPONSES <-- USER",
@@ -88,7 +88,7 @@ void messageHashes() {
 				"REQUESTS --> DB",
 				"DFB253E31C3C812A60E039379CCFA64D 0008 2.5 KiB",
 				"RESPONSES <-- DB",
-				"CF399FF557548CC74F579F5193CE2264 0008 1.1 KiB",
+				"B42E7EBB481EBD84DAC357E9178E09A4 0008 1.3 KiB",
 				"REQUESTS --> HISTOGRAM",
 				"BF8DDDB8D319B876FADD333BCB149E4F 0009 1.1 KiB",
 				"RESPONSES <-- HISTOGRAM",

example/app-store/src/main/java/com/mastercard/test/flow/example/app/store/StoreImp.java

@@ -42,9 +42,10 @@ public void store( String key, String data ) {
 		catch( NoSuchAlgorithmException e ) {
 			throw new IllegalStateException( e );
 		}
-		DB.update( db, "INSERT INTO item ( id, data, hash ) VALUES ( ?, ?, ? ) "
+		int rows = DB.update( db, "INSERT INTO item ( id, data, hash ) VALUES ( ?, ?, ? ) "
 				+ "ON DUPLICATE KEY UPDATE data = ?, hash = ?",
 				key, data, hash, data, hash );
+		LOG.info( "Updated {} rows", rows );
 	}
 
 	@Override
@@ -74,7 +75,8 @@ public String retrieve( String key ) {
 	public String delete( String key ) {
 		LOG.info( "Deleting {}", key );
 		String value = retrieve( key );
-		DB.update( db, "DELETE FROM item WHERE id = ?", key );
+		int rows = DB.update( db, "DELETE FROM item WHERE id = ?", key );
+		LOG.info( "Deleted {} rows", rows );
 		return value;
 	}
 

example/app-store/src/test/java/com/mastercard/test/flow/example/app/store/QueryTest.java

@@ -159,9 +159,10 @@ private static Runnable mockDBInteractions( Stream<Interaction> dbntr ) {
 						PreparedStatement ps = mock( PreparedStatement.class );
 						when( c.prepareStatement( anyString() ) ).thenReturn( ps );
 
-						if( sql.startsWith( "SELECT" ) ) {
+						Result r = (Result) ntr.response();
+						if( r.affectedRowCount() == null ) {
+							// for SELECT queries
 
-							Result r = (Result) ntr.response();
 							Deque<Map<String, Object>> rows = new ArrayDeque<>( r.get() );
 
 							AtomicReference<Map<String, Object>> current = new AtomicReference<>();
@@ -177,6 +178,11 @@ private static Runnable mockDBInteractions( Stream<Interaction> dbntr ) {
 							when( rs.getObject( anyString() ) )
 									.then( i -> current.get().get( i.getArgument( 0 ) ) );
 						}
+						else {
+							// for UPDATE queries
+							when( ps.executeUpdate() )
+									.thenReturn( r.affectedRowCount() );
+						}
 
 						verifies.add( () -> {
 							try {

message/message-sql/src/main/java/com/mastercard/test/flow/msg/sql/Result.java

@@ -10,6 +10,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Optional;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
@@ -38,6 +39,10 @@ public class Result extends AbstractMessage<Result> {
 	 * Field address to use to update the column list
 	 */
 	public static final String COLUMNS = "columns";
+	/**
+	 * Field address to use to update the affected row count
+	 */
+	public static final String ROW_COUNT = "row_count";
 	private static final Pattern ROW_COL_PTRN = Pattern.compile(
 			"(\\d+):(\\d+)" );
 
@@ -53,7 +58,8 @@ private Result( Supplier<ResultSetStructure> basis ) {
 	public Result( String... columns ) {
 		this( () -> new ResultSetStructure(
 				new ArrayList<>( Arrays.asList( columns ) ),
-				new ArrayList<>() ) );
+				new ArrayList<>(),
+				null ) );
 	}
 
 	private Result( byte[] bytes ) {
@@ -88,6 +94,13 @@ private ResultSetStructure data() {
 				Stream.of( String.valueOf( update.value() ).split( "," ) )
 						.forEach( data.columns::add );
 			}
+			else if( ROW_COUNT.equals( update.field() ) ) {
+				data.affectedRowCount = Optional.ofNullable( update.value() )
+						.map( String::valueOf )
+						.filter( s -> s.matches( "\\d+" ) )
+						.map( Integer::valueOf )
+						.orElse( null );
+			}
 			else if( update.field().matches( "\\d+" ) ) {
 				// whole-row operations
 				int row = Integer.parseInt( update.field() );
@@ -181,6 +194,10 @@ protected Object access( String field ) {
 			return new ArrayList<>( data.columns );
 		}
 
+		if( ROW_COUNT.equals( field ) ) {
+			return data.affectedRowCount;
+		}
+
 		if( field.matches( "\\d+" ) ) {
 			int row = Integer.parseInt( field );
 			if( data.maps.size() > row ) {
@@ -218,10 +235,21 @@ public List<Map<String, Object>> get() {
 				.collect( toList() );
 	}
 
+	/**
+	 * @return The number of affected rows, or <code>null</code> if this message
+	 *         does not model the result of a data manipulation query
+	 */
+	public Integer affectedRowCount() {
+		return data().affectedRowCount;
+	}
+
 	@Override
 	protected String asHuman() {
 		StringBuilder sb = new StringBuilder();
 		ResultSetStructure data = data();
+		if( data.affectedRowCount != null ) {
+			return String.format( "%s affected rows", data.affectedRowCount );
+		}
 		String colfmt = "%" + data.columns.stream()
 				.mapToInt( String::length )
 				.max().orElse( 0 ) + "s";
@@ -271,11 +299,16 @@ private static class ResultSetStructure {
 		@JsonProperty("rows")
 		private final List<List<TypedKVP<Integer>>> rows;
 
+		@JsonProperty("affected_row_count")
+		Integer affectedRowCount;
+
 		public ResultSetStructure(
 				@JsonProperty("columns") List<String> columns,
-				@JsonProperty("rows") List<List<TypedKVP<Integer>>> rows ) {
+				@JsonProperty("rows") List<List<TypedKVP<Integer>>> rows,
+				@JsonProperty("affected_row_count") Integer affectedRowCount ) {
 			this.columns = columns;
 			this.rows = rows;
+			this.affectedRowCount = affectedRowCount;
 
 			maps = new ArrayList<>();
 			rows.forEach( row -> {

message/message-sql/src/test/java/com/mastercard/test/flow/msg/sql/ResultTest.java

@@ -66,6 +66,12 @@ void asHuman() {
 				+ "    def : h\n"
 				+ " longer : i",
 				res.asHuman() );
+
+		res.set( Result.ROW_COUNT, 8 );
+
+		assertEquals( "8 affected rows",
+				res.asHuman(),
+				"A row count will override any result rows" );
 	}
 
 	/**
@@ -114,6 +120,25 @@ void delete() {
 				+ "    def : null\n"
 				+ " longer : null",
 				res.asHuman() );
+
+		res.set( Result.ROW_COUNT, "8" );
+
+		assertEquals( "8 affected rows",
+				res.asHuman(),
+				"A row count will override any result rows" );
+
+		res.set( Result.ROW_COUNT, null );
+
+		assertEquals( ""
+				+ " --- Row 0 ---\n"
+				+ "    abc : abc value\n"
+				+ "    def : null\n"
+				+ " longer : null\n"
+				+ " --- Row 1 ---\n"
+				+ "    abc : null\n"
+				+ "    def : null\n"
+				+ " longer : null",
+				res.asHuman() );
 	}
 
 	/**
@@ -168,6 +193,19 @@ void set() {
 				+ " ijk : null",
 				res.asHuman() );
 
+		res.set( Result.ROW_COUNT, "not valid" );
+
+		assertEquals( ""
+				+ " --- Row 0 ---\n"
+				+ " ghi : a\n"
+				+ " ijk : b\n"
+				+ " --- Row 1 ---\n"
+				+ " ghi : c\n"
+				+ " ijk : null",
+				res.asHuman(),
+				"The invalid row count has been ignored" );
+		assertEquals( null, res.affectedRowCount() );
+
 		Object o = new Object();
 		IllegalArgumentException iae = assertThrows( IllegalArgumentException.class,
 				() -> res.set( "0:0", o ) );
@@ -215,6 +253,12 @@ void get() {
 		assertEquals( null, res.get( "3" ) );
 
 		assertEquals( "[{abc=a}, {def=b}, {longer=c}]", res.get().toString() );
+
+		assertEquals( null, res.get( Result.ROW_COUNT ) );
+
+		res.set( Result.ROW_COUNT, "8" );
+
+		assertEquals( 8, res.get( Result.ROW_COUNT ) );
 	}
 
 	/**