<dependencies>
  <dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-r2dbc</artifactId>
  <version>2.3.3.RELEASE</version>
 </dependency>

  <!-- R2DBC H2 Driver -->
  <dependency>
   <groupId>io.r2dbc</groupId>
   <artifactId>r2dbc-h2</artifactId>
   <version>${r2dbc-h2.version}</version>
  </dependency>

  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-webflux</artifactId>
  </dependency>

@ConfigurationProperties(prefix = "spring.r2dbc")
public class R2dbcProperties {

 /**
  * Database name. Set if no name is specified in the url. Default to "testdb" when
  * using an embedded database.
  */
 private String name;

 /**
  * Whether to generate a random database name. Ignore any configured name when
  * enabled.
  */
 private boolean generateUniqueName;

 /**
  * R2DBC URL of the database. database name, username, password and pooling options
  * specified in the url take precedence over individual options.
  */
 private String url;

 /**
  * Login username of the database. Set if no username is specified in the url.
  */
 private String username;

 /**
  * Login password of the database. Set if no password is specified in the url.
  */
 private String password;

spring.r2dbc.url=r2dbc:h2:mem://./testdb
spring.r2dbc.username=sa
spring.r2dbc.password=password

@Bean
 public ApplicationRunner initDatabase(DatabaseClient client, UsersDao usersDao) {
  List<String> statements = Arrays.asList(
    "DROP TABLE IF EXISTS USERS;",
    "CREATE TABLE IF NOT EXISTS USERS ( id SERIAL PRIMARY KEY, firstname VARCHAR(100) NOT NULL, lastname VARCHAR(100) NOT NULL);");

  statements.forEach(sql -> executeSql(client,sql)
    .doOnSuccess(count -> log.info("Schema created, rows updated: {}", count))
    .doOnError(error -> log.error("got error : {}",error.getMessage(),error))
    .subscribe()
  );

  return args ->getUser().flatMap(usersDao::save).subscribe(user -> log.info("User saved: {}", user));

 }

 private Flux<Users> getUser() {
  return Flux.just(new Users(null, "John", "Doe"), new Users(null, "Jane", "Doe"));
 }

 private Mono<Integer> executeSql(DatabaseClient client, String sql) {
  return client.execute(sql).fetch().rowsUpdated();
 }

@Data
@AllArgsConstructor
public class Users {

 @Id
 private Integer id;
 private String firstname;
 private String lastname;

 boolean hasId() {
  return id != null;
 }
}

public interface UsersDao extends ReactiveCrudRepository<Users, Long> {

 @Query("select id, firstname, lastname from users c where c.lastname = :lastname")
 Flux<Users> findByLastname(String lastname);
}

@Component
public class UsersService {

 @Resource
 private UsersDao usersDao;

 @Transactional
 public Mono<Users> save(Users user) {

  return usersDao.save(user).map(it -> {

   if (it.getFirstname().equals("flydean")) {
    throw new IllegalStateException();
   } else {
    return it;
   }
  });
 }

}

@RestController
@RequiredArgsConstructor
public class UsersController {

 private final UsersDao usersDao;

 @GetMapping("/users")
 public Flux<Users> findAll() {
  return usersDao.findAll();
 }
}

curl "localhost:8080/users" 
[{"id":1,"firstname":"John","lastname":"Doe"},{"id":2,"firstname":"Jane","lastname":"Doe"}]%