1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 package org.apache.logging.log4j.core.appender;
18
19 import java.io.IOException;
20 import java.io.OutputStream;
21 import java.io.Serializable;
22 import java.nio.ByteBuffer;
23 import java.util.Objects;
24 import java.util.concurrent.TimeUnit;
25
26 import org.apache.logging.log4j.core.Layout;
27 import org.apache.logging.log4j.core.LoggerContext;
28 import org.apache.logging.log4j.core.layout.ByteBufferDestination;
29 import org.apache.logging.log4j.core.layout.ByteBufferDestinationHelper;
30 import org.apache.logging.log4j.core.util.Constants;
31
32
33
34
35
36 public class OutputStreamManager extends AbstractManager implements ByteBufferDestination {
37 protected final Layout<?> layout;
38 protected ByteBuffer byteBuffer;
39 private volatile OutputStream outputStream;
40 private boolean skipFooter;
41
42 protected OutputStreamManager(final OutputStream os, final String streamName, final Layout<?> layout,
43 final boolean writeHeader) {
44
45 this(os, streamName, layout, writeHeader, Constants.ENCODER_BYTE_BUFFER_SIZE);
46 }
47
48 protected OutputStreamManager(final OutputStream os, final String streamName, final Layout<?> layout,
49 final boolean writeHeader, final int bufferSize) {
50
51 this(os, streamName, layout, writeHeader, ByteBuffer.wrap(new byte[bufferSize]));
52 }
53
54
55
56
57
58 @Deprecated
59 protected OutputStreamManager(final OutputStream os, final String streamName, final Layout<?> layout,
60 final boolean writeHeader, final ByteBuffer byteBuffer) {
61 super(null, streamName);
62 this.outputStream = os;
63 this.layout = layout;
64 if (writeHeader && layout != null) {
65 final byte[] header = layout.getHeader();
66 if (header != null) {
67 try {
68 getOutputStream().write(header, 0, header.length);
69 } catch (final IOException e) {
70 logError("Unable to write header", e);
71 }
72 }
73 }
74 this.byteBuffer = Objects.requireNonNull(byteBuffer, "byteBuffer");
75 }
76
77
78
79
80 protected OutputStreamManager(final LoggerContext loggerContext, final OutputStream os, final String streamName,
81 final boolean createOnDemand, final Layout<? extends Serializable> layout, final boolean writeHeader,
82 final ByteBuffer byteBuffer) {
83 super(loggerContext, streamName);
84 if (createOnDemand && os != null) {
85 LOGGER.error(
86 "Invalid OutputStreamManager configuration for '{}': You cannot both set the OutputStream and request on-demand.",
87 streamName);
88 }
89 this.layout = layout;
90 this.byteBuffer = Objects.requireNonNull(byteBuffer, "byteBuffer");
91 this.outputStream = os;
92 if (writeHeader && layout != null) {
93 final byte[] header = layout.getHeader();
94 if (header != null) {
95 try {
96 getOutputStream().write(header, 0, header.length);
97 } catch (final IOException e) {
98 logError("Unable to write header for " + streamName, e);
99 }
100 }
101 }
102 }
103
104
105
106
107
108
109
110
111
112
113 public static <T> OutputStreamManager getManager(final String name, final T data,
114 final ManagerFactory<? extends OutputStreamManager, T> factory) {
115 return AbstractManager.getManager(name, factory, data);
116 }
117
118 @SuppressWarnings("unused")
119 protected OutputStream createOutputStream() throws IOException {
120 throw new IllegalStateException(getClass().getCanonicalName() + " must implement createOutputStream()");
121 }
122
123
124
125
126
127 public void skipFooter(final boolean skipFooter) {
128 this.skipFooter = skipFooter;
129 }
130
131
132
133
134 @Override
135 public boolean releaseSub(final long timeout, final TimeUnit timeUnit) {
136 writeFooter();
137 return closeOutputStream();
138 }
139
140
141
142
143 protected void writeFooter() {
144 if (layout == null || skipFooter) {
145 return;
146 }
147 final byte[] footer = layout.getFooter();
148 if (footer != null) {
149 write(footer);
150 }
151 }
152
153
154
155
156
157 public boolean isOpen() {
158 return getCount() > 0;
159 }
160
161 public boolean hasOutputStream() {
162 return outputStream != null;
163 }
164
165 protected OutputStream getOutputStream() throws IOException {
166 if (outputStream == null) {
167 outputStream = createOutputStream();
168 }
169 return outputStream;
170 }
171
172 protected void setOutputStream(final OutputStream os) {
173 final byte[] header = layout.getHeader();
174 if (header != null) {
175 try {
176 os.write(header, 0, header.length);
177 this.outputStream = os;
178 } catch (final IOException ioe) {
179 logError("Unable to write header", ioe);
180 }
181 } else {
182 this.outputStream = os;
183 }
184 }
185
186
187
188
189
190
191 protected void write(final byte[] bytes) {
192 write(bytes, 0, bytes.length, false);
193 }
194
195
196
197
198
199
200
201 protected void write(final byte[] bytes, final boolean immediateFlush) {
202 write(bytes, 0, bytes.length, immediateFlush);
203 }
204
205 @Override
206 public void writeBytes(final byte[] data, final int offset, final int length) {
207 write(data, offset, length, false);
208 }
209
210
211
212
213
214
215
216
217
218 protected void write(final byte[] bytes, final int offset, final int length) {
219 writeBytes(bytes, offset, length);
220 }
221
222
223
224
225
226
227
228
229
230
231 protected synchronized void write(final byte[] bytes, final int offset, final int length, final boolean immediateFlush) {
232 if (immediateFlush && byteBuffer.position() == 0) {
233 writeToDestination(bytes, offset, length);
234 flushDestination();
235 return;
236 }
237 if (length >= byteBuffer.capacity()) {
238
239 flush();
240 writeToDestination(bytes, offset, length);
241 } else {
242 if (length > byteBuffer.remaining()) {
243 flush();
244 }
245 byteBuffer.put(bytes, offset, length);
246 }
247 if (immediateFlush) {
248 flush();
249 }
250 }
251
252
253
254
255
256
257
258
259
260 protected synchronized void writeToDestination(final byte[] bytes, final int offset, final int length) {
261 try {
262 getOutputStream().write(bytes, offset, length);
263 } catch (final IOException ex) {
264 throw new AppenderLoggingException("Error writing to stream " + getName(), ex);
265 }
266 }
267
268
269
270
271
272 protected synchronized void flushDestination() {
273 final OutputStream stream = outputStream;
274 if (stream != null) {
275 try {
276 stream.flush();
277 } catch (final IOException ex) {
278 throw new AppenderLoggingException("Error flushing stream " + getName(), ex);
279 }
280 }
281 }
282
283
284
285
286
287
288
289
290
291 protected synchronized void flushBuffer(final ByteBuffer buf) {
292 buf.flip();
293 if (buf.remaining() > 0) {
294 writeToDestination(buf.array(), buf.arrayOffset() + buf.position(), buf.remaining());
295 }
296 buf.clear();
297 }
298
299
300
301
302 public synchronized void flush() {
303 flushBuffer(byteBuffer);
304 flushDestination();
305 }
306
307 protected synchronized boolean closeOutputStream() {
308 flush();
309 final OutputStream stream = outputStream;
310 if (stream == null || stream == System.out || stream == System.err) {
311 return true;
312 }
313 try {
314 stream.close();
315 } catch (final IOException ex) {
316 logError("Unable to close stream", ex);
317 return false;
318 }
319 return true;
320 }
321
322
323
324
325
326
327 @Override
328 public ByteBuffer getByteBuffer() {
329 return byteBuffer;
330 }
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349 @Override
350 public ByteBuffer drain(final ByteBuffer buf) {
351 flushBuffer(buf);
352 return buf;
353 }
354
355 @Override
356 public void writeBytes(final ByteBuffer data) {
357 if (data.remaining() == 0) {
358 return;
359 }
360 synchronized (this) {
361 ByteBufferDestinationHelper.writeToUnsynchronized(data, this);
362 }
363 }
364 }