-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
MODTLR-64 consume and handle user update events
- Loading branch information
1 parent
21e44eb
commit 7096a98
Showing
5 changed files
with
121 additions
and
60 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
47 changes: 47 additions & 0 deletions
47
src/main/java/org/folio/service/impl/AbstractEventHandler.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,47 @@ | ||
| package org.folio.service.impl; | ||
|
|
||
| import java.util.function.Consumer; | ||
|
|
||
| import org.folio.domain.dto.UserTenant; | ||
| import org.folio.service.ConsortiaService; | ||
| import org.folio.service.KafkaEventHandler; | ||
| import org.folio.service.UserTenantsService; | ||
| import org.folio.spring.service.SystemUserScopedExecutionService; | ||
| import org.folio.support.KafkaEvent; | ||
|
|
||
| import lombok.AllArgsConstructor; | ||
| import lombok.extern.log4j.Log4j2; | ||
|
|
||
| @AllArgsConstructor | ||
| @Log4j2 | ||
| public abstract class AbstractEventHandler<T> implements KafkaEventHandler<T> { | ||
|
|
||
| protected final UserTenantsService userTenantsService; | ||
| protected final ConsortiaService consortiaService; | ||
| protected final SystemUserScopedExecutionService systemUserScopedExecutionService; | ||
|
|
||
| protected void processEvent(KafkaEvent<T> event, Consumer<T> eventConsumer) { | ||
| log.debug("processEvent:: params: event={}", () -> event); | ||
| UserTenant firstUserTenant = userTenantsService.findFirstUserTenant(); | ||
| if (firstUserTenant == null) { | ||
| log.info("processEvent: Failed to get user-tenants info"); | ||
| return; | ||
| } | ||
| String consortiumId = firstUserTenant.getConsortiumId(); | ||
| String centralTenantId = firstUserTenant.getCentralTenantId(); | ||
| log.info("processEvent:: consortiumId: {}, centralTenantId: {}", consortiumId, centralTenantId); | ||
|
|
||
| if (!centralTenantId.equals(event.getTenantIdHeaderValue())) { | ||
| log.info("processEvent: Ignoring non-central tenant event"); | ||
| return; | ||
| } | ||
| processForAllDataTenants(consortiumId, () -> eventConsumer.accept(event.getData().getNewVersion())); | ||
| } | ||
|
|
||
| private void processForAllDataTenants(String consortiumId, Runnable action) { | ||
| log.debug("processForAllDataTenants:: params: consortiumId={}", consortiumId); | ||
| consortiaService.getAllConsortiumTenants(consortiumId).getTenants().stream() | ||
| .filter(tenant -> !tenant.getIsCentral()) | ||
| .forEach(tenant -> systemUserScopedExecutionService.executeAsyncSystemUserScoped(tenant.getId(), action)); | ||
| } | ||
| } |
36 changes: 36 additions & 0 deletions
36
src/main/java/org/folio/service/impl/UserEventHandler.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,36 @@ | ||
| package org.folio.service.impl; | ||
|
|
||
| import org.folio.domain.dto.User; | ||
| import org.folio.service.ConsortiaService; | ||
| import org.folio.service.UserService; | ||
| import org.folio.service.UserTenantsService; | ||
| import org.folio.spring.service.SystemUserScopedExecutionService; | ||
| import org.folio.support.KafkaEvent; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| import lombok.extern.log4j.Log4j2; | ||
|
|
||
| @Log4j2 | ||
| @Service | ||
| public class UserEventHandler extends AbstractEventHandler<User> { | ||
|
|
||
| private final UserService userService; | ||
|
|
||
| public UserEventHandler(UserTenantsService userTenantsService, | ||
| ConsortiaService consortiaService, | ||
| SystemUserScopedExecutionService systemUserScopedExecutionService, | ||
| UserService userService) { | ||
|
|
||
| super(userTenantsService, consortiaService, systemUserScopedExecutionService); | ||
| this.userService = userService; | ||
| } | ||
|
|
||
| @Override | ||
| public void handle(KafkaEvent<User> event) { | ||
| log.info("handle:: Processing user event: {}", () -> event); | ||
| if (event.getType() == KafkaEvent.EventType.UPDATED) { | ||
| processEvent(event, userService::update); | ||
| } | ||
| } | ||
| } | ||
|
|
73 changes: 15 additions & 58 deletions
73
src/main/java/org/folio/service/impl/UserGroupEventHandler.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 |
|---|---|---|
| @@ -1,80 +1,37 @@ | ||
| package org.folio.service.impl; | ||
|
|
||
| import java.util.function.Consumer; | ||
|
|
||
| import org.folio.domain.dto.UserGroup; | ||
| import org.folio.domain.dto.UserTenant; | ||
| import org.folio.service.ConsortiaService; | ||
| import org.folio.service.KafkaEventHandler; | ||
| import org.folio.service.UserGroupService; | ||
| import org.folio.service.UserTenantsService; | ||
| import org.folio.spring.service.SystemUserScopedExecutionService; | ||
| import org.folio.support.KafkaEvent; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| import lombok.AllArgsConstructor; | ||
| import lombok.extern.log4j.Log4j2; | ||
|
|
||
| @AllArgsConstructor | ||
| @Service | ||
| @Log4j2 | ||
| public class UserGroupEventHandler implements KafkaEventHandler<UserGroup> { | ||
| @Service | ||
| public class UserGroupEventHandler extends AbstractEventHandler<UserGroup> { | ||
|
|
||
| private final UserTenantsService userTenantsService; | ||
| private final ConsortiaService consortiaService; | ||
| private final SystemUserScopedExecutionService systemUserScopedExecutionService; | ||
| private final UserGroupService userGroupService; | ||
|
|
||
| @Override | ||
| public void handle(KafkaEvent<UserGroup> event) { | ||
| log.info("handle:: Processing user group event: {}", () -> event); | ||
|
|
||
| KafkaEvent.EventType eventType = event.getType(); | ||
| if (eventType == KafkaEvent.EventType.CREATED) { | ||
| processUserGroupCreateEvent(event); | ||
| } | ||
| if (eventType == KafkaEvent.EventType.UPDATED) { | ||
| processUserGroupUpdateEvent(event); | ||
| } | ||
| } | ||
|
|
||
| private void processUserGroupCreateEvent(KafkaEvent<UserGroup> event){ | ||
| log.debug("processUserGroupCreateEvent:: params: event={}", () -> event); | ||
| processUserGroupEvent(event, userGroupService::create); | ||
| } | ||
| public UserGroupEventHandler(UserTenantsService userTenantsService, | ||
| ConsortiaService consortiaService, | ||
| SystemUserScopedExecutionService systemUserScopedExecutionService, | ||
| UserGroupService userGroupService) { | ||
|
|
||
| private void processUserGroupUpdateEvent(KafkaEvent<UserGroup> event) { | ||
| log.debug("processUserGroupUpdateEvent:: params: event={}", () -> event); | ||
| processUserGroupEvent(event, userGroupService::update); | ||
| super(userTenantsService, consortiaService, systemUserScopedExecutionService); | ||
| this.userGroupService = userGroupService; | ||
| } | ||
|
|
||
| private void processUserGroupEvent(KafkaEvent<UserGroup> event, | ||
| Consumer<UserGroup> userGroupConsumer) { | ||
|
|
||
| log.debug("processUserGroupEvent:: params: event={}", () -> event); | ||
| UserTenant firstUserTenant = userTenantsService.findFirstUserTenant(); | ||
| if (firstUserTenant == null) { | ||
| log.info("processUserGroupEvent: Failed to get user-tenants info"); | ||
| return; | ||
| } | ||
| String consortiumId = firstUserTenant.getConsortiumId(); | ||
| String centralTenantId = firstUserTenant.getCentralTenantId(); | ||
| log.info("processUserGroupEvent:: consortiumId: {}, centralTenantId: {}", | ||
| consortiumId, centralTenantId); | ||
|
|
||
| if (!centralTenantId.equals(event.getTenantIdHeaderValue())) { | ||
| log.info("processUserGroupEvent: Ignoring non-central tenant event"); | ||
| return; | ||
| @Override | ||
| public void handle(KafkaEvent<UserGroup> event){ | ||
| log.info("handle:: Processing user group event: {}", () -> event); | ||
| if (event.getType() == KafkaEvent.EventType.CREATED) { | ||
| processEvent(event, userGroupService::create); | ||
| } else if (event.getType() == KafkaEvent.EventType.UPDATED) { | ||
| processEvent(event, userGroupService::update); | ||
| } | ||
| processUserGroupForAllDataTenants(consortiumId, | ||
| () -> userGroupConsumer.accept(event.getData().getNewVersion())); | ||
| } | ||
|
|
||
| private void processUserGroupForAllDataTenants(String consortiumId, Runnable action) { | ||
| log.debug("processUserGroupForAllDataTenants:: params: consortiumId={}", consortiumId); | ||
| consortiaService.getAllConsortiumTenants(consortiumId).getTenants().stream() | ||
| .filter(tenant -> !tenant.getIsCentral()) | ||
| .forEach(tenant -> systemUserScopedExecutionService.executeAsyncSystemUserScoped( | ||
| tenant.getId(), action)); | ||
| } | ||
| } |
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