@Test @DisplayName("trigger subscription callbacks matching by ID") void triggerSubscriptionsMatchingById() { CustomerAggregateRepository repository = new CustomerAggregateRepository(); Stand stand = createStand(repository); Collection<Customer> sampleCustomers = fillSampleCustomers(10); Topic someCustomers = requestFactory.topic() .select(Customer.class) .byId(ids(sampleCustomers)) .build(); Collection<Customer> callbackStates = newHashSet(); MemoizeSubscriptionCallback callback = new MemoizeSubscriptionCallback() { @Override public void accept(SubscriptionUpdate update) { super.accept(update); Customer customerInCallback = (Customer) update.state(0); callbackStates.add(customerInCallback); } }; subscribeAndActivate(stand, someCustomers, callback); for (Customer customer : sampleCustomers) { CustomerId customerId = customer.getId(); int version = 1; CustomerAggregate entity = aggregateOfClass(CustomerAggregate.class) .withId(customerId) .withState(customer) .withVersion(version) .build(); stand.post(entity, repository.lifecycleOf(customerId)); } assertEquals(newHashSet(sampleCustomers), callbackStates); }
@VisibleForTesting void post(Entity<?, ?> entity, EntityLifecycle lifecycle) { Any id = Identifier.pack(entity.id()); Any state = AnyPacker.pack(entity.state()); EntityRecord record = EntityRecord .newBuilder() .setEntityId(id) .setState(state) .vBuild(); EntityRecordChange change = EntityRecordChange .newBuilder() .setNewValue(record) .vBuild(); MessageId origin = Identity.byString("Stand-received-entity-update"); lifecycle.onStateChanged(change, ImmutableSet.of(origin), Origin.getDefaultInstance()); }
Stand implements AutoCloseable { @VisibleForTesting void post(Entity<?, ?> entity, EntityLifecycle lifecycle) { Any id = Identifier.pack(entity.id()); Any state = AnyPacker.pack(entity.state()); EntityRecord record = EntityRecord .newBuilder() .setEntityId(id) .setState(state) .vBuild(); EntityRecordChange change = EntityRecordChange .newBuilder() .setNewValue(record) .vBuild(); MessageId origin = Identity.byString("Stand-received-entity-update"); lifecycle.onStateChanged(change, ImmutableSet.of(origin), Origin.getDefaultInstance()); } }
Stand implements AutoCloseable { @VisibleForTesting void post(Entity<?, ?> entity, EntityLifecycle lifecycle) { Any id = Identifier.pack(entity.id()); Any state = AnyPacker.pack(entity.state()); EntityRecord record = EntityRecord .newBuilder() .setEntityId(id) .setState(state) .vBuild(); EntityRecordChange change = EntityRecordChange .newBuilder() .setNewValue(record) .vBuild(); MessageId origin = Identity.byString("Stand-received-entity-update"); lifecycle.onStateChanged(change, ImmutableSet.of(origin), Origin.getDefaultInstance()); } private  Stand(Builder builder); }
Stand implements AutoCloseable { @VisibleForTesting void post(Entity<?, ?> entity, EntityLifecycle lifecycle) { Any id = Identifier.pack(entity.id()); Any state = AnyPacker.pack(entity.state()); EntityRecord record = EntityRecord .newBuilder() .setEntityId(id) .setState(state) .vBuild(); EntityRecordChange change = EntityRecordChange .newBuilder() .setNewValue(record) .vBuild(); MessageId origin = Identity.byString("Stand-received-entity-update"); lifecycle.onStateChanged(change, ImmutableSet.of(origin), Origin.getDefaultInstance()); } private  Stand(Builder builder); static Builder newBuilder(); @Internal @VisibleForTesting boolean isMultitenant(); void subscribe(Topic topic, StreamObserver<Subscription> responseObserver); void subscribe(Subscription subscription); void activate(Subscription subscription,
                         SubscriptionCallback callback,
                         StreamObserver<Response> responseObserver); void cancel(Subscription subscription, StreamObserver<Response> responseObserver); Listener<EventEnvelope> eventListener(); ImmutableSet<TypeUrl> exposedTypes(); ImmutableSet<TypeUrl> exposedEventTypes(); ImmutableSet<TypeUrl> exposedAggregateTypes(); void execute(Query query, StreamObserver<QueryResponse> responseObserver); void registerTypeSupplier(Repository<?, ?> repository); @Override void close(); }
Stand implements AutoCloseable { @VisibleForTesting void post(Entity<?, ?> entity, EntityLifecycle lifecycle) { Any id = Identifier.pack(entity.id()); Any state = AnyPacker.pack(entity.state()); EntityRecord record = EntityRecord .newBuilder() .setEntityId(id) .setState(state) .vBuild(); EntityRecordChange change = EntityRecordChange .newBuilder() .setNewValue(record) .vBuild(); MessageId origin = Identity.byString("Stand-received-entity-update"); lifecycle.onStateChanged(change, ImmutableSet.of(origin), Origin.getDefaultInstance()); } private  Stand(Builder builder); static Builder newBuilder(); @Internal @VisibleForTesting boolean isMultitenant(); void subscribe(Topic topic, StreamObserver<Subscription> responseObserver); void subscribe(Subscription subscription); void activate(Subscription subscription,
                         SubscriptionCallback callback,
                         StreamObserver<Response> responseObserver); void cancel(Subscription subscription, StreamObserver<Response> responseObserver); Listener<EventEnvelope> eventListener(); ImmutableSet<TypeUrl> exposedTypes(); ImmutableSet<TypeUrl> exposedEventTypes(); ImmutableSet<TypeUrl> exposedAggregateTypes(); void execute(Query query, StreamObserver<QueryResponse> responseObserver); void registerTypeSupplier(Repository<?, ?> repository); @Override void close();  }