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

67.44

/api/src/main/java/org/openmrs/event/broker/BrokerEvent.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.broker;

import com.fasterxml.jackson.annotation.JsonIgnore;
import com.google.common.io.ByteSource;
import com.google.common.io.ByteStreams;
import com.google.common.io.FileBackedOutputStream;
import org.openmrs.event.BaseEvent;
import org.springframework.core.ResolvableType;
import org.springframework.core.ResolvableTypeProvider;
import org.springframework.lang.Nullable;

import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * Common code for {@link BrokerIncomingEvent} and {@link BrokerOutgoingEvent} to handle payload.
 * 
 * @param <T>
 * @since 2.9.0
 */
public abstract class BrokerEvent<T> extends BaseEvent implements ResolvableTypeProvider {
        private static final long serialVersionUID = 1L;
        
        private static final int MAX_MEMORY_PAYLOAD_SIZE = 256 * 1024; // 256 kB
        
        private final transient Object cacheLock = new Object();
        
        protected String broker;
        
        protected T payload;
        
        protected Map<String, Object> headers;
        
        private transient volatile ByteSource cachedPayloadSource;

        protected BrokerEvent() {
        }

        protected BrokerEvent(T payload) {
                this(payload, null);
        }

        protected BrokerEvent(T payload, String broker) {
                this(payload, broker, new HashMap<>());
        }


        protected BrokerEvent(T payload, String broker, Map<String, Object> headers) {
                this(payload, broker, headers, new HashSet<>());
        }

        protected BrokerEvent(T payload, String broker, Map<String, Object> headers, Set<String> tags) {
                super(tags);
                this.broker = broker;
                this.payload = payload;
                this.headers = headers;
        }

        /**
         * It implements caching of {@code InputStream} in memory or temporary file. Call {@code super.getPayload()} if
         * overriding.
         *
         * @return the payload
         */
        public T getPayload() {
                // Check if the payload is a stream that needs caching.
                if (payload instanceof InputStream) {
                        // Double-checked locking to ensure thread-safe, lazy initialization of the cache file.
                        if (cachedPayloadSource == null) {
                                synchronized (cacheLock) {
                                        if (cachedPayloadSource == null) {
                                                //noinspection UnstableApiUsage
                                                try (InputStream in = (InputStream) payload;
                                                     FileBackedOutputStream out = new FileBackedOutputStream(MAX_MEMORY_PAYLOAD_SIZE, true)) {

                                                        ByteStreams.copy(in, out);
                                                        //noinspection UnstableApiUsage
                                                        this.cachedPayloadSource = out.asByteSource();

                                                        // The original stream is consumed, null it out so we don't try to cache it again.
                                                        this.payload = null;
                                                } catch (IOException e) {
                                                        throw new UncheckedIOException("Failed to cache InputStream payload", e);
                                                }
                                        }
                                }
                        }
                }

                if (cachedPayloadSource != null) {
                        try {
                                //noinspection unchecked
                                return (T) cachedPayloadSource.openStream();
                        } catch (IOException e) {
                                throw new UncheckedIOException("Failed to open cached payload stream", e);
                        }
                }

                return payload;
        }

        public void setPayload(T payload) {
                this.payload = payload;
        }

        /**
         * The broker identifier this event came from.
         *
         * @return the broker
         */
        public String getBroker() {
                return broker;
        }

        public void setBroker(String broker) {
                this.broker = broker;
        }

        public Map<String, Object> getHeaders() {
                return headers;
        }

        public void setHeaders(Map<String, Object> headers) {
                this.headers = headers;
        }

        @JsonIgnore
        @Override
        public @Nullable ResolvableType getResolvableType() {
                if (payload != null && getClass().getTypeParameters().length == 1) {
                        return ResolvableType.forClassWithGenerics(getClass(), ResolvableType.forInstance(payload));
                } else {
                        return ResolvableType.forClass(getClass());
                }
        }
}

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.broker;
11
12	import com.fasterxml.jackson.annotation.JsonIgnore;
13	import com.google.common.io.ByteSource;
14	import com.google.common.io.ByteStreams;
15	import com.google.common.io.FileBackedOutputStream;
16	import org.openmrs.event.BaseEvent;
17	import org.springframework.core.ResolvableType;
18	import org.springframework.core.ResolvableTypeProvider;
19	import org.springframework.lang.Nullable;
20
21	import java.io.IOException;
22	import java.io.InputStream;
23	import java.io.UncheckedIOException;
24	import java.util.HashMap;
25	import java.util.HashSet;
26	import java.util.Map;
27	import java.util.Set;
28
29	/**
30	* Common code for {@link BrokerIncomingEvent} and {@link BrokerOutgoingEvent} to handle payload.
31	*
32	* @param <T>
33	* @since 2.9.0
34	*/
35	public abstract class BrokerEvent<T> extends BaseEvent implements ResolvableTypeProvider {
36	private static final long serialVersionUID = 1L;
37
38	private static final int MAX_MEMORY_PAYLOAD_SIZE = 256 * 1024; // 256 kB
39
40	private final transient Object cacheLock = new Object();	1✔
41
42	protected String broker;
43
44	protected T payload;
45
46	protected Map<String, Object> headers;
47
48	private transient volatile ByteSource cachedPayloadSource;
49
UNCOV 50	protected BrokerEvent() {	×
UNCOV 51	}	×
52
53	protected BrokerEvent(T payload) {
54	this(payload, null);	1✔
55	}	1✔
56
57	protected BrokerEvent(T payload, String broker) {
58	this(payload, broker, new HashMap<>());	1✔
59	}	1✔
60
61
62	protected BrokerEvent(T payload, String broker, Map<String, Object> headers) {
63	this(payload, broker, headers, new HashSet<>());	1✔
64	}	1✔
65
66	protected BrokerEvent(T payload, String broker, Map<String, Object> headers, Set<String> tags) {
67	super(tags);	1✔
68	this.broker = broker;	1✔
69	this.payload = payload;	1✔
70	this.headers = headers;	1✔
71	}	1✔
72
73	/**
74	* It implements caching of {@code InputStream} in memory or temporary file. Call {@code super.getPayload()} if
75	* overriding.
76	*
77	* @return the payload
78	*/
79	public T getPayload() {
80	// Check if the payload is a stream that needs caching.
81	if (payload instanceof InputStream) {	1✔
82	// Double-checked locking to ensure thread-safe, lazy initialization of the cache file.
83	if (cachedPayloadSource == null) {	1✔
84	synchronized (cacheLock) {	1✔
85	if (cachedPayloadSource == null) {	1✔
86	//noinspection UnstableApiUsage
87	try (InputStream in = (InputStream) payload;	1✔
88	FileBackedOutputStream out = new FileBackedOutputStream(MAX_MEMORY_PAYLOAD_SIZE, true)) {	1✔
89
90	ByteStreams.copy(in, out);	1✔
91	//noinspection UnstableApiUsage
92	this.cachedPayloadSource = out.asByteSource();	1✔
93
94	// The original stream is consumed, null it out so we don't try to cache it again.
95	this.payload = null;	1✔
96	} catch (IOException e) {	×
97	throw new UncheckedIOException("Failed to cache InputStream payload", e);	×
98	}	1✔
99	}
100	}	1✔
101	}
102	}
103
104	if (cachedPayloadSource != null) {	1✔
105	try {
106	//noinspection unchecked
107	return (T) cachedPayloadSource.openStream();	1✔
108	} catch (IOException e) {	×
109	throw new UncheckedIOException("Failed to open cached payload stream", e);	×
110	}
111	}
112
113	return payload;	1✔
114	}
115
116	public void setPayload(T payload) {
117	this.payload = payload;	×
118	}	×
119
120	/**
121	* The broker identifier this event came from.
122	*
123	* @return the broker
124	*/
125	public String getBroker() {
126	return broker;	1✔
127	}
128
129	public void setBroker(String broker) {
UNCOV 130	this.broker = broker;	×
UNCOV 131	}	×
132
133	public Map<String, Object> getHeaders() {
134	return headers;	×
135	}
136
137	public void setHeaders(Map<String, Object> headers) {
138	this.headers = headers;	×
139	}	×
140
141	@JsonIgnore
142	@Override
143	public @Nullable ResolvableType getResolvableType() {
144	if (payload != null && getClass().getTypeParameters().length == 1) {	1✔
145	return ResolvableType.forClassWithGenerics(getClass(), ResolvableType.forInstance(payload));	1✔
146	} else {
UNCOV 147	return ResolvableType.forClass(getClass());	×
148	}
149	}
150	}

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