diff --git a/akka-javasdk-tests/src/test/java/akkajavasdk/SdkIntegrationTest.java b/akka-javasdk-tests/src/test/java/akkajavasdk/SdkIntegrationTest.java index a0aa64ac9..ed51bb45e 100644 --- a/akka-javasdk-tests/src/test/java/akkajavasdk/SdkIntegrationTest.java +++ b/akka-javasdk-tests/src/test/java/akkajavasdk/SdkIntegrationTest.java @@ -6,7 +6,6 @@ import akka.javasdk.Metadata; import akka.javasdk.client.EventSourcedEntityClient; -import akka.javasdk.client.NoEntryFoundException; import akka.javasdk.testkit.TestKit; import akka.javasdk.testkit.TestKitSupport; import akkajavasdk.components.actions.echo.ActionWithMetadata; @@ -15,30 +14,15 @@ import akkajavasdk.components.eventsourcedentities.counter.Counter; import akkajavasdk.components.eventsourcedentities.counter.CounterEntity; import akkajavasdk.components.keyvalueentities.customer.CustomerEntity; -import akkajavasdk.components.keyvalueentities.user.AssignedCounterEntity; import akkajavasdk.components.keyvalueentities.user.User; import akkajavasdk.components.keyvalueentities.user.UserEntity; import akkajavasdk.components.keyvalueentities.user.UserSideEffect; import akkajavasdk.components.views.counter.CountersByValue; -import akkajavasdk.components.views.counter.CountersByValueSubscriptions; -import akkajavasdk.components.views.counter.CountersByValueWithIgnore; import akkajavasdk.components.views.customer.CustomerByCreationTime; -import akkajavasdk.components.views.UserCounter; -import akkajavasdk.components.views.UserCounters; -import akkajavasdk.components.views.UserCountersView; -import akkajavasdk.components.views.user.UserWithVersion; -import akkajavasdk.components.views.user.UserWithVersionView; -import akkajavasdk.components.views.user.UsersByEmailAndName; -import akkajavasdk.components.views.user.UsersByName; -import akkajavasdk.components.views.user.UsersByPrimitives; -import akkajavasdk.components.views.user.UsersView; -import akkajavasdk.components.views.hierarchy.HierarchyCountersByValue; import org.awaitility.Awaitility; import org.hamcrest.core.IsEqual; import org.hamcrest.core.IsNull; import org.jetbrains.annotations.NotNull; -import org.jetbrains.annotations.Nullable; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -178,30 +162,6 @@ public void verifyActionIsNotSubscribedToMultiplyAndRouterIgnores() { }); } - @Test - public void verifyViewIsNotSubscribedToMultiplyAndRouterIgnores() { - - var entityId = "counterId4"; - EventSourcedEntityClient counterClient = componentClient.forEventSourcedEntity(entityId); - await(counterClient.method(CounterEntity::increase).invokeAsync(1)); - await(counterClient.method(CounterEntity::times).invokeAsync(2)); - Integer counterGet = await(counterClient.method(CounterEntity::increase).invokeAsync(1)); - - assertThat(counterGet).isEqualTo(1 * 2 + 1); - - Awaitility.await() - .ignoreExceptions() - .atMost(10, TimeUnit.SECONDS) - .untilAsserted( - () -> { - var byValue = await( - componentClient.forView() - .method(CountersByValueWithIgnore::getCounterByValue) - .invokeAsync(CountersByValueWithIgnore.queryParam(2))); - - assertThat(byValue.value()).isEqualTo(1 + 1); - }); - } @Test public void verifyFindCounterByValue() { @@ -234,86 +194,9 @@ public void verifyFindCounterByValue() { }); } - @Disabled // pending primitive query parameters working - @Test - public void verifyHierarchyView() { - - var emptyCounter = await( - componentClient.forView() - .method(HierarchyCountersByValue::getCounterByValue) - .invokeAsync(201)); - - assertThat(emptyCounter).isEmpty(); - - await( - componentClient.forEventSourcedEntity("bcd") - .method(CounterEntity::increase) - .invokeAsync(201)); - - - // the view is eventually updated - Awaitility.await() - .ignoreExceptions() - .atMost(15, TimeUnit.of(SECONDS)) - .untilAsserted( - () -> { - var byValue = await( - componentClient.forView() - .method(HierarchyCountersByValue::getCounterByValue) - .invokeAsync(201)); - - assertThat(byValue).hasValue(new Counter(201)); - }); - } - - @Test - public void verifyCounterViewMultipleSubscriptions() { - await( - componentClient.forEventSourcedEntity("hello2") - .method(CounterEntity::increase) - .invokeAsync(74)); - await( - componentClient.forEventSourcedEntity("hello3") - .method(CounterEntity::increase) - .invokeAsync(74)); - Awaitility.await() - .ignoreExceptions() - .atMost(20, TimeUnit.SECONDS) - .until( - () -> - await(componentClient.forView() - .method(CountersByValueSubscriptions::getCounterByValue) - .invokeAsync(new CountersByValueSubscriptions.QueryParameters(74))) - .counters().size(), - new IsEqual<>(2)); - } - - @Test - public void verifyTransformedUserViewWiring() { - - TestUser user = new TestUser("123", "john123@doe.com", "JohnDoe"); - - createUser(user); - - // the view is eventually updated - Awaitility.await() - .ignoreExceptions() - .atMost(15, TimeUnit.of(SECONDS)) - .until(() -> getUserByEmail(user.email()).version, - new IsEqual(1)); - - updateUser(user.withName("JohnDoeJr")); - - // the view is eventually updated - Awaitility.await() - .ignoreExceptions() - .atMost(15, TimeUnit.of(SECONDS)) - .until(() -> getUserByEmail(user.email()).version, - new IsEqual(2)); - } @Test public void verifyUserSubscriptionAction() { @@ -338,204 +221,6 @@ public void verifyUserSubscriptionAction() { } - @Disabled // pending primitive query parameters working - @Test - public void shouldAcceptPrimitivesForViewQueries() { - - TestUser user1 = new TestUser("654321", "john654321@doe.com", "Bob2"); - TestUser user2 = new TestUser("7654321", "john7654321@doe.com", "Bob3"); - createUser(user1); - createUser(user2); - - Awaitility.await() - .ignoreExceptions() - .atMost(10, TimeUnit.of(SECONDS)) - .untilAsserted(() -> { - var resultByString = await( - componentClient.forView() - .method(UsersByPrimitives::getUserByString) - .invokeAsync(user1.email())); - assertThat(resultByString.users()).isNotEmpty(); - - var resultByInt = await( - componentClient.forView() - .method(UsersByPrimitives::getUserByInt) - .invokeAsync(123)); - assertThat(resultByInt.users()).isNotEmpty(); - - var resultByLong = await( - componentClient.forView() - .method(UsersByPrimitives::getUserByLong) - .invokeAsync(321l)); - assertThat(resultByLong.users()).isNotEmpty(); - - var resultByDouble = await( - componentClient.forView() - .method(UsersByPrimitives::getUserByDouble) - .invokeAsync(12.3d)); - assertThat(resultByDouble.users()).isNotEmpty(); - - var resultByBoolean = await( - componentClient.forView() - .method(UsersByPrimitives::getUserByBoolean) - .invokeAsync(true)); - assertThat(resultByBoolean.users()).isNotEmpty(); - - var resultByEmails = await( - componentClient.forView() - .method(UsersByPrimitives::getUserByEmails) - .invokeAsync(List.of(user1.email(), user2.email()))); - assertThat(resultByEmails.users()).hasSize(2); - }); - } - - @Test - public void shouldDeleteValueEntityAndDeleteViewsState() { - - TestUser user = new TestUser("userId", "john123@doe.com", "Bob123"); - createUser(user); - - Awaitility.await() - .ignoreExceptions() - .atMost(15, TimeUnit.of(SECONDS)) - .until(() -> getUserByEmail(user.email()).version, - new IsEqual(1)); - - Awaitility.await() - .ignoreExceptions() - .atMost(15, TimeUnit.of(SECONDS)) - .until(() -> getUsersByName(user.name()).size(), - new IsEqual(1)); - - deleteUser(user); - - Awaitility.await() - .atMost(15, TimeUnit.of(SECONDS)) - .ignoreExceptions() - .untilAsserted( - () -> { - var ex = - failed( - componentClient.forView() - .method(UserWithVersionView::getUser) - .invokeAsync(UserWithVersionView.queryParam(user.email()))); - assertThat(ex).isInstanceOf(NoEntryFoundException.class); - }); - - Awaitility.await() - .ignoreExceptions() - .atMost(15, TimeUnit.of(SECONDS)) - .until(() -> getUsersByName(user.name()).size(), - new IsEqual(0)); - } - - @Test - public void verifyFindUsersByEmail() { - - TestUser user = new TestUser("JohnDoe", "john3@doe.com", "JohnDoe"); - createUser(user); - - // the view is eventually updated - Awaitility.await() - .ignoreExceptions() - .atMost(10, TimeUnit.SECONDS) - .untilAsserted( - () -> { - var byEmail = await( - componentClient.forView() - .method(UsersView::getUserByEmail) - .invokeAsync(UsersView.byEmailParam(user.email()))); - - assertThat(byEmail.email).isEqualTo(user.email()); - }); - } - - @Test - public void verifyFindUsersByName() { - - TestUser user = new TestUser("JohnDoe2", "john4@doe.com", "JohnDoe2"); - createUser(user); - - // the view is eventually updated - Awaitility.await() - .ignoreExceptions() - .atMost(10, TimeUnit.SECONDS) - .untilAsserted( - () -> { - var byName = getUsersByName(user.name()).getFirst(); - assertThat(byName.name).isEqualTo(user.name()); - }); - } - - @Test - public void verifyFindUsersByEmailAndName() { - - TestUser user = new TestUser("JohnDoe2", "john3@doe.com2", "JohnDoe2"); - createUser(user); - - // the view is eventually updated - Awaitility.await() - .ignoreExceptions() - .atMost(20, TimeUnit.SECONDS) - .untilAsserted( - () -> { - var request = new UsersByEmailAndName.QueryParameters(user.email(), user.name()); - - var byEmail = - await( - componentClient.forView() - .method(UsersByEmailAndName::getUsers) - .invokeAsync(request)); - - assertThat(byEmail.email).isEqualTo(user.email()); - assertThat(byEmail.name).isEqualTo(user.name()); - }); - } - - @Test - public void verifyMultiTableViewForUserCounters() { - - TestUser alice = new TestUser("alice", "alice@foo.com", "Alice Foo"); - TestUser bob = new TestUser("bob", "bob@bar.com", "Bob Bar"); - - createUser(alice); - createUser(bob); - - assignCounter("c1", alice.id()); - assignCounter("c2", bob.id()); - assignCounter("c3", alice.id()); - assignCounter("c4", bob.id()); - - increaseCounter("c1", 11); - increaseCounter("c2", 22); - increaseCounter("c3", 33); - increaseCounter("c4", 44); - - // the view is eventually updated - - Awaitility.await() - .ignoreExceptions() - .atMost(20, TimeUnit.SECONDS) - .until(() -> getUserCounters(alice.id()).counters.size(), new IsEqual<>(2)); - - Awaitility.await() - .ignoreExceptions() - .atMost(20, TimeUnit.SECONDS) - .until(() -> getUserCounters(bob.id()).counters.size(), new IsEqual<>(2)); - - UserCounters aliceCounters = getUserCounters(alice.id()); - assertThat(aliceCounters.id).isEqualTo(alice.id()); - assertThat(aliceCounters.email).isEqualTo(alice.email()); - assertThat(aliceCounters.name).isEqualTo(alice.name()); - assertThat(aliceCounters.counters).containsOnly(new UserCounter("c1", 11), new UserCounter("c3", 33)); - - UserCounters bobCounters = getUserCounters(bob.id()); - - assertThat(bobCounters.id).isEqualTo(bob.id()); - assertThat(bobCounters.email).isEqualTo(bob.email()); - assertThat(bobCounters.name).isEqualTo(bob.name()); - assertThat(bobCounters.counters).containsOnly(new UserCounter("c2", 22), new UserCounter("c4", 44)); - } @Test public void verifyActionWithMetadata() { @@ -577,34 +262,6 @@ public void searchWithInstant() { .until(() -> getCustomersByCreationDate(later).size(), new IsEqual(0)); } - - @NotNull - private List getUsersByName(String name) { - return await( - componentClient.forView() - .method(UsersByName::getUsers) - .invokeAsync(new UsersByName.QueryParameters(name))) - .users(); - } - - @Nullable - private UserWithVersion getUserByEmail(String email) { - return await( - componentClient.forView() - .method(UserWithVersionView::getUser) - .invokeAsync(UserWithVersionView.queryParam(email))); - } - - private void updateUser(TestUser user) { - Ok userUpdate = - await( - componentClient.forKeyValueEntity(user.id()) - .method(UserEntity::createOrUpdateUser) - .invokeAsync(new UserEntity.CreatedUser(user.name(), user.email()))); - - assertThat(userUpdate).isEqualTo(Ok.instance); - } - private void createUser(TestUser user) { Ok userCreation = await( @@ -648,31 +305,6 @@ private void deleteUser(TestUser user) { assertThat(userDeleted).isEqualTo(Ok.instance); } - private void increaseCounter(String id, int value) { - await( - componentClient.forEventSourcedEntity(id) - .method(CounterEntity::increase) - .invokeAsync(value)); - } - - private void multiplyCounter(String id, int value) { - await( - componentClient.forEventSourcedEntity(id) - .method(CounterEntity::times) - .invokeAsync(value)); - } - private void assignCounter(String id, String assignee) { - await( - componentClient.forKeyValueEntity(id) - .method(AssignedCounterEntity::assign) - .invokeAsync(assignee)); - } - - private UserCounters getUserCounters(String userId) { - return await( - componentClient.forView().method(UserCountersView::get) - .invokeAsync(UserCountersView.queryParam(userId))); - } } diff --git a/akka-javasdk-tests/src/test/java/akkajavasdk/ViewIntegrationTest.java b/akka-javasdk-tests/src/test/java/akkajavasdk/ViewIntegrationTest.java new file mode 100644 index 000000000..cd048c731 --- /dev/null +++ b/akka-javasdk-tests/src/test/java/akkajavasdk/ViewIntegrationTest.java @@ -0,0 +1,455 @@ +/* + * Copyright (C) 2021-2024 Lightbend Inc. + */ + +package akkajavasdk; + +import akka.javasdk.client.EventSourcedEntityClient; +import akka.javasdk.client.NoEntryFoundException; +import akka.javasdk.testkit.TestKit; +import akka.javasdk.testkit.TestKitSupport; +import akka.stream.javadsl.Sink; +import akkajavasdk.components.eventsourcedentities.counter.Counter; +import akkajavasdk.components.eventsourcedentities.counter.CounterEntity; +import akkajavasdk.components.keyvalueentities.user.AssignedCounterEntity; +import akkajavasdk.components.keyvalueentities.user.User; +import akkajavasdk.components.keyvalueentities.user.UserEntity; +import akkajavasdk.components.views.AllTheTypesKvEntity; +import akkajavasdk.components.views.AllTheTypesView; +import akkajavasdk.components.views.UserCounter; +import akkajavasdk.components.views.UserCounters; +import akkajavasdk.components.views.UserCountersView; +import akkajavasdk.components.views.counter.CountersByValueSubscriptions; +import akkajavasdk.components.views.counter.CountersByValueWithIgnore; +import akkajavasdk.components.views.hierarchy.HierarchyCountersByValue; +import akkajavasdk.components.views.user.UserWithVersion; +import akkajavasdk.components.views.user.UserWithVersionView; +import akkajavasdk.components.views.user.UsersByEmailAndName; +import akkajavasdk.components.views.user.UsersByName; +import akkajavasdk.components.views.user.UsersByPrimitives; +import akkajavasdk.components.views.user.UsersView; +import org.awaitility.Awaitility; +import org.hamcrest.core.IsEqual; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + + +import static java.time.temporal.ChronoUnit.SECONDS; +import static org.assertj.core.api.Assertions.assertThat; + +@ExtendWith(Junit5LogCapturing.class) +public class ViewIntegrationTest extends TestKitSupport { + + private String newId() { + return UUID.randomUUID().toString(); + } + + @Test + public void verifyTransformedUserViewWiring() { + + var id = newId(); + var email = id + "@example.com"; + var user = new TestUser(id, email, "JohnDoe"); + + createUser(user); + + // the view is eventually updated + Awaitility.await() + .ignoreExceptions() + .atMost(15, TimeUnit.of(SECONDS)) + .until(() -> getUserByEmail(user.email()).version, + new IsEqual(1)); + + updateUser(user.withName("JohnDoeJr")); + + // the view is eventually updated + Awaitility.await() + .ignoreExceptions() + .atMost(15, TimeUnit.of(SECONDS)) + .until(() -> getUserByEmail(user.email()).version, + new IsEqual(2)); + } + + @Test + public void verifyViewIsNotSubscribedToMultiplyAndRouterIgnores() { + + var entityId = newId(); + EventSourcedEntityClient counterClient = componentClient.forEventSourcedEntity(entityId); + await(counterClient.method(CounterEntity::increase).invokeAsync(1)); + await(counterClient.method(CounterEntity::times).invokeAsync(2)); + Integer counterGet = await(counterClient.method(CounterEntity::increase).invokeAsync(1)); + + assertThat(counterGet).isEqualTo(1 * 2 + 1); + + Awaitility.await() + .ignoreExceptions() + .atMost(10, TimeUnit.SECONDS) + .untilAsserted( + () -> { + var byValue = await( + componentClient.forView() + .method(CountersByValueWithIgnore::getCounterByValue) + .invokeAsync(CountersByValueWithIgnore.queryParam(2))); + + assertThat(byValue.value()).isEqualTo(1 + 1); + }); + } + + @Disabled // pending primitive query parameters working + @Test + public void verifyHierarchyView() { + + var emptyCounter = await( + componentClient.forView() + .method(HierarchyCountersByValue::getCounterByValue) + .invokeAsync(201)); + + assertThat(emptyCounter).isEmpty(); + + var esId = newId(); + await( + componentClient.forEventSourcedEntity(esId) + .method(CounterEntity::increase) + .invokeAsync(201)); + + + // the view is eventually updated + Awaitility.await() + .ignoreExceptions() + .atMost(15, TimeUnit.of(SECONDS)) + .untilAsserted( + () -> { + var byValue = await( + componentClient.forView() + .method(HierarchyCountersByValue::getCounterByValue) + .invokeAsync(201)); + + assertThat(byValue).hasValue(new Counter(201)); + }); + } + + @Test + public void verifyCounterViewMultipleSubscriptions() { + + var id1 = newId(); + await( + componentClient.forEventSourcedEntity(id1) + .method(CounterEntity::increase) + .invokeAsync(74)); + + var id2 = newId(); + await( + componentClient.forEventSourcedEntity(id2) + .method(CounterEntity::increase) + .invokeAsync(74)); + + Awaitility.await() + .ignoreExceptions() + .atMost(20, TimeUnit.SECONDS) + .until( + () -> + await(componentClient.forView() + .method(CountersByValueSubscriptions::getCounterByValue) + .invokeAsync(new CountersByValueSubscriptions.QueryParameters(74))) + .counters().size(), + new IsEqual<>(2)); + } + + @Test + public void verifyAllTheFieldTypesView() throws Exception { + // see that we can persist and read a row with all fields, no indexed columns + var id = newId(); + var row = new AllTheTypesKvEntity.AllTheTypes(1, 2L, 3F, 4D, true, "text", 5, 6L, 7F, 8D, false, Instant.EPOCH, Optional.of("optional"), List.of("text1", "text2"), + new AllTheTypesKvEntity.ByEmail("test@example.com"), + AllTheTypesKvEntity.AnEnum.THREE, new AllTheTypesKvEntity.Recursive(new AllTheTypesKvEntity.Recursive(null))); + await(componentClient.forKeyValueEntity(id).method(AllTheTypesKvEntity::store).invokeAsync(row)); + + + Awaitility.await() + .ignoreExceptions() + .atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> { + var rows = await(componentClient.forView() + .stream(AllTheTypesView::allRows) + .source().runWith(Sink.seq(), testKit.getMaterializer())); + + assertThat(rows).hasSize(1); + } + ); + + + } + + @Disabled // pending primitive query parameters working + @Test + public void shouldAcceptPrimitivesForViewQueries() { + + TestUser user1 = new TestUser(newId(), "john654321@doe.com", "Bob2"); + TestUser user2 = new TestUser(newId(), "john7654321@doe.com", "Bob3"); + createUser(user1); + createUser(user2); + + Awaitility.await() + .ignoreExceptions() + .atMost(10, TimeUnit.of(SECONDS)) + .untilAsserted(() -> { + var resultByString = await( + componentClient.forView() + .method(UsersByPrimitives::getUserByString) + .invokeAsync(user1.email())); + assertThat(resultByString.users()).isNotEmpty(); + + var resultByInt = await( + componentClient.forView() + .method(UsersByPrimitives::getUserByInt) + .invokeAsync(123)); + assertThat(resultByInt.users()).isNotEmpty(); + + var resultByLong = await( + componentClient.forView() + .method(UsersByPrimitives::getUserByLong) + .invokeAsync(321l)); + assertThat(resultByLong.users()).isNotEmpty(); + + var resultByDouble = await( + componentClient.forView() + .method(UsersByPrimitives::getUserByDouble) + .invokeAsync(12.3d)); + assertThat(resultByDouble.users()).isNotEmpty(); + + var resultByBoolean = await( + componentClient.forView() + .method(UsersByPrimitives::getUserByBoolean) + .invokeAsync(true)); + assertThat(resultByBoolean.users()).isNotEmpty(); + + var resultByEmails = await( + componentClient.forView() + .method(UsersByPrimitives::getUserByEmails) + .invokeAsync(List.of(user1.email(), user2.email()))); + assertThat(resultByEmails.users()).hasSize(2); + }); + } + + + @Test + public void shouldDeleteValueEntityAndDeleteViewsState() { + + TestUser user = new TestUser(newId(), "john123@doe.com", "Bob123"); + createUser(user); + + Awaitility.await() + .ignoreExceptions() + .atMost(15, TimeUnit.of(SECONDS)) + .until(() -> getUserByEmail(user.email()).version, + new IsEqual(1)); + + Awaitility.await() + .ignoreExceptions() + .atMost(15, TimeUnit.of(SECONDS)) + .until(() -> getUsersByName(user.name()).size(), + new IsEqual(1)); + + deleteUser(user); + + Awaitility.await() + .atMost(15, TimeUnit.of(SECONDS)) + .ignoreExceptions() + .untilAsserted( + () -> { + var ex = + failed( + componentClient.forView() + .method(UserWithVersionView::getUser) + .invokeAsync(UserWithVersionView.queryParam(user.email()))); + assertThat(ex).isInstanceOf(NoEntryFoundException.class); + }); + + Awaitility.await() + .ignoreExceptions() + .atMost(15, TimeUnit.of(SECONDS)) + .until(() -> getUsersByName(user.name()).size(), + new IsEqual(0)); + } + + @Test + public void verifyFindUsersByEmailView() { + + TestUser user = new TestUser(newId(), "john3@doe.com", "JohnDoe"); + createUser(user); + + // the view is eventually updated + Awaitility.await() + .ignoreExceptions() + .atMost(10, TimeUnit.SECONDS) + .untilAsserted( + () -> { + var byEmail = await( + componentClient.forView() + .method(UsersView::getUserByEmail) + .invokeAsync(UsersView.byEmailParam(user.email()))); + + assertThat(byEmail.email).isEqualTo(user.email()); + }); + } + + @Test + public void verifyFindUsersByNameView() { + + TestUser user = new TestUser(newId(), "john4@doe.com", "JohnDoe2"); + createUser(user); + + // the view is eventually updated + Awaitility.await() + .ignoreExceptions() + .atMost(10, TimeUnit.SECONDS) + .untilAsserted( + () -> { + var byName = getUsersByName(user.name()).getFirst(); + assertThat(byName.name).isEqualTo(user.name()); + }); + } + + @Test + public void verifyFindUsersByEmailAndNameView() { + + TestUser user = new TestUser(newId(), "john3@doe.com2", "JohnDoe2"); + createUser(user); + + // the view is eventually updated + Awaitility.await() + .ignoreExceptions() + .atMost(20, TimeUnit.SECONDS) + .untilAsserted( + () -> { + var request = new UsersByEmailAndName.QueryParameters(user.email(), user.name()); + + var byEmail = + await( + componentClient.forView() + .method(UsersByEmailAndName::getUsers) + .invokeAsync(request)); + + assertThat(byEmail.email).isEqualTo(user.email()); + assertThat(byEmail.name).isEqualTo(user.name()); + }); + } + + @Test + public void verifyMultiTableViewForUserCounters() { + + TestUser alice = new TestUser(newId(), "alice@foo.com", "Alice Foo"); + TestUser bob = new TestUser(newId(), "bob@bar.com", "Bob Bar"); + + createUser(alice); + createUser(bob); + + assignCounter("c1", alice.id()); + assignCounter("c2", bob.id()); + assignCounter("c3", alice.id()); + assignCounter("c4", bob.id()); + + increaseCounter("c1", 11); + increaseCounter("c2", 22); + increaseCounter("c3", 33); + increaseCounter("c4", 44); + + // the view is eventually updated + + Awaitility.await() + .ignoreExceptions() + .atMost(20, TimeUnit.SECONDS) + .until(() -> getUserCounters(alice.id()).counters.size(), new IsEqual<>(2)); + + Awaitility.await() + .ignoreExceptions() + .atMost(20, TimeUnit.SECONDS) + .until(() -> getUserCounters(bob.id()).counters.size(), new IsEqual<>(2)); + + UserCounters aliceCounters = getUserCounters(alice.id()); + assertThat(aliceCounters.id).isEqualTo(alice.id()); + assertThat(aliceCounters.email).isEqualTo(alice.email()); + assertThat(aliceCounters.name).isEqualTo(alice.name()); + assertThat(aliceCounters.counters).containsOnly(new UserCounter("c1", 11), new UserCounter("c3", 33)); + + UserCounters bobCounters = getUserCounters(bob.id()); + + assertThat(bobCounters.id).isEqualTo(bob.id()); + assertThat(bobCounters.email).isEqualTo(bob.email()); + assertThat(bobCounters.name).isEqualTo(bob.name()); + assertThat(bobCounters.counters).containsOnly(new UserCounter("c2", 22), new UserCounter("c4", 44)); + } + + private void createUser(TestUser user) { + Ok userCreation = + await( + componentClient.forKeyValueEntity(user.id()) + .method(UserEntity::createOrUpdateUser) + .invokeAsync(new UserEntity.CreatedUser(user.name(), user.email()))); + assertThat(userCreation).isEqualTo(Ok.instance); + } + + private void updateUser(TestUser user) { + Ok userUpdate = + await( + componentClient.forKeyValueEntity(user.id()) + .method(UserEntity::createOrUpdateUser) + .invokeAsync(new UserEntity.CreatedUser(user.name(), user.email()))); + + assertThat(userUpdate).isEqualTo(Ok.instance); + } + + private UserWithVersion getUserByEmail(String email) { + return await( + componentClient.forView() + .method(UserWithVersionView::getUser) + .invokeAsync(UserWithVersionView.queryParam(email))); + } + + private void increaseCounter(String id, int value) { + await( + componentClient.forEventSourcedEntity(id) + .method(CounterEntity::increase) + .invokeAsync(value)); + } + + private void assignCounter(String id, String assignee) { + await( + componentClient.forKeyValueEntity(id) + .method(AssignedCounterEntity::assign) + .invokeAsync(assignee)); + } + + private UserCounters getUserCounters(String userId) { + return await( + componentClient.forView().method(UserCountersView::get) + .invokeAsync(UserCountersView.queryParam(userId))); + } + + + private List getUsersByName(String name) { + return await( + componentClient.forView() + .method(UsersByName::getUsers) + .invokeAsync(new UsersByName.QueryParameters(name))) + .users(); + } + + private void deleteUser(TestUser user) { + Ok userDeleted = + await( + componentClient + .forKeyValueEntity(user.id()) + .method(UserEntity::deleteUser) + .invokeAsync(new UserEntity.Delete())); + assertThat(userDeleted).isEqualTo(Ok.instance); + } + +} diff --git a/akka-javasdk-tests/src/test/java/akkajavasdk/components/views/AllTheTypesKvEntity.java b/akka-javasdk-tests/src/test/java/akkajavasdk/components/views/AllTheTypesKvEntity.java new file mode 100644 index 000000000..9af3dc2e4 --- /dev/null +++ b/akka-javasdk-tests/src/test/java/akkajavasdk/components/views/AllTheTypesKvEntity.java @@ -0,0 +1,53 @@ +/* + * Copyright (C) 2021-2024 Lightbend Inc. + */ + +package akkajavasdk.components.views; + +import akka.javasdk.annotations.ComponentId; +import akka.javasdk.keyvalueentity.KeyValueEntity; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; + +@ComponentId("all-the-types-kve") +public class AllTheTypesKvEntity extends KeyValueEntity { + + public enum AnEnum { + ONE, TWO, THREE + } + + // common query parameter for views in this file + public record ByEmail(String email) { + } + + public record Recursive(Recursive recurse) {} + + public record AllTheTypes( + int intValue, + long longValue, + float floatValue, + double doubleValue, + boolean booleanValue, + String stringValue, + Integer wrappedInt, + Long wrappedLong, + Float wrappedFloat, + Double wrappedDouble, + Boolean wrappedBoolean, + Instant instant, + // FIXME bytes does not work yet in runtime Byte[] bytes, + Optional optionalString, + List repeatedString, + ByEmail nestedMessage, + AnEnum anEnum, + Recursive recursive + ) {} + + + + public Effect store(AllTheTypes value) { + return effects().updateState(value).thenReply("OK"); + } +} diff --git a/akka-javasdk-tests/src/test/java/akkajavasdk/components/views/AllTheTypesView.java b/akka-javasdk-tests/src/test/java/akkajavasdk/components/views/AllTheTypesView.java new file mode 100644 index 000000000..08426a9c0 --- /dev/null +++ b/akka-javasdk-tests/src/test/java/akkajavasdk/components/views/AllTheTypesView.java @@ -0,0 +1,25 @@ +/* + * Copyright (C) 2021-2024 Lightbend Inc. + */ + +package akkajavasdk.components.views; + +import akka.javasdk.annotations.ComponentId; +import akka.javasdk.annotations.Consume; +import akka.javasdk.annotations.Query; +import akka.javasdk.view.TableUpdater; +import akka.javasdk.view.View; + +@ComponentId("all_the_field_types_view") +public class AllTheTypesView extends View { + + + @Consume.FromKeyValueEntity(AllTheTypesKvEntity.class) + public static class Events extends TableUpdater { } + + @Query("SELECT * FROM events") + public QueryStreamEffect allRows() { + return queryStreamResult(); + } + +} diff --git a/akka-javasdk-tests/src/test/resources/META-INF/akka-javasdk-components.conf b/akka-javasdk-tests/src/test/resources/META-INF/akka-javasdk-components.conf index 7e99f3633..990278add 100644 --- a/akka-javasdk-tests/src/test/resources/META-INF/akka-javasdk-components.conf +++ b/akka-javasdk-tests/src/test/resources/META-INF/akka-javasdk-components.conf @@ -30,7 +30,8 @@ akka.javasdk { "akkajavasdk.components.keyvalueentities.user.UserEntity", "akkajavasdk.components.workflowentities.WalletEntity", "akkajavasdk.components.keyvalueentities.user.AssignedCounterEntity", - "akkajavasdk.components.keyvalueentities.hierarchy.TextKvEntity" + "akkajavasdk.components.keyvalueentities.hierarchy.TextKvEntity", + "akkajavasdk.components.views.AllTheTypesKvEntity" ] view = [ "akkajavasdk.components.views.user.UsersByEmailAndName", @@ -44,7 +45,8 @@ akka.javasdk { "akkajavasdk.components.views.counter.CountersByValueSubscriptions", "akkajavasdk.components.pubsub.ViewFromCounterEventsTopic", "akkajavasdk.components.views.user.UsersByPrimitives", - "akkajavasdk.components.views.hierarchy.HierarchyCountersByValue" + "akkajavasdk.components.views.hierarchy.HierarchyCountersByValue", + "akkajavasdk.components.views.AllTheTypesView" ] workflow = [ "akkajavasdk.components.workflowentities.TransferWorkflow", diff --git a/akka-javasdk/src/main/scala/akka/javasdk/impl/view/ViewDescriptorFactory.scala b/akka-javasdk/src/main/scala/akka/javasdk/impl/view/ViewDescriptorFactory.scala index 00bc6e3f6..1b4f36797 100644 --- a/akka-javasdk/src/main/scala/akka/javasdk/impl/view/ViewDescriptorFactory.scala +++ b/akka-javasdk/src/main/scala/akka/javasdk/impl/view/ViewDescriptorFactory.scala @@ -81,7 +81,9 @@ private[impl] object ViewDescriptorFactory { tableUpdaterClass.getAnnotation(classOf[Table]).value() } else { // figure out from first query - val query = allQueryStrings.head + val query = allQueryStrings.headOption.getOrElse( + throw new IllegalArgumentException( + s"View [$componentId] does not have any queries defined, must have at least one query")) TableNamePattern .findFirstMatchIn(query) .map(_.group(1)) diff --git a/akka-javasdk/src/main/scala/akka/javasdk/impl/view/ViewSchema.scala b/akka-javasdk/src/main/scala/akka/javasdk/impl/view/ViewSchema.scala index a9f53d64b..10214e14c 100644 --- a/akka-javasdk/src/main/scala/akka/javasdk/impl/view/ViewSchema.scala +++ b/akka-javasdk/src/main/scala/akka/javasdk/impl/view/ViewSchema.scala @@ -48,40 +48,49 @@ private[view] object ViewSchema { classOf[String] -> SpiString, classOf[java.time.Instant] -> SpiTimestamp) - def apply(javaType: Type): SpiType = - typeNameMap.get(javaType.getTypeName) match { - case Some(found) => found - case None => - val clazz = javaType match { - case c: Class[_] => c - case p: ParameterizedType => p.getRawType.asInstanceOf[Class[_]] - } - knownConcreteClasses.get(clazz) match { - case Some(found) => found - case None => - // trickier ones where we have to look at type parameters etc - if (clazz.isArray && clazz.componentType() == classOf[java.lang.Byte]) { - SpiByteString - } else if (clazz.isEnum) { - new SpiType.SpiEnum(clazz.getName) - } else { - javaType match { - case p: ParameterizedType if clazz == classOf[Optional[_]] => - new SpiType.SpiOptional(apply(p.getActualTypeArguments.head).asInstanceOf[SpiNestableType]) - case p: ParameterizedType if classOf[java.util.Collection[_]].isAssignableFrom(clazz) => - new SpiType.SpiList(apply(p.getActualTypeArguments.head).asInstanceOf[SpiNestableType]) - case _: Class[_] => - new SpiType.SpiClass( - clazz.getName, - clazz.getDeclaredFields - .filterNot(f => f.accessFlags().contains(AccessFlag.STATIC)) - // FIXME recursive classes with fields of their own type - .filterNot(_.getType == clazz) - .map(field => new SpiType.SpiField(field.getName, apply(field.getGenericType))) - .toSeq) - } + def apply(rootType: Type): SpiType = { + // Note: not tail recursive but trees should not ever be deep enough that it is a problem + def loop(currentType: Type, seenClasses: Set[Class[_]]): SpiType = + typeNameMap.get(currentType.getTypeName) match { + case Some(found) => found + case None => + val clazz = currentType match { + case c: Class[_] => c + case p: ParameterizedType => p.getRawType.asInstanceOf[Class[_]] + } + if (seenClasses.contains(clazz)) new SpiType.SpiClassRef(clazz.getName) + else + knownConcreteClasses.get(clazz) match { + case Some(found) => found + case None => + // trickier ones where we have to look at type parameters etc + if (clazz.isArray && clazz.componentType() == classOf[java.lang.Byte]) { + SpiByteString + } else if (clazz.isEnum) { + new SpiType.SpiEnum(clazz.getName) + } else { + currentType match { + case p: ParameterizedType if clazz == classOf[Optional[_]] => + new SpiType.SpiOptional( + loop(p.getActualTypeArguments.head, seenClasses).asInstanceOf[SpiNestableType]) + case p: ParameterizedType if classOf[java.util.Collection[_]].isAssignableFrom(clazz) => + new SpiType.SpiList( + loop(p.getActualTypeArguments.head, seenClasses).asInstanceOf[SpiNestableType]) + case _: Class[_] => + val seenIncludingThis = seenClasses + clazz + new SpiType.SpiClass( + clazz.getName, + clazz.getDeclaredFields + .filterNot(f => f.accessFlags().contains(AccessFlag.STATIC)) + .map(field => + new SpiType.SpiField(field.getName, loop(field.getGenericType, seenIncludingThis))) + .toSeq) + } + } } - } - } + } + + loop(rootType, Set.empty) + } } diff --git a/akka-javasdk/src/test/java/akka/javasdk/testmodels/view/ViewTestModels.java b/akka-javasdk/src/test/java/akka/javasdk/testmodels/view/ViewTestModels.java index 3031e34d8..d54718490 100644 --- a/akka-javasdk/src/test/java/akka/javasdk/testmodels/view/ViewTestModels.java +++ b/akka-javasdk/src/test/java/akka/javasdk/testmodels/view/ViewTestModels.java @@ -23,7 +23,6 @@ import akka.javasdk.testmodels.keyvalueentity.TimeTrackerEntity; import akka.javasdk.testmodels.keyvalueentity.User; import akka.javasdk.testmodels.keyvalueentity.UserEntity; -import akka.util.ByteString; import java.time.Instant; import java.util.List; @@ -47,13 +46,21 @@ public record EveryType( Byte[] bytes, Optional optionalString, List repeatedString, - ByEmail nestedMessage + ByEmail nestedMessage, + AnEnum anEnum ) {} + public enum AnEnum { + ONE, TWO, THREE + } + // common query parameter for views in this file public record ByEmail(String email) { } + public record Recursive(String id, Recursive child) {} + public record TwoStepRecursive(TwoStepRecursiveChild child) {} + public record TwoStepRecursiveChild(TwoStepRecursive recursive) {} @ComponentId("users_view") public static class UserByEmailWithGet extends View { @@ -720,4 +727,29 @@ public QueryEffect getEmployeeByEmail(ByEmail byEmail) { return queryResult(); } } + + + public record ById(String id) {} + + @ComponentId("recursive_view") + public static class RecursiveViewStateView extends View { + @Consume.FromTopic(value = "recursivetopic") + public static class Events extends TableUpdater { } + + @Query("SELECT * FROM events WHERE id = :id") + public QueryEffect getEmployeeByEmail(ById id) { + return queryResult(); + } + } + + @ComponentId("all_the_field_types_view") + public static class AllTheFieldTypesView extends View { + @Consume.FromTopic(value = "allthetypestopic") + public static class Events extends TableUpdater { } + + @Query("SELECT * FROM rows") + public QueryStreamEffect allRows() { + return queryStreamResult(); + } + } } diff --git a/akka-javasdk/src/test/scala/akka/javasdk/impl/view/ViewDescriptorFactorySpec.scala b/akka-javasdk/src/test/scala/akka/javasdk/impl/view/ViewDescriptorFactorySpec.scala index 32767f069..2f47b7d20 100644 --- a/akka-javasdk/src/test/scala/akka/javasdk/impl/view/ViewDescriptorFactorySpec.scala +++ b/akka-javasdk/src/test/scala/akka/javasdk/impl/view/ViewDescriptorFactorySpec.scala @@ -483,5 +483,18 @@ class ViewDescriptorFactorySpec extends AnyWordSpec with Matchers { table.updateHandler shouldBe defined } } + + "create a descriptor for a view with a recursive table type" in { + assertDescriptor[RecursiveViewStateView] { desc => + // just check that it parses + } + } + + "create a descriptor for a view with a table type with all possible column types" in { + assertDescriptor[AllTheFieldTypesView] { desc => + // just check that it parses + } + } + } } diff --git a/akka-javasdk/src/test/scala/akka/javasdk/impl/view/ViewSchemaSpec.scala b/akka-javasdk/src/test/scala/akka/javasdk/impl/view/ViewSchemaSpec.scala index fc4e91608..10f0f2c34 100644 --- a/akka-javasdk/src/test/scala/akka/javasdk/impl/view/ViewSchemaSpec.scala +++ b/akka-javasdk/src/test/scala/akka/javasdk/impl/view/ViewSchemaSpec.scala @@ -8,7 +8,9 @@ import akka.javasdk.testmodels.view.ViewTestModels import akka.runtime.sdk.spi.views.SpiType.SpiBoolean import akka.runtime.sdk.spi.views.SpiType.SpiByteString import akka.runtime.sdk.spi.views.SpiType.SpiClass +import akka.runtime.sdk.spi.views.SpiType.SpiClassRef import akka.runtime.sdk.spi.views.SpiType.SpiDouble +import akka.runtime.sdk.spi.views.SpiType.SpiEnum import akka.runtime.sdk.spi.views.SpiType.SpiField import akka.runtime.sdk.spi.views.SpiType.SpiFloat import akka.runtime.sdk.spi.views.SpiType.SpiInteger @@ -47,8 +49,9 @@ class ViewSchemaSpec extends AnyWordSpec with Matchers { "optionalString" -> new SpiOptional(SpiString), "repeatedString" -> new SpiList(SpiString), "nestedMessage" -> new SpiClass( - "akka.javasdk.testmodels.view.ViewTestModels$ByEmail", - Seq(new SpiField("email", SpiString)))) + classOf[ViewTestModels.ByEmail].getName, + Seq(new SpiField("email", SpiString))), + "anEnum" -> new SpiEnum(classOf[ViewTestModels.AnEnum].getName)) clazz.fields should have size expectedFields.size expectedFields.foreach { case (name, expectedType) => @@ -59,7 +62,26 @@ class ViewSchemaSpec extends AnyWordSpec with Matchers { } } - // FIXME self-referencing/recursive types + "handle self referencing type trees" in { + val result = ViewSchema(classOf[ViewTestModels.Recursive]) + result shouldBe a[SpiClass] + result.asInstanceOf[SpiClass].getField("child").get.fieldType shouldBe new SpiClassRef( + classOf[ViewTestModels.Recursive].getName) + } + + "handle self referencing type trees with longer cycles" in { + val result = ViewSchema(classOf[ViewTestModels.TwoStepRecursive]) + result shouldBe a[SpiClass] + result + .asInstanceOf[SpiClass] + .getField("child") + .get + .fieldType + .asInstanceOf[SpiClass] + .getField("recursive") + .get + .fieldType shouldBe new SpiClassRef(classOf[ViewTestModels.TwoStepRecursive].getName) + } } }