diff --git a/README.md b/README.md index f95541e..010ed7a 100644 --- a/README.md +++ b/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. diff --git a/eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/Client.java b/eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/Client.java index 56b5a44..19bb9c9 100644 --- a/eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/Client.java +++ b/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 runEventQlQuery(String query) { + return runEventQlQuery(query, (payload, line) -> payload); + } + + /** + * Runs an EventQL query, and deserializes each row into the given type. + * + *

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 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 Stream runEventQlQuery(String query, Class rowType) { + return runEventQlQuery(query, (payload, line) -> dataMapper.treeToValue(payload, rowType)); + } + + private Stream runEventQlQuery(String query, LineStream.Parser 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); } /** diff --git a/eventsourcingdb/src/test/java/io/thenativeweb/eventsourcingdb/RunEventQlQueryTest.java b/eventsourcingdb/src/test/java/io/thenativeweb/eventsourcingdb/RunEventQlQueryTest.java index b3eb1ee..2d6380e 100644 --- a/eventsourcingdb/src/test/java/io/thenativeweb/eventsourcingdb/RunEventQlQueryTest.java +++ b/eventsourcingdb/src/test/java/io/thenativeweb/eventsourcingdb/RunEventQlQueryTest.java @@ -1,6 +1,7 @@ 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; @@ -8,7 +9,10 @@ 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 List query(Client client, String query, Class 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)); + } }