29371476372

Committed 14 Jul 2026 09:59PM UTC coverage: 63.839% (+0.03%) from 63.813%

Build # 29371476372

Build Type

Pull #6313

github

Committed by

web-flow

Commit Message

Merge 3ceccf384 into 918f05117

Pull Request Pull Request #6313: TRUNK-6516: Fix outbox/broker issues found in events/CDC sign-off (2.9.x)

Coverage Stats

19 of 27 new or added lines in 5 files covered. (70.37%)

12 existing lines in 4 files now uncovered.

24112 of 37770 relevant lines covered (63.84%)

0.64 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.67

/api/src/main/java/org/openmrs/event/outbox/OutboxEventRegistry.java

/**
 * This Source Code Form is subject to the terms of the Mozilla Public License,
 * v. 2.0. If a copy of the MPL was not distributed with this file, You can
 * obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
 * the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
 *
 * Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
 * graphic logo is a trademark of OpenMRS Inc.
 */
package org.openmrs.event.outbox;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.openmrs.event.outbox.tasks.OutboxTaskSchedulerInitializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.core.ResolvableType;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;

/**
 * It is used to discover any {@link OutboxEventListener} listeners and
 * dispatch events to them.
 * <p>
 * After scanning for listeners at startup it drives the outbox scheduler lifecycle, triggering
 * {@link OutboxTaskSchedulerInitializer#schedule()} when at least one listener is present and
 * {@link OutboxTaskSchedulerInitializer#deleteScheduledTasks()} when there are none or the feature
 * is disabled.
 *
 * @since 2.9.0
 */
@Component
public class OutboxEventRegistry implements SmartInitializingSingleton {

        private static final Logger log = LoggerFactory.getLogger(OutboxEventRegistry.class);

        private final ApplicationContext applicationContext;
        private final List<ListenerMethod> registry;
        private final Cache<ResolvableType, Boolean> hasOutboxListenersCache;
        private final boolean enabled;
        private final OutboxTaskSchedulerInitializer schedulerInitializer;

        public OutboxEventRegistry(ApplicationContext applicationContext, @Value("${outboxevent.enabled:true}") boolean enabled,
            OutboxTaskSchedulerInitializer schedulerInitializer) {
                this.applicationContext = applicationContext;
                this.registry = new ArrayList<>();
                this.hasOutboxListenersCache = CacheBuilder.newBuilder().expireAfterWrite(Duration.ofMinutes(10)).build();
                this.enabled = enabled;
                this.schedulerInitializer = schedulerInitializer;
        }

        @Override
        public void afterSingletonsInstantiated() {
                if (!enabled) {
                        schedulerInitializer.deleteScheduledTasks();
                        return;
                }

                for (String beanName : applicationContext.getBeanDefinitionNames()) {
                        Class<?> type = applicationContext.getType(beanName);
                        if (type != null) {
                                // Extract original class just in case it is wrapped by CGLIB proxy (e.g., @Transactional beans)
                                Class<?> userClass = ClassUtils.getUserClass(type);
                                ReflectionUtils.doWithMethods(userClass, method -> {
                                        if (AnnotatedElementUtils.hasAnnotation(method, OutboxEventListener.class)) {
                                                Class<?>[] parameterTypes = method.getParameterTypes();
                                                if (parameterTypes.length == 1) {
                                                        ResolvableType eventType = ResolvableType.forMethodParameter(method, 0);
                                                        Order orderAnnotation = AnnotatedElementUtils.findMergedAnnotation(method, Order.class);
                                                        int order = orderAnnotation != null ? orderAnnotation.value() : Ordered.LOWEST_PRECEDENCE;
                                                        registry.add(new ListenerMethod(eventType, beanName, method, order));
                                                }
                                        }
                                });
                        }
                }

                // Sort all listeners once at application startup
                Collections.sort(registry);

                // Trigger scheduling only after scanning is complete, so the poller starts when at least
                // one listener is present and is removed when none remain.
                if (hasOutboxListeners()) {
                        schedulerInitializer.schedule();
                } else {
                        schedulerInitializer.deleteScheduledTasks();
                }
        }
        
        public boolean hasOutboxListeners() {
                return !registry.isEmpty();
        }
        

        public boolean hasOutboxListeners(Object event) {
                if (!hasOutboxListeners()) {
                        return false;
                }
                
                ResolvableType resolvableType =  ResolvableType.forInstance(event);

                try {
                        return hasOutboxListenersCache.get(resolvableType, () -> 
                                registry.stream().anyMatch(listener -> listener.supports(resolvableType))
                        );
                } catch (ExecutionException e) {
                        throw new RuntimeException(e);
                }

        }
        
        public void dispatchOutboxEvent(Object event, Set<String> completedListeners, Runnable listenerCallback) {
                ResolvableType resolvableType = ResolvableType.forInstance(event);
                
                for (ListenerMethod listener : registry) {
                        if (listener.supports(resolvableType) && !completedListeners.contains(listener.getListenerId())) {
                                listener.invoke(applicationContext, event);
                                completedListeners.add(listener.getListenerId());
                                listenerCallback.run();
                        }
                }
        }

        private static class ListenerMethod implements Comparable<ListenerMethod> {
                private final ResolvableType targetEventType;
                private final String beanName;
                private final Method method;
                private final int order;
                private final String listenerId;

                public ListenerMethod(ResolvableType targetEventType, String beanName, Method method, int order) {
                        this.targetEventType = targetEventType;
                        this.beanName = beanName;
                        this.method = method;
                        this.order = order;
                        this.listenerId = beanName + "." + method.getName();
                }

                public boolean supports(ResolvableType eventType) {
                        return this.targetEventType.isAssignableFrom(eventType);
                }

                public String getListenerId() {
                        return listenerId;
                }

                public void invoke(ApplicationContext context, Object event) {
                        Object bean = context.getBean(beanName);
                        // Ensure we invoke on the proxy if the bean is wrapped (e.g., @Transactional)
                        Method methodToInvoke = ClassUtils.getMostSpecificMethod(method, bean.getClass());
                        ReflectionUtils.makeAccessible(methodToInvoke);
                        ReflectionUtils.invokeMethod(methodToInvoke, bean, event);
                }

                @Override
                public int compareTo(ListenerMethod other) {
                        return Integer.compare(this.order, other.order);
                }
        }
}

1	/**
2	* This Source Code Form is subject to the terms of the Mozilla Public License,
3	* v. 2.0. If a copy of the MPL was not distributed with this file, You can
4	* obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
5	* the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
6	*
7	* Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
8	* graphic logo is a trademark of OpenMRS Inc.
9	*/
10	package org.openmrs.event.outbox;
11
12	import com.google.common.cache.Cache;
13	import com.google.common.cache.CacheBuilder;
14	import org.openmrs.event.outbox.tasks.OutboxTaskSchedulerInitializer;
15	import org.slf4j.Logger;
16	import org.slf4j.LoggerFactory;
17	import org.springframework.beans.factory.SmartInitializingSingleton;
18	import org.springframework.beans.factory.annotation.Value;
19	import org.springframework.context.ApplicationContext;
20	import org.springframework.core.Ordered;
21	import org.springframework.core.ResolvableType;
22	import org.springframework.core.annotation.AnnotatedElementUtils;
23	import org.springframework.core.annotation.Order;
24	import org.springframework.stereotype.Component;
25	import org.springframework.util.ClassUtils;
26	import org.springframework.util.ReflectionUtils;
27
28	import java.lang.reflect.Method;
29	import java.time.Duration;
30	import java.util.ArrayList;
31	import java.util.Collections;
32	import java.util.List;
33	import java.util.Set;
34	import java.util.concurrent.ExecutionException;
35
36	/**
37	* It is used to discover any {@link OutboxEventListener} listeners and
38	* dispatch events to them.
39	* <p>
40	* After scanning for listeners at startup it drives the outbox scheduler lifecycle, triggering
41	* {@link OutboxTaskSchedulerInitializer#schedule()} when at least one listener is present and
42	* {@link OutboxTaskSchedulerInitializer#deleteScheduledTasks()} when there are none or the feature
43	* is disabled.
44	*
45	* @since 2.9.0
46	*/
47	@Component
48	public class OutboxEventRegistry implements SmartInitializingSingleton {
49
50	private static final Logger log = LoggerFactory.getLogger(OutboxEventRegistry.class);	1✔
51
52	private final ApplicationContext applicationContext;
53	private final List<ListenerMethod> registry;
54	private final Cache<ResolvableType, Boolean> hasOutboxListenersCache;
55	private final boolean enabled;
56	private final OutboxTaskSchedulerInitializer schedulerInitializer;
57
58	public OutboxEventRegistry(ApplicationContext applicationContext, @Value("${outboxevent.enabled:true}") boolean enabled,
59	OutboxTaskSchedulerInitializer schedulerInitializer) {	1✔
60	this.applicationContext = applicationContext;	1✔
61	this.registry = new ArrayList<>();	1✔
62	this.hasOutboxListenersCache = CacheBuilder.newBuilder().expireAfterWrite(Duration.ofMinutes(10)).build();	1✔
63	this.enabled = enabled;	1✔
64	this.schedulerInitializer = schedulerInitializer;	1✔
65	}	1✔
66
67	@Override
68	public void afterSingletonsInstantiated() {
69	if (!enabled) {	1✔
NEW 70	schedulerInitializer.deleteScheduledTasks();	×
UNCOV 71	return;	×
72	}
73
74	for (String beanName : applicationContext.getBeanDefinitionNames()) {	1✔
75	Class<?> type = applicationContext.getType(beanName);	1✔
76	if (type != null) {	1✔
77	// Extract original class just in case it is wrapped by CGLIB proxy (e.g., @Transactional beans)
78	Class<?> userClass = ClassUtils.getUserClass(type);	1✔
79	ReflectionUtils.doWithMethods(userClass, method -> {	1✔
80	if (AnnotatedElementUtils.hasAnnotation(method, OutboxEventListener.class)) {	1✔
81	Class<?>[] parameterTypes = method.getParameterTypes();	1✔
82	if (parameterTypes.length == 1) {	1✔
83	ResolvableType eventType = ResolvableType.forMethodParameter(method, 0);	1✔
84	Order orderAnnotation = AnnotatedElementUtils.findMergedAnnotation(method, Order.class);	1✔
85	int order = orderAnnotation != null ? orderAnnotation.value() : Ordered.LOWEST_PRECEDENCE;	1✔
86	registry.add(new ListenerMethod(eventType, beanName, method, order));	1✔
87	}
88	}
89	});	1✔
90	}
91	}
92
93	// Sort all listeners once at application startup
94	Collections.sort(registry);	1✔
95
96	// Trigger scheduling only after scanning is complete, so the poller starts when at least
97	// one listener is present and is removed when none remain.
98	if (hasOutboxListeners()) {	1✔
99	schedulerInitializer.schedule();	1✔
100	} else {
NEW 101	schedulerInitializer.deleteScheduledTasks();	×
102	}
103	}	1✔
104
105	public boolean hasOutboxListeners() {
106	return !registry.isEmpty();	1✔
107	}
108
109
110	public boolean hasOutboxListeners(Object event) {
111	if (!hasOutboxListeners()) {	1✔
112	return false;	×
113	}
114
115	ResolvableType resolvableType = ResolvableType.forInstance(event);	1✔
116
117	try {
118	return hasOutboxListenersCache.get(resolvableType, () ->	1✔
119	registry.stream().anyMatch(listener -> listener.supports(resolvableType))	1✔
120	);
121	} catch (ExecutionException e) {	×
122	throw new RuntimeException(e);	×
123	}
124
125	}
126
127	public void dispatchOutboxEvent(Object event, Set<String> completedListeners, Runnable listenerCallback) {
128	ResolvableType resolvableType = ResolvableType.forInstance(event);	×
129
130	for (ListenerMethod listener : registry) {	×
131	if (listener.supports(resolvableType) && !completedListeners.contains(listener.getListenerId())) {	×
132	listener.invoke(applicationContext, event);	×
133	completedListeners.add(listener.getListenerId());	×
134	listenerCallback.run();	×
135	}
136	}	×
137	}	×
138
139	private static class ListenerMethod implements Comparable<ListenerMethod> {
140	private final ResolvableType targetEventType;
141	private final String beanName;
142	private final Method method;
143	private final int order;
144	private final String listenerId;
145
146	public ListenerMethod(ResolvableType targetEventType, String beanName, Method method, int order) {	1✔
147	this.targetEventType = targetEventType;	1✔
148	this.beanName = beanName;	1✔
149	this.method = method;	1✔
150	this.order = order;	1✔
151	this.listenerId = beanName + "." + method.getName();	1✔
152	}	1✔
153
154	public boolean supports(ResolvableType eventType) {
155	return this.targetEventType.isAssignableFrom(eventType);	1✔
156	}
157
158	public String getListenerId() {
159	return listenerId;	×
160	}
161
162	public void invoke(ApplicationContext context, Object event) {
163	Object bean = context.getBean(beanName);	×
164	// Ensure we invoke on the proxy if the bean is wrapped (e.g., @Transactional)
165	Method methodToInvoke = ClassUtils.getMostSpecificMethod(method, bean.getClass());	×
166	ReflectionUtils.makeAccessible(methodToInvoke);	×
167	ReflectionUtils.invokeMethod(methodToInvoke, bean, event);	×
168	}	×
169
170	@Override
171	public int compareTo(ListenerMethod other) {
172	return Integer.compare(this.order, other.order);	1✔
173	}
174	}
175	}

openmrs / openmrs-core / 29371476372

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous