001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache license, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the license for the specific language governing permissions and
015 * limitations under the license.
016 */
017package org.apache.logging.log4j.core.appender;
018
019import java.io.IOException;
020import java.io.OutputStream;
021import java.nio.ByteBuffer;
022import java.util.Objects;
023
024import org.apache.logging.log4j.core.Layout;
025import org.apache.logging.log4j.core.layout.ByteBufferDestination;
026import org.apache.logging.log4j.core.util.Constants;
027
028/**
029 * Manages an OutputStream so that it can be shared by multiple Appenders and will
030 * allow appenders to reconfigure without requiring a new stream.
031 */
032public class OutputStreamManager extends AbstractManager implements ByteBufferDestination {
033    protected final Layout<?> layout;
034    protected ByteBuffer byteBuffer;
035    private volatile OutputStream os;
036
037    protected OutputStreamManager(final OutputStream os, final String streamName, final Layout<?> layout,
038            final boolean writeHeader) {
039        this(os, streamName, layout, writeHeader, ByteBuffer.wrap(new byte[Constants.ENCODER_BYTE_BUFFER_SIZE]));
040    }
041
042    /**
043     *
044     * @param os
045     * @param streamName
046     * @param layout
047     * @param writeHeader
048     * @param byteBuffer
049     * @since 2.6
050     */
051    protected OutputStreamManager(final OutputStream os, final String streamName, final Layout<?> layout,
052            final boolean writeHeader, final ByteBuffer byteBuffer) {
053        super(streamName);
054        this.os = os;
055        this.layout = layout;
056        if (writeHeader && layout != null) {
057            final byte[] header = layout.getHeader();
058            if (header != null) {
059                try {
060                    this.os.write(header, 0, header.length);
061                } catch (final IOException e) {
062                    logError("unable to write header", e);
063                }
064            }
065        }
066        this.byteBuffer = Objects.requireNonNull(byteBuffer, "byteBuffer");
067    }
068
069    /**
070     * Creates a Manager.
071     *
072     * @param name The name of the stream to manage.
073     * @param data The data to pass to the Manager.
074     * @param factory The factory to use to create the Manager.
075     * @param <T> The type of the OutputStreamManager.
076     * @return An OutputStreamManager.
077     */
078    public static <T> OutputStreamManager getManager(final String name, final T data,
079                                                 final ManagerFactory<? extends OutputStreamManager, T> factory) {
080        return AbstractManager.getManager(name, factory, data);
081    }
082
083    /**
084     * Default hook to write footer during close.
085     */
086    @Override
087    public void releaseSub() {
088        writeFooter();
089        close();
090    }
091
092    /**
093     * Writes the footer.
094     */
095    protected void writeFooter() {
096        if (layout == null) {
097            return;
098        }
099        final byte[] footer = layout.getFooter();
100        if (footer != null) {
101            write(footer);
102        }
103    }
104
105    /**
106     * Returns the status of the stream.
107     * @return true if the stream is open, false if it is not.
108     */
109    public boolean isOpen() {
110        return getCount() > 0;
111    }
112
113    protected OutputStream getOutputStream() {
114        return os;
115    }
116
117    protected void setOutputStream(final OutputStream os) {
118        final byte[] header = layout.getHeader();
119        if (header != null) {
120            try {
121                os.write(header, 0, header.length);
122                this.os = os; // only update field if os.write() succeeded
123            } catch (final IOException ioe) {
124                logError("unable to write header", ioe);
125            }
126        } else {
127            this.os = os;
128        }
129    }
130
131    /**
132     * Some output streams synchronize writes while others do not.
133     * @param bytes The serialized Log event.
134     * @throws AppenderLoggingException if an error occurs.
135     */
136    protected void write(final byte[] bytes)  {
137        write(bytes, 0, bytes.length, false);
138    }
139
140    /**
141     * Some output streams synchronize writes while others do not.
142     * @param bytes The serialized Log event.
143     * @param immediateFlush If true, flushes after writing.
144     * @throws AppenderLoggingException if an error occurs.
145     */
146    protected void write(final byte[] bytes, boolean immediateFlush)  {
147        write(bytes, 0, bytes.length, immediateFlush);
148    }
149
150    /**
151     * Some output streams synchronize writes while others do not. Synchronizing here insures that
152     * log events won't be intertwined.
153     * @param bytes The serialized Log event.
154     * @param offset The offset into the byte array.
155     * @param length The number of bytes to write.
156     * @throws AppenderLoggingException if an error occurs.
157     */
158    protected void write(final byte[] bytes, final int offset, final int length) {
159        write(bytes, offset, length, false);
160    }
161
162    /**
163     * Some output streams synchronize writes while others do not. Synchronizing here insures that
164     * log events won't be intertwined.
165     * @param bytes The serialized Log event.
166     * @param offset The offset into the byte array.
167     * @param length The number of bytes to write.
168     * @param immediateFlush flushes immediately after writing.
169     * @throws AppenderLoggingException if an error occurs.
170     */
171    protected synchronized void write(final byte[] bytes, final int offset, final int length, boolean immediateFlush) {
172        if (immediateFlush && byteBuffer.position() == 0) {
173            writeToDestination(bytes, offset, length);
174            flushDestination();
175            return;
176        }
177        if (length >= byteBuffer.capacity()) {
178            // if request length exceeds buffer capacity, flush the buffer and write the data directly
179            flush();
180            writeToDestination(bytes, offset, length);
181        } else {
182            if (length > byteBuffer.remaining()) {
183                flush();
184            }
185            byteBuffer.put(bytes, offset, length);
186        }
187        if (immediateFlush) {
188            flush();
189        }
190    }
191
192    /**
193     * Writes the specified section of the specified byte array to the stream.
194     *
195     * @param bytes the array containing data
196     * @param offset from where to write
197     * @param length how many bytes to write
198     * @since 2.6
199     */
200    protected synchronized void writeToDestination(final byte[] bytes, final int offset, final int length) {
201        try {
202            os.write(bytes, offset, length);
203        } catch (final IOException ex) {
204            final String msg = "Error writing to stream " + getName();
205            throw new AppenderLoggingException(msg, ex);
206        }
207    }
208
209    /**
210     * Calls {@code flush()} on the underlying output stream.
211     * @since 2.6
212     */
213    protected synchronized void flushDestination() {
214        try {
215            os.flush();
216        } catch (final IOException ex) {
217            final String msg = "Error flushing stream " + getName();
218            throw new AppenderLoggingException(msg, ex);
219        }
220    }
221
222    /**
223     * Drains the ByteBufferDestination's buffer into the destination. By default this calls
224     * {@link OutputStreamManager#write(byte[], int, int, boolean)} with the buffer contents.
225     * The underlying stream is not {@linkplain OutputStream#flush() flushed}.
226     *
227     * @see #flushDestination()
228     * @since 2.6
229     */
230    protected synchronized void flushBuffer(final ByteBuffer buf) {
231        buf.flip();
232        if (buf.limit() > 0) {
233            writeToDestination(buf.array(), 0, buf.limit());
234        }
235        buf.clear();
236    }
237
238    /**
239     * Flushes any buffers.
240     */
241    public synchronized void flush() {
242        flushBuffer(byteBuffer);
243        flushDestination();
244    }
245
246    protected synchronized void close() {
247        flush();
248        final OutputStream stream = os; // access volatile field only once per method
249        if (stream == System.out || stream == System.err) {
250            return;
251        }
252        try {
253            stream.close();
254        } catch (final IOException ex) {
255            logError("unable to close stream", ex);
256        }
257    }
258
259    /**
260     * Returns this {@code ByteBufferDestination}'s buffer.
261     * @return the buffer
262     * @since 2.6
263     */
264    @Override
265    public ByteBuffer getByteBuffer() {
266        return byteBuffer;
267    }
268
269    /**
270     * Drains the ByteBufferDestination's buffer into the destination. By default this calls
271     * {@link #flushBuffer(ByteBuffer)} with the specified buffer. Subclasses may override.
272     * <p>
273     * Do not call this method lightly! For some subclasses this is a very expensive operation. For example,
274     * {@link MemoryMappedFileManager} will assume this method was called because the end of the mapped region
275     * was reached during a text encoding operation and will {@linkplain MemoryMappedFileManager#remap() remap} its
276     * buffer.
277     * </p><p>
278     * To just flush the buffered contents to the underlying stream, call
279     * {@link #flushBuffer(ByteBuffer)} directly instead.
280     * </p>
281     *
282     * @param buf the buffer whose contents to write the the destination
283     * @return the specified buffer
284     * @since 2.6
285     */
286    @Override
287    public ByteBuffer drain(final ByteBuffer buf) {
288        flushBuffer(buf);
289        return buf;
290    }
291}