feat: Run an EventQL query whose rows are deserialized into a given type.

README.md

@@ -329,6 +329,21 @@ try (var rows = client.runEventQlQuery(
 
 *Note that each row returned by the stream is of type `JsonNode` and matches the projection specified in your query.*
 
+To get each row as a type of your own instead, e.g. a record that matches the projection, hand over its class. The client deserializes each row with its data mapper:
+
+```java
+record BookTitle(String title) {}
+
+try (var rows = client.runEventQlQuery(
+  "FROM e IN events PROJECT INTO { title: e.data.title }",
+  BookTitle.class
+)) {
+  rows.forEach(row -> {
+    // ...
+  });
+}
+```
+
 #### Aborting a Query
 
 If you need to abort a query, stop consuming the stream. However, this only works if there is currently a row to consume.

eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/Client.java

@@ -295,15 +295,30 @@ private static ObjectNode toRequest(String subject, String type, Enum<?> ifEvent
      * @return a stream of the rows, each of which matches the projection of the query
      */
     public Stream<JsonNode> runEventQlQuery(String query) {
+        return runEventQlQuery(query, (payload, line) -> payload);
+    }
+
+    /**
+     * Runs an EventQL query, and deserializes each row into the given type.
+     *
+     * <p>This works like {@link #runEventQlQuery(String)}, except that each row is deserialized with the data mapper
+     * of the client, i.e. the one given by {@link ClientOptions#withDataMapper(JsonMapper)}, or the client's own. A
+     * row that does not match the type ends the stream with the exception of the mapper.
+     *
+     * @param <T> the type to deserialize each row into
+     * @param query the query, e.g. {@code FROM e IN events PROJECT INTO { id: e.id, type: e.type }}
+     * @param rowType the type to deserialize each row into, e.g. a record that matches the projection of the query
+     * @return a stream of the rows
+     */
+    public <T> Stream<T> runEventQlQuery(String query, Class<T> rowType) {
+        return runEventQlQuery(query, (payload, line) -> dataMapper.treeToValue(payload, rowType));
+    }
+
+    private <T> Stream<T> runEventQlQuery(String query, LineStream.Parser<T> parser) {
         var body = Json.MAPPER.createObjectNode();
         body.put("query", query);
 
-        return stream(
-                "run EventQL query",
-                post("/api/v1/run-eventql-query", body),
-                "row",
-                (payload, line) -> payload,
-                heartbeatTimeout);
+        return stream("run EventQL query", post("/api/v1/run-eventql-query", body), "row", parser, heartbeatTimeout);
     }
 
     /**

eventsourcingdb/src/test/java/io/thenativeweb/eventsourcingdb/RunEventQlQueryTest.java

@@ -1,14 +1,18 @@
 package io.thenativeweb.eventsourcingdb;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import io.thenativeweb.eventsourcingdb.testcontainers.Container;
 import java.util.List;
 import java.util.Map;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import tools.jackson.core.JacksonException;
 import tools.jackson.databind.JsonNode;
+import tools.jackson.databind.PropertyNamingStrategies;
+import tools.jackson.databind.json.JsonMapper;
 
 class RunEventQlQueryTest {
     private Container container;
@@ -52,4 +56,64 @@ void readsAllRowsTheQueryReturns() {
         assertEquals("1", rows.get(1).path("id").asString());
         assertEquals(42, rows.get(1).path("data").path("value").asInt());
     }
+
+    record Row(String id, int value) {}
+
+    record BookTitle(String bookTitle) {}
+
+    private void writeTwoEvents() {
+        client.writeEvents(List.of(
+                new EventCandidate(
+                        "https://www.eventsourcingdb.io", "/test", "io.eventsourcingdb.test", Map.of("value", 23)),
+                new EventCandidate(
+                        "https://www.eventsourcingdb.io", "/test", "io.eventsourcingdb.test", Map.of("value", 42))));
+    }
+
+    private <T> List<T> query(Client client, String query, Class<T> rowType) {
+        try (var rows = client.runEventQlQuery(query, rowType)) {
+            return rows.toList();
+        }
+    }
+
+    @Test
+    void readsRowsAsTheGivenType() {
+        writeTwoEvents();
+
+        var rows = query(client, "FROM e IN events PROJECT INTO { id: e.id, value: e.data.value }", Row.class);
+
+        assertEquals(List.of(new Row("0", 23), new Row("1", 42)), rows);
+    }
+
+    @Test
+    void readsScalarRowsAsTheGivenType() {
+        writeTwoEvents();
+
+        assertEquals(List.of(2), query(client, "FROM e IN events PROJECT INTO COUNT()", Integer.class));
+    }
+
+    @Test
+    void readsRowsWithTheDataMapperOfTheClient() {
+        writeTwoEvents();
+        var snakeCaseClient = new Client(
+                container.getBaseUrl(),
+                container.getApiToken(),
+                new ClientOptions()
+                        .withDataMapper(JsonMapper.builder()
+                                .propertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE)
+                                .build()));
+
+        var rows = query(
+                snakeCaseClient,
+                "FROM e IN events WHERE e.id == \"0\" PROJECT INTO { book_title: e.type }",
+                BookTitle.class);
+
+        assertEquals(List.of(new BookTitle("io.eventsourcingdb.test")), rows);
+    }
+
+    @Test
+    void endsTheStreamIfARowDoesNotMatchTheType() {
+        writeTwoEvents();
+
+        assertThrows(JacksonException.class, () -> query(client, "FROM e IN events PROJECT INTO e.id", Row.class));
+    }
 }