30399987753

Committed 28 Jul 2026 09:16PM UTC coverage: 66.493% (+0.2%) from 66.313%

Build # 30399987753

Build Type

push

github

Committed by

ibacher

Commit Message

Fix Mockito inline mocks on JDK 8 and 11 (#6403)

Coverage Stats

24796 of 37291 relevant lines covered (66.49%)

0.67 hits per line

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

67.9

/api/src/main/java/org/openmrs/api/stream/StreamDataService.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.api.stream;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.unit.DataSize;

/**
 * The service can be used to convert data from OutputStream to InputStream without copying all data in memory.
 * <p>
 * The {@link #streamData(StreamDataWriter, Long)} method may run {@link StreamDataWriter#write(OutputStream)} in a 
 * separate thread using {@link TaskExecutor}.
 * <p>
 * It's providing the {@link java.io.PipedInputStream}/{@link java.io.PipedOutputStream} mechanism in a thread safe way 
 * with the use of {@link BlockingQueue}.
 * 
 * @since 2.8.0, 2.7.5, 2.6.16, 2.5.15
 */
@Service
public class StreamDataService {
        public static final int BUFFER_SIZE = (int) DataSize.ofKilobytes(128).toBytes();
        
        private static final Logger log = LoggerFactory.getLogger(StreamDataService.class);
        private final TaskExecutor taskExecutor;
        
        public StreamDataService(@Autowired TaskExecutor taskExecutor) {
                this.taskExecutor = taskExecutor;
        }
        
        private static class QueueInputStream extends InputStream {
                private final BlockingQueue<Integer> blockingQueue;
                private final long timeoutNanos;
                private volatile IOException streamException;

                public QueueInputStream() {
                        this.blockingQueue = new LinkedBlockingQueue<>(BUFFER_SIZE);
                        this.timeoutNanos = Duration.ofSeconds(30).toNanos();
                }

                public QueueOutputStream newQueueOutputStream() {
                        return new QueueOutputStream(this);
                }

                @Override
                public int read() throws IOException {
                        try {
                                checkStreamException();
                                
                                int result;
                                Integer peek = this.blockingQueue.peek();
                                if (Integer.valueOf(-1).equals(peek)) {
                                        result = -1;
                                } else {
                                        Integer value = this.blockingQueue.poll(this.timeoutNanos, TimeUnit.NANOSECONDS);
                                        if (value == null) {
                                                // Timeout
                                                result = -1;
                                        } else if (value == -1) {
                                                // End of stream. Put the end of stream back in the queue for consistency.
                                                this.blockingQueue.clear();
                                                if (!this.blockingQueue.offer(-1, timeoutNanos, TimeUnit.NANOSECONDS)) {
                                                        throw new IOException("Failed to write to full queue");
                                                }
                                                result = -1;
                                        } else {
                                                result = 255 & value;
                                        }
                                }

                                checkStreamException();
                                return result;
                        } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                                InterruptedIOException interruptedIoException = new InterruptedIOException();
                                interruptedIoException.initCause(e);
                                throw interruptedIoException;
                        }
                }

                @Override
                public void close() throws IOException {
                        checkStreamException();
                        super.close();
                }

                /**
                 * Propagate exception from a writing thread to a reading thread so that processing is stopped.
                 * 
                 * @param streamException exception
                 * @throws UncheckedIOException rethrows e
                 */
                public void propagateStreamException(IOException streamException) {
                        this.streamException = streamException;
                        // Wake a reader blocked on poll() so it sees the exception now instead of after the timeout.
                        // Non-blocking: if the queue is full the reader isn't waiting and will see it on its next read().
                        this.blockingQueue.offer(-1);
                }
                
                public void checkStreamException() throws IOException {
                        if (streamException != null) {
                                throw streamException;
                        }
                }
        }
        
        private static class QueueOutputStream extends OutputStream {
                private final QueueInputStream queueInputStream;
                
                public QueueOutputStream(QueueInputStream queueInputStream) {
                        this.queueInputStream = queueInputStream;
                }

                /**
                 * @param b   the <code>byte</code>.
                 * @throws IOException when queue full or interrupted
                 */
                @Override
                public void write(int b) throws IOException {
                        try {
                                queueInputStream.checkStreamException();
                                
                                if (!queueInputStream.blockingQueue.offer(255 & b, queueInputStream.timeoutNanos, TimeUnit.NANOSECONDS)) {
                                        IOException streamException = new IOException("Failed to write to full queue");
                                        queueInputStream.propagateStreamException(streamException);
                                }
                        } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                                InterruptedIOException interruptedIoException = new InterruptedIOException();
                                interruptedIoException.initCause(e);
                                throw interruptedIoException;
                        }
                }

                /**
                 * Closing the stream doesn't fail any following writes, but effectively only data up to closing the stream
                 * is read.
                 * 
                 * @throws IOException when queue full or interrupted
                 */
                @Override
                public void close() throws IOException {
                        try {
                                queueInputStream.checkStreamException();
                                
                                // Indicate the end of stream
                                if (!this.queueInputStream.blockingQueue.offer(-1, queueInputStream.timeoutNanos, TimeUnit.NANOSECONDS)) {
                                        IOException streamException = new IOException("Failed to write to full queue");
                                        queueInputStream.propagateStreamException(streamException);
                                }
                        } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                                InterruptedIOException interruptedIoException = new InterruptedIOException();
                                interruptedIoException.initCause(e);
                                throw interruptedIoException;
                        }
                }
        }

        /**
         * Runs {@link StreamDataWriter#write(OutputStream)} in a separate thread using {@link TaskExecutor} or copies 
         * in-memory if the length is smaller than {@link #BUFFER_SIZE}.
         * <p>
         * The returned InputStream doesn't need to be closed and the close operation takes no effect.
         * 
         * @param writer the write method
         * @param length the number of bytes if known or null
         * @return InputStream
         * 
         * @throws IOException when failing to stream data
         */
        public InputStream streamData(StreamDataWriter writer, Long length) throws IOException {
                if (length != null && length < BUFFER_SIZE) {
                        ByteArrayOutputStream out = new ByteArrayOutputStream(length.intValue());
                        try {
                                writer.write(out);
                        } catch (Exception e) {
                                throw new IOException("Failed to write data to byte array", e);
                        }
                        return new ByteArrayInputStream(out.toByteArray());
                } else {
                        QueueInputStream in = new QueueInputStream();

                        taskExecutor.execute(() -> {
                                QueueOutputStream out = in.newQueueOutputStream();
                                try {
                                        writer.write(out);
                                } catch (Exception e) {
                                        log.error("Failed to write data in parallel", e);
                                        in.propagateStreamException(new IOException("Failed to write data in parallel", e));
                                } finally {
                                        // Closing quietly as any exceptions in QueueOutputStream.close() are propagated
                                        IOUtils.closeQuietly(out);
                                }
                        });

                        return in;
                }
        }
}

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.api.stream;
11
12	import java.io.ByteArrayInputStream;
13	import java.io.ByteArrayOutputStream;
14	import java.io.IOException;
15	import java.io.InputStream;
16	import java.io.InterruptedIOException;
17	import java.io.OutputStream;
18	import java.io.UncheckedIOException;
19	import java.time.Duration;
20	import java.util.Objects;
21	import java.util.concurrent.BlockingQueue;
22	import java.util.concurrent.LinkedBlockingQueue;
23	import java.util.concurrent.TimeUnit;
24
25	import org.apache.commons.io.IOUtils;
26	import org.slf4j.Logger;
27	import org.slf4j.LoggerFactory;
28	import org.springframework.beans.factory.annotation.Autowired;
29	import org.springframework.core.task.TaskExecutor;
30	import org.springframework.stereotype.Service;
31	import org.springframework.util.unit.DataSize;
32
33	/**
34	* The service can be used to convert data from OutputStream to InputStream without copying all data in memory.
35	* <p>
36	* The {@link #streamData(StreamDataWriter, Long)} method may run {@link StreamDataWriter#write(OutputStream)} in a
37	* separate thread using {@link TaskExecutor}.
38	* <p>
39	* It's providing the {@link java.io.PipedInputStream}/{@link java.io.PipedOutputStream} mechanism in a thread safe way
40	* with the use of {@link BlockingQueue}.
41	*
42	* @since 2.8.0, 2.7.5, 2.6.16, 2.5.15
43	*/
44	@Service
45	public class StreamDataService {
46	public static final int BUFFER_SIZE = (int) DataSize.ofKilobytes(128).toBytes();	1✔
47
48	private static final Logger log = LoggerFactory.getLogger(StreamDataService.class);	1✔
49	private final TaskExecutor taskExecutor;
50
51	public StreamDataService(@Autowired TaskExecutor taskExecutor) {	1✔
52	this.taskExecutor = taskExecutor;	1✔
53	}	1✔
54
55	private static class QueueInputStream extends InputStream {
56	private final BlockingQueue<Integer> blockingQueue;
57	private final long timeoutNanos;
58	private volatile IOException streamException;
59
60	public QueueInputStream() {	1✔
61	this.blockingQueue = new LinkedBlockingQueue<>(BUFFER_SIZE);	1✔
62	this.timeoutNanos = Duration.ofSeconds(30).toNanos();	1✔
63	}	1✔
64
65	public QueueOutputStream newQueueOutputStream() {
66	return new QueueOutputStream(this);	1✔
67	}
68
69	@Override
70	public int read() throws IOException {
71	try {
72	checkStreamException();	1✔
73
74	int result;
75	Integer peek = this.blockingQueue.peek();	1✔
76	if (Integer.valueOf(-1).equals(peek)) {	1✔
77	result = -1;	1✔
78	} else {
79	Integer value = this.blockingQueue.poll(this.timeoutNanos, TimeUnit.NANOSECONDS);	1✔
80	if (value == null) {	1✔
81	// Timeout
82	result = -1;	×
83	} else if (value == -1) {	1✔
84	// End of stream. Put the end of stream back in the queue for consistency.
85	this.blockingQueue.clear();	1✔
86	if (!this.blockingQueue.offer(-1, timeoutNanos, TimeUnit.NANOSECONDS)) {	1✔
87	throw new IOException("Failed to write to full queue");	×
88	}
89	result = -1;	1✔
90	} else {
91	result = 255 & value;	1✔
92	}
93	}
94
95	checkStreamException();	1✔
96	return result;	1✔
97	} catch (InterruptedException e) {	×
98	Thread.currentThread().interrupt();	×
99	InterruptedIOException interruptedIoException = new InterruptedIOException();	×
100	interruptedIoException.initCause(e);	×
101	throw interruptedIoException;	×
102	}
103	}
104
105	@Override
106	public void close() throws IOException {
107	checkStreamException();	×
108	super.close();	×
109	}	×
110
111	/**
112	* Propagate exception from a writing thread to a reading thread so that processing is stopped.
113	*
114	* @param streamException exception
115	* @throws UncheckedIOException rethrows e
116	*/
117	public void propagateStreamException(IOException streamException) {
118	this.streamException = streamException;	1✔
119	// Wake a reader blocked on poll() so it sees the exception now instead of after the timeout.
120	// Non-blocking: if the queue is full the reader isn't waiting and will see it on its next read().
121	this.blockingQueue.offer(-1);	1✔
122	}	1✔
123
124	public void checkStreamException() throws IOException {
125	if (streamException != null) {	1✔
126	throw streamException;	1✔
127	}
128	}	1✔
129	}
130
131	private static class QueueOutputStream extends OutputStream {
132	private final QueueInputStream queueInputStream;
133
134	public QueueOutputStream(QueueInputStream queueInputStream) {	1✔
135	this.queueInputStream = queueInputStream;	1✔
136	}	1✔
137
138	/**
139	* @param b the <code>byte</code>.
140	* @throws IOException when queue full or interrupted
141	*/
142	@Override
143	public void write(int b) throws IOException {
144	try {
145	queueInputStream.checkStreamException();	1✔
146
147	if (!queueInputStream.blockingQueue.offer(255 & b, queueInputStream.timeoutNanos, TimeUnit.NANOSECONDS)) {	1✔
148	IOException streamException = new IOException("Failed to write to full queue");	×
149	queueInputStream.propagateStreamException(streamException);	×
150	}
151	} catch (InterruptedException e) {	×
152	Thread.currentThread().interrupt();	×
153	InterruptedIOException interruptedIoException = new InterruptedIOException();	×
154	interruptedIoException.initCause(e);	×
155	throw interruptedIoException;	×
156	}	1✔
157	}	1✔
158
159	/**
160	* Closing the stream doesn't fail any following writes, but effectively only data up to closing the stream
161	* is read.
162	*
163	* @throws IOException when queue full or interrupted
164	*/
165	@Override
166	public void close() throws IOException {
167	try {
168	queueInputStream.checkStreamException();	1✔
169
170	// Indicate the end of stream
171	if (!this.queueInputStream.blockingQueue.offer(-1, queueInputStream.timeoutNanos, TimeUnit.NANOSECONDS)) {	1✔
172	IOException streamException = new IOException("Failed to write to full queue");	×
173	queueInputStream.propagateStreamException(streamException);	×
174	}
175	} catch (InterruptedException e) {	×
176	Thread.currentThread().interrupt();	×
177	InterruptedIOException interruptedIoException = new InterruptedIOException();	×
178	interruptedIoException.initCause(e);	×
179	throw interruptedIoException;	×
180	}	1✔
181	}	1✔
182	}
183
184	/**
185	* Runs {@link StreamDataWriter#write(OutputStream)} in a separate thread using {@link TaskExecutor} or copies
186	* in-memory if the length is smaller than {@link #BUFFER_SIZE}.
187	* <p>
188	* The returned InputStream doesn't need to be closed and the close operation takes no effect.
189	*
190	* @param writer the write method
191	* @param length the number of bytes if known or null
192	* @return InputStream
193	*
194	* @throws IOException when failing to stream data
195	*/
196	public InputStream streamData(StreamDataWriter writer, Long length) throws IOException {
197	if (length != null && length < BUFFER_SIZE) {	1✔
198	ByteArrayOutputStream out = new ByteArrayOutputStream(length.intValue());	1✔
199	try {
200	writer.write(out);	1✔
201	} catch (Exception e) {	×
202	throw new IOException("Failed to write data to byte array", e);	×
203	}	1✔
204	return new ByteArrayInputStream(out.toByteArray());	1✔
205	} else {
206	QueueInputStream in = new QueueInputStream();	1✔
207
208	taskExecutor.execute(() -> {	1✔
209	QueueOutputStream out = in.newQueueOutputStream();	1✔
210	try {
211	writer.write(out);	1✔
212	} catch (Exception e) {	1✔
213	log.error("Failed to write data in parallel", e);	1✔
214	in.propagateStreamException(new IOException("Failed to write data in parallel", e));	1✔
215	} finally {
216	// Closing quietly as any exceptions in QueueOutputStream.close() are propagated
217	IOUtils.closeQuietly(out);	1✔
218	}
219	});	1✔
220
221	return in;	1✔
222	}
223	}
224	}

openmrs / openmrs-core / 30399987753

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