FINERACT-1694: Reworked bulk external events
This commit is contained in:
committed by
Arnold Gálovics
parent
0980aa8abf
commit
17b36a0919
+3
@@ -47,4 +47,7 @@ public interface BusinessEventNotifierService {
|
||||
*/
|
||||
<T extends BusinessEvent<?>> void addPostBusinessEventListener(Class<T> eventType, BusinessEventListener<T> listener);
|
||||
|
||||
void startExternalEventRecording();
|
||||
|
||||
void stopExternalEventRecording();
|
||||
}
|
||||
|
||||
+44
-5
@@ -26,6 +26,7 @@ import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.fineract.infrastructure.core.config.FineractProperties;
|
||||
import org.apache.fineract.infrastructure.event.business.BusinessEventListener;
|
||||
import org.apache.fineract.infrastructure.event.business.domain.BulkBusinessEvent;
|
||||
import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
|
||||
import org.apache.fineract.infrastructure.event.external.service.ExternalEventService;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
@@ -40,6 +41,9 @@ public class BusinessEventNotifierServiceImpl implements BusinessEventNotifierSe
|
||||
private final Map<Class, List<BusinessEventListener>> preListeners = new HashMap<>();
|
||||
private final Map<Class, List<BusinessEventListener>> postListeners = new HashMap<>();
|
||||
|
||||
private final ThreadLocal<Boolean> eventRecordingEnabled = ThreadLocal.withInitial(() -> false);
|
||||
private final ThreadLocal<List<BusinessEvent<?>>> recordedEvents = ThreadLocal.withInitial(ArrayList::new);
|
||||
|
||||
private final ExternalEventService externalEventService;
|
||||
private final FineractProperties fineractProperties;
|
||||
|
||||
@@ -54,6 +58,7 @@ public class BusinessEventNotifierServiceImpl implements BusinessEventNotifierSe
|
||||
|
||||
@Override
|
||||
public void notifyPreBusinessEvent(BusinessEvent<?> businessEvent) {
|
||||
throwExceptionIfBulkEvent(businessEvent);
|
||||
List<BusinessEventListener> businessEventListeners = preListeners.get(businessEvent.getClass());
|
||||
if (businessEventListeners != null) {
|
||||
for (BusinessEventListener eventListener : businessEventListeners) {
|
||||
@@ -74,6 +79,7 @@ public class BusinessEventNotifierServiceImpl implements BusinessEventNotifierSe
|
||||
|
||||
@Override
|
||||
public void notifyPostBusinessEvent(BusinessEvent<?> businessEvent) {
|
||||
throwExceptionIfBulkEvent(businessEvent);
|
||||
List<BusinessEventListener> businessEventListeners = postListeners.get(businessEvent.getClass());
|
||||
if (businessEventListeners != null) {
|
||||
for (BusinessEventListener eventListener : businessEventListeners) {
|
||||
@@ -82,14 +88,14 @@ public class BusinessEventNotifierServiceImpl implements BusinessEventNotifierSe
|
||||
}
|
||||
if (isExternalEventPostingEnabled()) {
|
||||
// we only want to create external events for operations that were successful, hence the post listener
|
||||
externalEventService.postEvent(businessEvent);
|
||||
if (isExternalEventRecordingEnabled()) {
|
||||
recordedEvents.get().add(businessEvent);
|
||||
} else {
|
||||
externalEventService.postEvent(businessEvent);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isExternalEventPostingEnabled() {
|
||||
return fineractProperties.getEvents().getExternal().isEnabled();
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T extends BusinessEvent<?>> void addPostBusinessEventListener(Class<T> eventType, BusinessEventListener<T> listener) {
|
||||
List<BusinessEventListener> businessEventListeners = postListeners.get(eventType);
|
||||
@@ -99,4 +105,37 @@ public class BusinessEventNotifierServiceImpl implements BusinessEventNotifierSe
|
||||
}
|
||||
businessEventListeners.add(listener);
|
||||
}
|
||||
|
||||
private boolean isExternalEventRecordingEnabled() {
|
||||
return eventRecordingEnabled.get();
|
||||
}
|
||||
|
||||
private boolean isExternalEventPostingEnabled() {
|
||||
return fineractProperties.getEvents().getExternal().isEnabled();
|
||||
}
|
||||
|
||||
private void throwExceptionIfBulkEvent(BusinessEvent<?> businessEvent) {
|
||||
if (businessEvent instanceof BulkBusinessEvent) {
|
||||
throw new IllegalArgumentException("BulkBusinessEvent cannot be raised directly");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void startExternalEventRecording() {
|
||||
eventRecordingEnabled.set(true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stopExternalEventRecording() {
|
||||
eventRecordingEnabled.set(false);
|
||||
try {
|
||||
List<BusinessEvent<?>> recordedBusinessEvents = recordedEvents.get();
|
||||
if (isExternalEventPostingEnabled()) {
|
||||
log.debug("Posting the BulkBusinessEvent for the recorded {} events", recordedBusinessEvents.size());
|
||||
externalEventService.postEvent(new BulkBusinessEvent(recordedBusinessEvents));
|
||||
}
|
||||
} finally {
|
||||
recordedEvents.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
-64
@@ -1,64 +0,0 @@
|
||||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
package org.apache.fineract.infrastructure.event.external.service;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import org.apache.fineract.infrastructure.event.business.domain.BulkBusinessEvent;
|
||||
import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class DelayedExternalEventService {
|
||||
|
||||
private final ThreadLocal<List<BusinessEvent<?>>> localEventStorage = ThreadLocal.withInitial(ArrayList::new);
|
||||
|
||||
private final ExternalEventService delegate;
|
||||
|
||||
public <T> void enqueueEvent(BusinessEvent<T> event) {
|
||||
if (event == null) {
|
||||
throw new IllegalArgumentException("event cannot be null");
|
||||
}
|
||||
|
||||
localEventStorage.get().add(event);
|
||||
}
|
||||
|
||||
public boolean hasEnqueuedEvents() {
|
||||
return !localEventStorage.get().isEmpty();
|
||||
}
|
||||
|
||||
public void clearEnqueuedEvents() {
|
||||
localEventStorage.get().clear();
|
||||
}
|
||||
|
||||
public List<BusinessEvent<?>> getEnqueuedEvents() {
|
||||
return List.copyOf(localEventStorage.get());
|
||||
}
|
||||
|
||||
public void postEnqueuedEvents() {
|
||||
List<BusinessEvent<?>> enqueuedEvents = localEventStorage.get();
|
||||
if (enqueuedEvents.isEmpty()) {
|
||||
throw new IllegalStateException("No events have been enqueued");
|
||||
}
|
||||
|
||||
delegate.postEvent(new BulkBusinessEvent(enqueuedEvents));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user