- Add a new benchmark for spring-webflux using pgclient to asynchronously access the database. (#4111)

- Refactored the code to make it easy to add new datasource implementation without needing to add a new controller.

frameworks/Java/spring-webflux/README.md

@@ -4,44 +4,43 @@ This is the Spring Webflux portion of a [benchmarking test suite](../) comparing
 
 Netty is used for the async web server, with nearly everything configured with default settings. The only thing changed is Hikari can use up to (2 * cores count) connections (the default is 10). See [About-Pool-Sizing](https://github.com/brettwooldridge/HikariCP/wiki/About-Pool-Sizing)
 
-A fixed thread pool of size equals to the number of database connections is used to run all the blocking code (postgresql database accesses) to not block netty's event loop.
+A fixed thread pool of size equals to the number of database connections is used to run all the blocking code (jdbc database accesses) to not block netty's event loop.
 
-For postgresql access, JdbcTemplate is used.
-For mongoDB access, spring-data-mongodb with reactive support is used.
+For postgresql access, there are two implementations.
+* [JdbcDbRepository](src/main/java/benchmark/JdbcDbRepository.java) is using JdbcTemplate.
+* [PgClientDbRepository](src/main/java/benchmark/PgClientDbRepository.java) is using reactive-pg-client
+For mongoDB access, spring-data-mongodb with reactive support is used. See [MongoDbRepository](src/main/java/benchmark/MongoDbRepository.java)
 
 ### Plaintext Test
 
-* [Plaintext test source](src/main/java/benchmark/BenchmarkController.java)
+* [Plaintext test source](src/main/java/benchmark/Controller/BenchmarkController.java)
 
 ### JSON Serialization Test
 
-* [JSON test source](src/main/java/benchmark/BenchmarkController.java)
+* [JSON test source](src/main/java/benchmark/Controller/BenchmarkController.java)
 
 ### Database Query Test
 
-* [Postgresql Query test source](src/main/java/benchmark/SQLController.java)
-* [MongoDB Query test source](src/main/java/benchmark/NoSQLController.java)
+* [Query test source](src/main/java/benchmark/Controller/ReactiveController.java)
 
 ### Database Queries Test
 
-* [Postgresql Queries test source](src/main/java/benchmark/SQLController.java)
-* [MongoDB Queries test source](src/main/java/benchmark/NoSQLController.java)
+* [Queries test source](src/main/java/benchmark/Controller/ReactiveController.java)
 
 ### Database Update Test
 
-* [Postgresql Update test source](src/main/java/benchmark/SQLController.java)
-* [MongoDB Update test source](src/main/java/benchmark/NoSQLController.java)
+* [Update test source](src/main/java/benchmark/Controller/ReactiveController.java)
 
 ### Template rendering Test
 
-* [Postgresql Template rendering test source](src/main/java/benchmark/SQLController.java)
-* [MongoDB Template rendering test source](src/main/java/benchmark/NoSQLController.java)
+* [Template rendering test source](src/main/java/benchmark/Controller/ReactiveController.java)
 
 ## Versions
 
 * [Java OpenJDK 10](http://openjdk.java.net/)
-* [Spring boot 2.0.4](https://spring.io/projects/spring-boot)
+* [Spring boot 2.0.5](https://spring.io/projects/spring-boot)
 * [Spring data mongodb 2.0.9](https://projects.spring.io/spring-data-mongodb/)
+* [reactive-pg-client 0.10.6](https://github.com/reactiverse/reactive-pg-client)
 
 ## Test URLs
 
@@ -56,19 +55,15 @@ For mongoDB access, spring-data-mongodb with reactive support is used.
 ### Database Query Test
 
     http://localhost:8080/db
-    http://localhost:8080/mongo/db
 
 ### Database Queries Test
 
     http://localhost:8080/queries?queries=5
-    http://localhost:8080/mongo/queries?queries=5
 
 ### Database Update Test
 
     http://localhost:8080/updates?queries=5
-    http://localhost:8080/mongo/updates?queries=5
 
 ### Template rendering Test
 
     http://localhost:8080/fortunes
-    http://localhost:8080/mongo/fortunes

frameworks/Java/spring-webflux/benchmark_config.json

@@ -20,7 +20,28 @@
       "notes": "",
       "versus": "spring"
     },
-    "postgres": {
+    "mongo": {
+      "db_url": "/db",
+      "query_url": "/queries?queries=",
+      "fortune_url": "/fortunes",
+      "update_url": "/updates?queries=",
+      "port": 8080,
+      "approach": "Realistic",
+      "classification": "Fullstack",
+      "database": "mongodb",
+      "framework": "spring",
+      "language": "Java",
+      "flavor": "None",
+      "orm": "Full",
+      "platform": "Netty",
+      "webserver": "None",
+      "os": "Linux",
+      "database_os": "Linux",
+      "display_name": "spring-webflux-mongo",
+      "notes": "",
+      "versus": "spring"
+    },
+    "pgclient": {
       "db_url": "/db",
       "query_url": "/queries?queries=",
       "fortune_url": "/fortunes",
@@ -37,28 +58,28 @@
       "webserver": "None",
       "os": "Linux",
       "database_os": "Linux",
-      "display_name": "spring-webflux-postgres",
+      "display_name": "spring-webflux-pgclient",
       "notes": "",
       "versus": "spring"
     },
-    "mongo": {
-      "db_url": "/mongo/db",
-      "query_url": "/mongo/queries?queries=",
-      "fortune_url": "/mongo/fortunes",
-      "update_url": "/mongo/updates?queries=",
+    "jdbc": {
+      "db_url": "/db",
+      "query_url": "/queries?queries=",
+      "fortune_url": "/fortunes",
+      "update_url": "/updates?queries=",
       "port": 8080,
       "approach": "Realistic",
       "classification": "Fullstack",
-      "database": "mongodb",
+      "database": "Postgres",
       "framework": "spring",
       "language": "Java",
       "flavor": "None",
-      "orm": "Full",
+      "orm": "Micro",
       "platform": "Netty",
       "webserver": "None",
       "os": "Linux",
       "database_os": "Linux",
-      "display_name": "spring-webflux-mongo",
+      "display_name": "spring-webflux-jdbc",
       "notes": "",
       "versus": "spring"
     }

frameworks/Java/spring-webflux/pom.xml

@@ -13,7 +13,7 @@
     <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.0.4.RELEASE</version>
+        <version>2.0.5.RELEASE</version>
     </parent>
 
     <properties>
@@ -21,6 +21,7 @@
         <maven.compiler.target>10</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <postgresql.version>42.2.5</postgresql.version>
+        <pgclient.version>0.10.6</pgclient.version>
     </properties>
 
     <dependencies>
@@ -45,6 +46,11 @@
             <artifactId>postgresql</artifactId>
             <version>${postgresql.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.reactiverse</groupId>
+            <artifactId>reactive-pg-client</artifactId>
+            <version>${pgclient.version}</version>
+        </dependency>
     </dependencies>
 
     <build>

frameworks/Java/spring-webflux/spring-webflux-postgres.dockerfile → frameworks/Java/spring-webflux/spring-webflux-jdbc.dockerfile

@@ -7,4 +7,4 @@ RUN mvn package -q
 FROM openjdk:10-jre-slim
 WORKDIR /spring
 COPY --from=maven /spring/target/spring-webflux-benchmark-1.0-SNAPSHOT.jar app.jar
-CMD ["java", "-server", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-Dlogging.level.root=OFF", "-jar", "app.jar"]
+CMD ["java", "-server", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-Dlogging.level.root=OFF", "-jar", "app.jar", "--spring.profiles.active=jdbc"]

frameworks/Java/spring-webflux/spring-webflux-mongo.dockerfile

@@ -7,4 +7,4 @@ RUN mvn package -q
 FROM openjdk:10-jre-slim
 WORKDIR /spring
 COPY --from=maven /spring/target/spring-webflux-benchmark-1.0-SNAPSHOT.jar app.jar
-CMD ["java", "-server", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-Dlogging.level.root=OFF", "-jar", "app.jar"]
+CMD ["java", "-server", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-Dlogging.level.root=OFF", "-jar", "app.jar", "--spring.profiles.active=mongo"]

frameworks/Java/spring-webflux/spring-webflux-pgclient.dockerfile

@@ -0,0 +1,10 @@
+FROM maven:3.5.3-jdk-10-slim as maven
+WORKDIR /spring
+COPY src src
+COPY pom.xml pom.xml
+RUN mvn package -q
+
+FROM openjdk:10-jre-slim
+WORKDIR /spring
+COPY --from=maven /spring/target/spring-webflux-benchmark-1.0-SNAPSHOT.jar app.jar
+CMD ["java", "-server", "-XX:+UseNUMA", "-XX:+UseParallelGC", "-Dlogging.level.root=OFF", "-jar", "app.jar", "--spring.profiles.active=pgclient"]

frameworks/Java/spring-webflux/src/main/java/benchmark/App.java

@@ -7,8 +7,10 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
 import org.springframework.boot.autoconfigure.mustache.MustacheResourceTemplateLoader;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.boot.web.reactive.result.view.MustacheViewResolver;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Profile;
 import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.web.reactive.config.EnableWebFlux;
@@ -24,6 +26,7 @@ import java.util.concurrent.Executors;
 @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
 @EnableWebFlux
 @EnableScheduling
+@EnableConfigurationProperties
 public class App implements WebFluxConfigurer {
 
     public static void main(String[] args) {
@@ -62,6 +65,7 @@ public class App implements WebFluxConfigurer {
     }
 
     @Bean
+    @Profile("jdbc")
     public DataSource datasource(DataSourceProperties dataSourceProperties) {
         HikariDataSource dataSource = (HikariDataSource) dataSourceProperties.initializeDataSourceBuilder().type(HikariDataSource.class).build();
         dataSource.setMaximumPoolSize(Runtime.getRuntime().availableProcessors() * 2);

frameworks/Java/spring-webflux/src/main/java/benchmark/BaseDBController.java

@@ -1,22 +0,0 @@
-package benchmark;
-
-import java.util.concurrent.ThreadLocalRandom;
-
-public abstract class BaseDBController {
-    protected int randomWorldNumber() {
-        return 1 + ThreadLocalRandom.current().nextInt(10000);
-    }
-
-    protected int parseQueryCount(String textValue) {
-        if (textValue == null) {
-            return 1;
-        }
-        int parsedValue;
-        try {
-            parsedValue = Integer.parseInt(textValue);
-        } catch (NumberFormatException e) {
-            return 1;
-        }
-        return Math.min(500, Math.max(1, parsedValue));
-    }
-}

frameworks/Java/spring-webflux/src/main/java/benchmark/PgClients.java

@@ -0,0 +1,19 @@
+package benchmark;
+
+import io.reactiverse.pgclient.PgClient;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.stream.Stream;
+
+public class PgClients {
+    private final Iterator<PgClient> iterator;
+
+    public PgClients(Collection<PgClient> clients) {
+        iterator = Stream.generate(() -> clients).flatMap(Collection::stream).iterator();
+    }
+
+    public synchronized PgClient getOne() {
+        return iterator.next();
+    }
+}

frameworks/Java/spring-webflux/src/main/java/benchmark/SQLController.java

@@ -1,87 +0,0 @@
-package benchmark;
-
-import benchmark.model.Fortune;
-import benchmark.model.World;
-import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.stereotype.Controller;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestParam;
-import org.springframework.web.bind.annotation.ResponseBody;
-import org.springframework.web.reactive.result.view.Rendering;
-import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Scheduler;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ThreadLocalRandom;
-
-import static java.util.Comparator.comparing;
-
-@Controller
-public final class SQLController extends BaseDBController {
-
-    private final JdbcTemplate jdbcTemplate;
-    private final Scheduler scheduler;
-
-    public SQLController(JdbcTemplate jdbcTemplate, Scheduler scheduler) {
-        this.jdbcTemplate = jdbcTemplate;
-        this.scheduler = scheduler;
-    }
-
-    @GetMapping(value = "/db", produces = "application/json")
-    @ResponseBody
-    public Mono<World> db() {
-        return Mono.fromCallable(() -> randomWorld()).subscribeOn(scheduler);
-    }
-
-    @GetMapping(value = "/queries", produces = "application/json")
-    @ResponseBody
-    public Mono<World[]> queries(@RequestParam String queries) {
-        return Mono.fromCallable(() -> {
-            var worlds = new World[parseQueryCount(queries)];
-            Arrays.setAll(worlds, i -> randomWorld());
-            return worlds;
-        }).subscribeOn(scheduler);
-    }
-
-    @GetMapping(value = "/updates", produces = "application/json")
-    @ResponseBody
-    public Mono<World[]> updates(@RequestParam String queries) {
-        return Mono.fromCallable(() -> {
-            var worlds = new World[parseQueryCount(queries)];
-            Arrays.setAll(worlds, i -> randomWorld());
-            for (var world : worlds) {
-                world.randomNumber = randomWorldNumber();
-                jdbcTemplate.update(
-                        "UPDATE world SET randomnumber = ? WHERE id = ?",
-                        world.randomNumber,
-                        world.id);
-            }
-            return worlds;
-        }).subscribeOn(scheduler);
-    }
-
-    @GetMapping(value = "/fortunes")
-    public Rendering fortunes() {
-        Mono<List<Fortune>> fortunes = Mono.fromCallable(() -> {
-            var list =
-                    jdbcTemplate.query(
-                            "SELECT * FROM fortune",
-                            (rs, rn) -> new Fortune(rs.getInt("id"), rs.getString("message")));
-
-            list.add(new Fortune(0, "Additional fortune added at request time."));
-            list.sort(comparing(fortune -> fortune.message));
-            return list;
-        }).subscribeOn(scheduler);
-
-        return Rendering.view("fortunes").modelAttribute("fortunes", fortunes).build();
-    }
-
-    private World randomWorld() {
-        return jdbcTemplate.queryForObject(
-                "SELECT * FROM world WHERE id = ?",
-                (rs, rn) -> new World(rs.getInt("id"), rs.getInt("randomnumber")),
-                randomWorldNumber());
-    }
-}

frameworks/Java/spring-webflux/src/main/java/benchmark/config/PgClientConfig.java

@@ -0,0 +1,94 @@
+package benchmark.config;
+
+import benchmark.PgClients;
+import io.reactiverse.pgclient.PgClient;
+import io.reactiverse.pgclient.PgPool;
+import io.reactiverse.pgclient.PgPoolOptions;
+import io.vertx.core.Vertx;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Configuration
+@Profile("pgclient")
+@ConfigurationProperties(prefix = "database")
+public class PgClientConfig {
+    private String name;
+    private String host;
+    private int port;
+    private String username;
+    private String password;
+
+    @Bean
+    public Vertx vertx() {
+        return Vertx.vertx();
+    }
+
+    @Bean
+    public PgClients pgClients(Vertx vertx) {
+        List<PgClient> clients = new ArrayList<>();
+
+        for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) {
+            clients.add(pgClient(vertx));
+        }
+
+        return new PgClients(clients);
+    }
+
+
+    public PgPool pgClient(Vertx vertx) {
+        PgPoolOptions options = new PgPoolOptions();
+        options.setDatabase(name);
+        options.setHost(host);
+        options.setPort(port);
+        options.setUser(username);
+        options.setPassword(password);
+        options.setCachePreparedStatements(true);
+        options.setMaxSize(1);
+        return PgClient.pool(vertx, options);
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getHost() {
+        return host;
+    }
+
+    public void setHost(String host) {
+        this.host = host;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public void setPort(int port) {
+        this.port = port;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+}

frameworks/Java/spring-webflux/src/main/java/benchmark/ReactiveMongoConfig.java → frameworks/Java/spring-webflux/src/main/java/benchmark/config/ReactiveMongoConfig.java

@@ -1,28 +1,52 @@
-package benchmark;
+package benchmark.config;
 
 import com.mongodb.reactivestreams.client.MongoClient;
 import com.mongodb.reactivestreams.client.MongoClients;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
 import org.springframework.data.mongodb.config.AbstractReactiveMongoConfiguration;
 import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
 import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
 
 @Configuration
 @EnableReactiveMongoRepositories
+@Profile("mongo")
+@ConfigurationProperties(prefix = "database")
 public class ReactiveMongoConfig extends AbstractReactiveMongoConfiguration {
+    private String url;
+    private String name;
+
     @Override
     public MongoClient reactiveMongoClient() {
-        return MongoClients.create("mongodb://tfb-database:27017");
+        return MongoClients.create(url);
     }
 
     @Override
     protected String getDatabaseName() {
-        return "hello_world";
+        return name;
     }
 
     @Bean
     public ReactiveMongoTemplate reactiveMongoTemplate() {
         return new ReactiveMongoTemplate(reactiveMongoClient(), getDatabaseName());
     }
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
 }

frameworks/Java/spring-webflux/src/main/java/benchmark/BenchmarkController.java → frameworks/Java/spring-webflux/src/main/java/benchmark/controller/BenchmarkController.java

@@ -1,4 +1,4 @@
-package benchmark;
+package benchmark.controller;
 
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Controller;

frameworks/Java/spring-webflux/src/main/java/benchmark/NoSQLController.java → frameworks/Java/spring-webflux/src/main/java/benchmark/controller/ReactiveController.java

@@ -1,45 +1,37 @@
-package benchmark;
+package benchmark.controller;
 
 import benchmark.model.Fortune;
 import benchmark.model.World;
-import org.springframework.data.mongodb.core.FindAndModifyOptions;
-import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
-import org.springframework.data.mongodb.core.query.Criteria;
-import org.springframework.data.mongodb.core.query.Query;
-import org.springframework.data.mongodb.core.query.Update;
+import benchmark.repository.DbRepository;
+import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.reactive.result.view.Rendering;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Scheduler;
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
 
 import static java.util.Comparator.comparing;
-import static org.springframework.data.mongodb.core.FindAndModifyOptions.*;
-import static org.springframework.data.mongodb.core.query.Criteria.*;
-import static org.springframework.data.mongodb.core.query.Query.*;
-import static org.springframework.data.mongodb.core.query.Update.*;
 
 @Controller()
-@RequestMapping("/mongo")
-public final class NoSQLController extends BaseDBController {
+@Profile(value = {"jdbc", "pgclient", "mongo"})
+public final class ReactiveController {
 
-    private final ReactiveMongoTemplate mongoTemplate;
+    private final DbRepository dbRepository;
 
-    public NoSQLController(ReactiveMongoTemplate mongoTemplate) {
-        this.mongoTemplate = mongoTemplate;
+    public ReactiveController(DbRepository dbRepository) {
+        this.dbRepository = dbRepository;
     }
 
     @GetMapping(value = "/db", produces = "application/json")
     @ResponseBody
     public Mono<World> db() {
-        return mongoTemplate.findById(randomWorldNumber(), World.class);
+        return dbRepository.getWorld(randomWorldNumber());
     }
 
     @GetMapping(value = "/queries", produces = "application/json")
@@ -56,18 +48,14 @@ public final class NoSQLController extends BaseDBController {
     public Mono<List<World>> updates(@RequestParam String queries) {
         Mono<World>[] worlds = new Mono[parseQueryCount(queries)];
 
-        Arrays.setAll(worlds, i -> mongoTemplate.findAndModify(
-                query(where("id").is(randomWorldNumber())),
-                update("randomNumber", randomWorldNumber()),
-                options().returnNew(true),
-                World.class));
+        Arrays.setAll(worlds, i -> dbRepository.findAndUpdateWorld(randomWorldNumber(), randomWorldNumber()));
 
         return Flux.merge(worlds).collectList();
     }
 
     @GetMapping(value = "/fortunes")
     public Rendering fortunes() {
-        Mono<List<Fortune>> result = mongoTemplate.findAll(Fortune.class).collectList().flatMap(fortunes -> {
+        Mono<List<Fortune>> result = dbRepository.fortunes().collectList().flatMap(fortunes -> {
             fortunes.add(new Fortune(0, "Additional fortune added at request time."));
             fortunes.sort(comparing(fortune -> fortune.message));
             return Mono.just(fortunes);
@@ -75,4 +63,21 @@ public final class NoSQLController extends BaseDBController {
 
         return Rendering.view("fortunes").modelAttribute("fortunes", result).build();
     }
+
+    protected int randomWorldNumber() {
+        return 1 + ThreadLocalRandom.current().nextInt(10000);
+    }
+
+    protected int parseQueryCount(String textValue) {
+        if (textValue == null) {
+            return 1;
+        }
+        int parsedValue;
+        try {
+            parsedValue = Integer.parseInt(textValue);
+        } catch (NumberFormatException e) {
+            return 1;
+        }
+        return Math.min(500, Math.max(1, parsedValue));
+    }
 }

frameworks/Java/spring-webflux/src/main/java/benchmark/repository/DbRepository.java

@@ -0,0 +1,14 @@
+package benchmark.repository;
+
+import benchmark.model.Fortune;
+import benchmark.model.World;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public interface DbRepository {
+    Mono<World> getWorld(int id);
+
+    Mono<World> findAndUpdateWorld(int id, int randomNumber);
+
+    Flux<Fortune> fortunes();
+}

frameworks/Java/spring-webflux/src/main/java/benchmark/repository/JdbcDbRepository.java

@@ -0,0 +1,63 @@
+package benchmark.repository;
+
+import benchmark.model.Fortune;
+import benchmark.model.World;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+
+@Component
+@Profile("jdbc")
+public class JdbcDbRepository implements DbRepository {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final JdbcTemplate jdbcTemplate;
+    private final Scheduler scheduler;
+
+    public JdbcDbRepository(JdbcTemplate jdbcTemplate, Scheduler scheduler) {
+        this.jdbcTemplate = jdbcTemplate;
+        this.scheduler = scheduler;
+    }
+
+    @Override
+    public Mono<World> getWorld(int id) {
+        log.debug("getWorld({})", id);
+        return Mono.fromCallable(() -> {
+            return jdbcTemplate.queryForObject(
+                    "SELECT * FROM world WHERE id = ?",
+                    (rs, rn) -> new World(rs.getInt("id"), rs.getInt("randomnumber")),
+                    id);
+        }).subscribeOn(scheduler);
+    }
+
+    private Mono<World> updateWorld(World world) {
+        return Mono.fromCallable(() -> {
+            jdbcTemplate.update(
+                    "UPDATE world SET randomnumber = ? WHERE id = ?",
+                    world.randomNumber,
+                    world.id);
+            return world;
+        }).subscribeOn(scheduler);
+    }
+
+    @Override
+    public Mono<World> findAndUpdateWorld(int id, int randomNumber) {
+        return getWorld(id).flatMap(world -> {
+            world.randomNumber = randomNumber;
+            return updateWorld(world);
+        });
+    }
+
+    @Override
+    public Flux<Fortune> fortunes() {
+        return Mono.fromCallable(() -> {
+            return jdbcTemplate.query(
+                    "SELECT * FROM fortune",
+                    (rs, rn) -> new Fortune(rs.getInt("id"), rs.getString("message")));
+        }).subscribeOn(scheduler).flatMapIterable(fortunes -> fortunes);
+    }
+}

frameworks/Java/spring-webflux/src/main/java/benchmark/repository/MongoDbRepository.java

@@ -0,0 +1,47 @@
+package benchmark.repository;
+
+import benchmark.model.Fortune;
+import benchmark.model.World;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import static org.springframework.data.mongodb.core.FindAndModifyOptions.options;
+import static org.springframework.data.mongodb.core.query.Criteria.where;
+import static org.springframework.data.mongodb.core.query.Query.query;
+import static org.springframework.data.mongodb.core.query.Update.update;
+
+@Component
+@Profile("mongo")
+public class MongoDbRepository implements DbRepository {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final ReactiveMongoTemplate mongoTemplate;
+
+    public MongoDbRepository(ReactiveMongoTemplate mongoTemplate) {
+        this.mongoTemplate = mongoTemplate;
+    }
+
+    @Override
+    public Mono<World> getWorld(int id) {
+        log.debug("getWorld({})", id);
+        return mongoTemplate.findById(id, World.class);
+    }
+
+    @Override
+    public Mono<World> findAndUpdateWorld(int id, int randomNumber) {
+        return mongoTemplate.findAndModify(
+                query(where("id").is(id)),
+                update("randomNumber", randomNumber),
+                options().returnNew(true),
+                World.class);
+    }
+
+    @Override
+    public Flux<Fortune> fortunes() {
+        return mongoTemplate.findAll(Fortune.class);
+    }
+}

frameworks/Java/spring-webflux/src/main/java/benchmark/repository/PgClientDbRepository.java

@@ -0,0 +1,79 @@
+package benchmark.repository;
+
+import benchmark.PgClients;
+import benchmark.model.Fortune;
+import benchmark.model.World;
+import io.reactiverse.pgclient.PgIterator;
+import io.reactiverse.pgclient.Row;
+import io.reactiverse.pgclient.Tuple;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@Component
+@Profile("pgclient")
+public class PgClientDbRepository implements DbRepository {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private final PgClients pgClients;
+
+    public PgClientDbRepository(PgClients pgClients) {
+        this.pgClients = pgClients;
+    }
+
+    @Override
+    public Mono<World> getWorld(int id) {
+        return Mono.create(sink ->
+                pgClients.getOne().preparedQuery("SELECT * FROM world WHERE id = $1", Tuple.of(id), ar -> {
+                    if (ar.failed()) {
+                        sink.error(ar.cause());
+                    } else {
+
+                        final Row row = ar.result().iterator().next();
+
+                        World world = new World(row.getInteger(0), row.getInteger(1));
+                        sink.success(world);
+                    }
+                }));
+    }
+
+    private Mono<World> updateWorld(World world) {
+        return Mono.create(sink -> {
+            pgClients.getOne().preparedQuery("UPDATE world SET randomnumber = $1 WHERE id = $2", Tuple.of(world.randomNumber, world.id), ar -> {
+                if (ar.failed()) {
+                    sink.error(ar.cause());
+                } else {
+                    sink.success(world);
+                }
+            });
+        });
+    }
+
+    @Override
+    public Mono<World> findAndUpdateWorld(int id, int randomNumber) {
+        return getWorld(id).flatMap(world -> {
+            world.randomNumber = randomNumber;
+            return updateWorld(world);
+        });
+    }
+
+    @Override
+    public Flux<Fortune> fortunes() {
+        return Flux.create(sink ->
+                pgClients.getOne().preparedQuery("SELECT * FROM fortune", ar -> {
+                    if (ar.failed()) {
+                        sink.error(ar.cause());
+                        return;
+                    }
+
+                    PgIterator resultSet = ar.result().iterator();
+                    while (resultSet.hasNext()) {
+                        Tuple row = resultSet.next();
+                        sink.next(new Fortune(row.getInteger(0), row.getString(1)));
+                    }
+                    sink.complete();
+                }));
+    }
+}

frameworks/Java/spring-webflux/src/main/resources/application.properties

@@ -1,3 +0,0 @@
-spring.datasource.url=jdbc:postgresql://tfb-database:5432/hello_world
-spring.datasource.username=benchmarkdbuser
-spring.datasource.password=benchmarkdbpass

frameworks/Java/spring-webflux/src/main/resources/application.yml

@@ -0,0 +1,26 @@
+---
+spring:
+  profiles: jdbc
+  datasource:
+    url: jdbc:postgresql://tfb-database:5432/hello_world
+    username: benchmarkdbuser
+    password: benchmarkdbpass
+---
+
+spring:
+  profiles: pgclient
+
+database:
+  name: hello_world
+  host: tfb-database
+  port: 5432
+  username: benchmarkdbuser
+  password: benchmarkdbpass
+---
+
+spring:
+  profiles: mongo
+
+database:
+  url: mongodb://tfb-database:27017
+  name: hello_world