r2dbc/src/test/java/rxtx/special/attention/TransactionalTests.java
// Note: Switch to R2dbcPostgresConnectionExtension, Enable pooling, run in tx
Switch to R2dbcPostgresConnectionExtension, Enable pooling, run in tx
Switch to R2dbcPostgresConnectionExtension Enable pooling run in tx
@Test
	void nPlusOne(ConnectionFactory connectionFactory) {
		ConnectionPool pool = new ConnectionPool(ConnectionPoolConfiguration.builder(connectionFactory).build());
		R2dbcTransactionManager transactionManager = new R2dbcTransactionManager(connectionFactory);
		DatabaseClient client = DatabaseClient.create(connectionFactory);
		TransactionalOperator rxtx = TransactionalOperator.create(transactionManager);
		starbucksRecords().concatMap(it -> {
			String name = it[0];
			String storeNumber = it[2];
			String address = it[5];
			String city = it[9];
			return client.insert().into("starbucks") 
					.value("store_number", storeNumber) 
					.value("name", name) 
					.value("address", address) 
					.value("city", city) 
					.then();
		}).then().as(StepVerifier::create).verifyComplete();
		client.select().from(Starbucks.class).fetch().all().flatMap(it -> {
			return client.select().from(Starbucks.class)
					.matching(where("city").is(it.getCity()).and("store_number").not(it.getStoreNumber())).fetch().all()
					.collectList().map(shopsInTheSameCity -> {
						it.setShopsInTheSameCity(shopsInTheSameCity);
						return it;
					});
		}).filter(it -> !it.shopsInTheSameCity.isEmpty()).doOnNext(System.out::println).as(StepVerifier::create)
				.expectNextCount(924).verifyComplete();
	}
@ Test void nPlusOne ( ConnectionFactory connectionFactory ) { ConnectionPool pool = new ConnectionPool ( ConnectionPoolConfiguration . builder ( connectionFactory ) . build ( ) ) ; R2dbcTransactionManager transactionManager = new R2dbcTransactionManager ( connectionFactory ) ; DatabaseClient client = DatabaseClient . create ( connectionFactory ) ; TransactionalOperator rxtx = TransactionalOperator . create ( transactionManager ) ; starbucksRecords ( ) . concatMap ( it -> { String name = it [ 0 ] ; String storeNumber = it [ 2 ] ; String address = it [ 5 ] ; String city = it [ 9 ] ; return client . insert ( ) . into ( "starbucks" ) . value ( "store_number" , storeNumber ) . value ( "name" , name ) . value ( "address" , address ) . value ( "city" , city ) . then ( ) ; } ) . then ( ) . as ( StepVerifier :: create ) . verifyComplete ( ) ; client . select ( ) . from ( Starbucks . class ) . fetch ( ) . all ( ) . flatMap ( it -> { return client . select ( ) . from ( Starbucks . class ) . matching ( where ( "city" ) . is ( it . getCity ( ) ) . and ( "store_number" ) . not ( it . getStoreNumber ( ) ) ) . fetch ( ) . all ( ) . collectList ( ) . map ( shopsInTheSameCity -> { it . setShopsInTheSameCity ( shopsInTheSameCity ) ; return it ; } ) ; } ) . filter ( it -> ! it . shopsInTheSameCity . isEmpty ( ) ) . doOnNext ( System . out :: println ) . as ( StepVerifier :: create ) . expectNextCount ( 924 ) . verifyComplete ( ) ; }
Note: Switch to R2dbcPostgresConnectionExtension, Enable pooling, run in tx
Note : Switch to R2dbcPostgresConnectionExtension Enable pooling run in tx
// // // // //