-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
GROUP-89 Updated public update streams to fetch, maintain, and concat…
… initial state
- Loading branch information
Showing
34 changed files
with
1,322 additions
and
78 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
54 changes: 54 additions & 0 deletions
54
src/main/java/org/grouphq/groupsync/group/sync/state/DormantState.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| package org.grouphq.groupsync.group.sync.state; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.concurrent.atomic.AtomicReference; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.grouphq.groupsync.config.ClientProperties; | ||
| import reactor.core.publisher.Mono; | ||
| import reactor.core.publisher.SignalType; | ||
|
|
||
| /** | ||
| * State representing when {@link GroupInitialStateService} is waiting for a client to trigger the initialization. | ||
| */ | ||
| @Slf4j | ||
| public class DormantState extends State { | ||
|
|
||
| private final ClientProperties clientProperties; | ||
|
|
||
| private final AtomicReference<Mono<Void>> initialRequest = new AtomicReference<>(); | ||
|
|
||
| public DormantState(GroupInitialStateService groupInitialStateService, ClientProperties clientProperties) { | ||
| super(groupInitialStateService); | ||
| this.clientProperties = clientProperties; | ||
| } | ||
|
|
||
| /** | ||
| * Triggers the initialization of the current state of groups and their members. | ||
| * Note that this method will only trigger the initialization once, subsequent calls will return the same request. | ||
| * This is done by caching the mono, making it a hot source. | ||
| * | ||
| * @return a Mono that will initialize the current state of groups and their members | ||
| */ | ||
| @Override | ||
| public Mono<Void> onRequest() { | ||
| initialRequest.compareAndSet(null, | ||
| groupInitialStateService.initializeGroupState() | ||
| .timeout(Duration.ofMillis(clientProperties.getGroupsTimeoutMilliseconds())) | ||
| .doOnError(throwable -> log.error("Error initializing group state", throwable)) | ||
| .doOnSuccess(empty -> log.info("Successfully initialized group state")) | ||
| .doFinally(signalType -> { | ||
| if (signalType == SignalType.ON_COMPLETE) { | ||
| groupInitialStateService.setState(new ReadyState(groupInitialStateService)); | ||
| } else if (signalType == SignalType.ON_ERROR) { | ||
| groupInitialStateService.setState( | ||
| new DormantState(groupInitialStateService, clientProperties)); | ||
| } | ||
| }).cache() | ||
| ); | ||
|
|
||
| final Mono<Void> cachedRequest = initialRequest.get(); | ||
|
|
||
| groupInitialStateService.setState(new LoadingState(groupInitialStateService, cachedRequest)); | ||
| return cachedRequest; | ||
| } | ||
| } |
117 changes: 117 additions & 0 deletions
117
src/main/java/org/grouphq/groupsync/group/sync/state/GroupInitialStateService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| package org.grouphq.groupsync.group.sync.state; | ||
|
|
||
| import jakarta.annotation.PreDestroy; | ||
| import java.time.Duration; | ||
| import java.util.HashSet; | ||
| import lombok.AccessLevel; | ||
| import lombok.Getter; | ||
| import lombok.Setter; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.grouphq.groupsync.config.ClientProperties; | ||
| import org.grouphq.groupsync.group.domain.PublicOutboxEvent; | ||
| import org.grouphq.groupsync.group.sync.GroupFetchService; | ||
| import org.grouphq.groupsync.group.sync.GroupUpdateService; | ||
| import org.grouphq.groupsync.groupservice.domain.groups.Group; | ||
| import org.springframework.stereotype.Service; | ||
| import reactor.core.Disposable; | ||
| import reactor.core.publisher.Flux; | ||
| import reactor.core.publisher.Mono; | ||
| import reactor.core.scheduler.Schedulers; | ||
| import reactor.util.retry.Retry; | ||
|
|
||
| /** | ||
| * Maintains the current relevant events to enable clients | ||
| * to build the current state of groups and their members. | ||
| * Relies on Java's Standard Library Concurrency Utilities | ||
| */ | ||
| @Service | ||
| @Slf4j | ||
| public class GroupInitialStateService { | ||
|
|
||
| private final GroupFetchService groupFetchService; | ||
| private final GroupUpdateService groupUpdateService; | ||
| private final GroupStateService groupStateService; | ||
| private final ClientProperties clientProperties; | ||
|
|
||
| @Getter | ||
| @Setter(AccessLevel.PACKAGE) | ||
| private volatile State state; | ||
|
|
||
| @Getter(AccessLevel.NONE) | ||
| @Setter(AccessLevel.NONE) | ||
| private transient Disposable updateSubscription; | ||
|
|
||
|
|
||
| public GroupInitialStateService(GroupFetchService groupFetchService, GroupUpdateService groupUpdateService, | ||
| GroupStateService groupStateService, ClientProperties clientProperties) { | ||
| this.groupFetchService = groupFetchService; | ||
| this.groupUpdateService = groupUpdateService; | ||
| this.groupStateService = groupStateService; | ||
| this.clientProperties = clientProperties; | ||
| setState(new DormantState(this, clientProperties)); | ||
| } | ||
|
|
||
| public Flux<PublicOutboxEvent> requestCurrentEvents() { | ||
| return state.onRequest() | ||
| .thenMany(buildState()); | ||
| } | ||
|
|
||
| private Flux<PublicOutboxEvent> buildState() { | ||
| return groupStateService.getCurrentGroupEvents() | ||
| .filter(event -> event.eventData() instanceof Group) | ||
| .flatMap(event -> { | ||
| final Group group = (Group) event.eventData(); | ||
| return groupStateService.getMembersForGroup(group.id()) | ||
| .collectList() | ||
| .map(publicMembers -> { | ||
| final Group groupWithMembers = group.withMembers(new HashSet<>(publicMembers)); | ||
| return event.withNewEventData(groupWithMembers); | ||
| }); | ||
| }); | ||
| } | ||
|
|
||
| protected Mono<Void> initializeGroupState() { | ||
| return groupStateService.resetState().then( | ||
| groupFetchService.getGroupsAsEvents() | ||
| .flatMap(groupStateService::saveGroupEvent) | ||
| .retryWhen( | ||
| Retry.backoff(clientProperties.getGroupsRetryAttempts(), | ||
| Duration.ofMillis(clientProperties.getGroupsRetryBackoffMilliseconds())) | ||
| .maxBackoff(Duration.ofSeconds(10)) | ||
| .doBeforeRetry(retrySignal -> log.warn("Retrying due to error", retrySignal.failure()))) | ||
| .doOnComplete(this::createUpdateSubscription) | ||
| .doOnError(error -> log.error("Error getting initial state of events", error)) | ||
| .then(Mono.empty()) | ||
| ); | ||
| } | ||
|
|
||
| private void createUpdateSubscription() { | ||
| disposeUpdateSubscription(); | ||
|
|
||
| updateSubscription = keepGroupsAndMembersUpToDate() | ||
| .subscribeOn(Schedulers.boundedElastic()) | ||
| .subscribe(); | ||
| } | ||
|
|
||
| private void disposeUpdateSubscription() { | ||
| if (updateSubscription != null && !updateSubscription.isDisposed()) { | ||
| updateSubscription.dispose(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Methods annotated with PreDestroy are called by the Spring framework before destroying the service bean. | ||
| */ | ||
| @PreDestroy | ||
| public void cleanUp() { | ||
| disposeUpdateSubscription(); | ||
| } | ||
|
|
||
| protected Flux<PublicOutboxEvent> keepGroupsAndMembersUpToDate() { | ||
| return groupUpdateService.publicUpdatesStream() | ||
| .flatMap(groupStateService::handleEventUpdate) | ||
| .doOnError(throwable -> log.error("Error keeping groups and members up to date", throwable)); | ||
| } | ||
|
|
||
|
|
||
| } |
Oops, something went wrong.