You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@logging.apache.org by rp...@apache.org on 2016/09/08 14:51:08 UTC
[39/50] [abbrv] logging-log4j2 git commit:
org.apache.logging.log4j.core.appender.OutputStreamManager.closeOutputStream()
now returns a boolean to signal whether there was a problem closing any
resouce.
http://git-wip-us.apache.org/repos/asf/logging-log4j2/blob/97e45ed6/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/rolling/RollingRandomAccessFileManager.java
----------------------------------------------------------------------
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/rolling/RollingRandomAccessFileManager.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/rolling/RollingRandomAccessFileManager.java
index 564934f..9be448b 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/rolling/RollingRandomAccessFileManager.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/rolling/RollingRandomAccessFileManager.java
@@ -1,263 +1,265 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache license, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the license for the specific language governing permissions and
- * limitations under the license.
- */
-package org.apache.logging.log4j.core.appender.rolling;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.io.RandomAccessFile;
-import java.io.Serializable;
-import java.nio.ByteBuffer;
-
-import org.apache.logging.log4j.core.Layout;
-import org.apache.logging.log4j.core.LoggerContext;
-import org.apache.logging.log4j.core.appender.AppenderLoggingException;
-import org.apache.logging.log4j.core.appender.ConfigurationFactoryData;
-import org.apache.logging.log4j.core.appender.ManagerFactory;
-import org.apache.logging.log4j.core.config.Configuration;
-import org.apache.logging.log4j.core.util.NullOutputStream;
-
-/**
- * Extends RollingFileManager but instead of using a buffered output stream, this class uses a {@code ByteBuffer} and a
- * {@code RandomAccessFile} to do the I/O.
- */
-public class RollingRandomAccessFileManager extends RollingFileManager {
- /**
- * The default buffer size.
- */
- public static final int DEFAULT_BUFFER_SIZE = 256 * 1024;
-
- private static final RollingRandomAccessFileManagerFactory FACTORY = new RollingRandomAccessFileManagerFactory();
-
- private RandomAccessFile randomAccessFile;
- private final ThreadLocal<Boolean> isEndOfBatch = new ThreadLocal<>();
-
- public RollingRandomAccessFileManager(final LoggerContext loggerContext, final RandomAccessFile raf,
- final String fileName, final String pattern, final OutputStream os, final boolean append,
- final boolean immediateFlush, final int bufferSize, final long size, final long time,
- final TriggeringPolicy policy, final RolloverStrategy strategy, final String advertiseURI,
- final Layout<? extends Serializable> layout, final boolean writeHeader) {
- super(loggerContext, fileName, pattern, os, append, false, size, time, policy, strategy, advertiseURI, layout,
- writeHeader, ByteBuffer.wrap(new byte[bufferSize]));
- this.randomAccessFile = raf;
- isEndOfBatch.set(Boolean.FALSE);
- writeHeader();
- }
-
- /**
- * Writes the layout's header to the file if it exists.
- */
- private void writeHeader() {
- if (layout == null) {
- return;
- }
- final byte[] header = layout.getHeader();
- if (header == null) {
- return;
- }
- try {
- if (randomAccessFile.length() == 0) {
- // write to the file, not to the buffer: the buffer may not be empty
- randomAccessFile.write(header, 0, header.length);
- }
- } catch (final IOException e) {
- logError("Unable to write header", e);
- }
- }
-
- public static RollingRandomAccessFileManager getRollingRandomAccessFileManager(final String fileName,
- final String filePattern, final boolean isAppend, final boolean immediateFlush, final int bufferSize,
- final TriggeringPolicy policy, final RolloverStrategy strategy, final String advertiseURI,
- final Layout<? extends Serializable> layout, final Configuration configuration) {
- return (RollingRandomAccessFileManager) getManager(fileName, new FactoryData(filePattern, isAppend,
- immediateFlush, bufferSize, policy, strategy, advertiseURI, layout, configuration), FACTORY);
- }
-
- public Boolean isEndOfBatch() {
- return isEndOfBatch.get();
- }
-
- public void setEndOfBatch(final boolean endOfBatch) {
- this.isEndOfBatch.set(Boolean.valueOf(endOfBatch));
- }
-
- // override to make visible for unit tests
- @Override
- protected synchronized void write(final byte[] bytes, final int offset, final int length,
- final boolean immediateFlush) {
- super.write(bytes, offset, length, immediateFlush);
- }
-
- @Override
- protected synchronized void writeToDestination(final byte[] bytes, final int offset, final int length) {
- try {
- randomAccessFile.write(bytes, offset, length);
- size += length;
- } catch (final IOException ex) {
- final String msg = "Error writing to RandomAccessFile " + getName();
- throw new AppenderLoggingException(msg, ex);
- }
- }
-
- @Override
- protected void createFileAfterRollover() throws IOException {
- this.randomAccessFile = new RandomAccessFile(getFileName(), "rw");
- if (isAppend()) {
- randomAccessFile.seek(randomAccessFile.length());
- }
- writeHeader();
- }
-
- @Override
- public synchronized void flush() {
- flushBuffer(byteBuffer);
- }
-
- @Override
- public synchronized void closeOutputStream() {
- flush();
- try {
- randomAccessFile.close();
- } catch (final IOException e) {
- logError("Unable to close RandomAccessFile", e);
- }
- }
-
- /**
- * Returns the buffer capacity.
- *
- * @return the buffer size
- */
- @Override
- public int getBufferSize() {
- return byteBuffer.capacity();
- }
-
- /**
- * Factory to create a RollingRandomAccessFileManager.
- */
- private static class RollingRandomAccessFileManagerFactory implements
- ManagerFactory<RollingRandomAccessFileManager, FactoryData> {
-
- /**
- * Create the RollingRandomAccessFileManager.
- *
- * @param name The name of the entity to manage.
- * @param data The data required to create the entity.
- * @return a RollingFileManager.
- */
- @Override
- public RollingRandomAccessFileManager createManager(final String name, final FactoryData data) {
- final File file = new File(name);
- final File parent = file.getParentFile();
- if (null != parent && !parent.exists()) {
- parent.mkdirs();
- }
-
- if (!data.append) {
- file.delete();
- }
- final long size = data.append ? file.length() : 0;
- final long time = file.exists() ? file.lastModified() : System.currentTimeMillis();
-
- final boolean writeHeader = !data.append || !file.exists();
- RandomAccessFile raf = null;
- try {
- raf = new RandomAccessFile(name, "rw");
- if (data.append) {
- final long length = raf.length();
- LOGGER.trace("RandomAccessFile {} seek to {}", name, length);
- raf.seek(length);
- } else {
- LOGGER.trace("RandomAccessFile {} set length to 0", name);
- raf.setLength(0);
- }
- return new RollingRandomAccessFileManager(data.getLoggerContext(), raf, name, data.pattern,
- NullOutputStream.NULL_OUTPUT_STREAM, data.append, data.immediateFlush, data.bufferSize, size, time, data.policy,
- data.strategy, data.advertiseURI, data.layout, writeHeader);
- } catch (final IOException ex) {
- LOGGER.error("Cannot access RandomAccessFile " + ex, ex);
- if (raf != null) {
- try {
- raf.close();
- } catch (final IOException e) {
- LOGGER.error("Cannot close RandomAccessFile {}", name, e);
- }
- }
- }
- return null;
- }
- }
-
- /**
- * Factory data.
- */
- private static class FactoryData extends ConfigurationFactoryData {
- private final String pattern;
- private final boolean append;
- private final boolean immediateFlush;
- private final int bufferSize;
- private final TriggeringPolicy policy;
- private final RolloverStrategy strategy;
- private final String advertiseURI;
- private final Layout<? extends Serializable> layout;
-
- /**
- * Create the data for the factory.
- *
- * @param pattern The pattern.
- * @param append The append flag.
- * @param immediateFlush
- * @param bufferSize
- * @param policy
- * @param strategy
- * @param advertiseURI
- * @param layout
- * @param configuration
- */
- public FactoryData(final String pattern, final boolean append, final boolean immediateFlush,
- final int bufferSize, final TriggeringPolicy policy, final RolloverStrategy strategy,
- final String advertiseURI, final Layout<? extends Serializable> layout, final Configuration configuration) {
- super(configuration);
- this.pattern = pattern;
- this.append = append;
- this.immediateFlush = immediateFlush;
- this.bufferSize = bufferSize;
- this.policy = policy;
- this.strategy = strategy;
- this.advertiseURI = advertiseURI;
- this.layout = layout;
- }
-
- public TriggeringPolicy getTriggeringPolicy()
- {
- return this.policy;
- }
-
- public RolloverStrategy getRolloverStrategy()
- {
- return this.strategy;
- }
- }
-
- @Override
- public void updateData(final Object data) {
- final FactoryData factoryData = (FactoryData) data;
- setRolloverStrategy(factoryData.getRolloverStrategy());
- setTriggeringPolicy(factoryData.getTriggeringPolicy());
- }
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache license, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the license for the specific language governing permissions and
+ * limitations under the license.
+ */
+package org.apache.logging.log4j.core.appender.rolling;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.RandomAccessFile;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+
+import org.apache.logging.log4j.core.Layout;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AppenderLoggingException;
+import org.apache.logging.log4j.core.appender.ConfigurationFactoryData;
+import org.apache.logging.log4j.core.appender.ManagerFactory;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.util.NullOutputStream;
+
+/**
+ * Extends RollingFileManager but instead of using a buffered output stream, this class uses a {@code ByteBuffer} and a
+ * {@code RandomAccessFile} to do the I/O.
+ */
+public class RollingRandomAccessFileManager extends RollingFileManager {
+ /**
+ * The default buffer size.
+ */
+ public static final int DEFAULT_BUFFER_SIZE = 256 * 1024;
+
+ private static final RollingRandomAccessFileManagerFactory FACTORY = new RollingRandomAccessFileManagerFactory();
+
+ private RandomAccessFile randomAccessFile;
+ private final ThreadLocal<Boolean> isEndOfBatch = new ThreadLocal<>();
+
+ public RollingRandomAccessFileManager(final LoggerContext loggerContext, final RandomAccessFile raf,
+ final String fileName, final String pattern, final OutputStream os, final boolean append,
+ final boolean immediateFlush, final int bufferSize, final long size, final long time,
+ final TriggeringPolicy policy, final RolloverStrategy strategy, final String advertiseURI,
+ final Layout<? extends Serializable> layout, final boolean writeHeader) {
+ super(loggerContext, fileName, pattern, os, append, false, size, time, policy, strategy, advertiseURI, layout,
+ writeHeader, ByteBuffer.wrap(new byte[bufferSize]));
+ this.randomAccessFile = raf;
+ isEndOfBatch.set(Boolean.FALSE);
+ writeHeader();
+ }
+
+ /**
+ * Writes the layout's header to the file if it exists.
+ */
+ private void writeHeader() {
+ if (layout == null) {
+ return;
+ }
+ final byte[] header = layout.getHeader();
+ if (header == null) {
+ return;
+ }
+ try {
+ if (randomAccessFile.length() == 0) {
+ // write to the file, not to the buffer: the buffer may not be empty
+ randomAccessFile.write(header, 0, header.length);
+ }
+ } catch (final IOException e) {
+ logError("Unable to write header", e);
+ }
+ }
+
+ public static RollingRandomAccessFileManager getRollingRandomAccessFileManager(final String fileName,
+ final String filePattern, final boolean isAppend, final boolean immediateFlush, final int bufferSize,
+ final TriggeringPolicy policy, final RolloverStrategy strategy, final String advertiseURI,
+ final Layout<? extends Serializable> layout, final Configuration configuration) {
+ return (RollingRandomAccessFileManager) getManager(fileName, new FactoryData(filePattern, isAppend,
+ immediateFlush, bufferSize, policy, strategy, advertiseURI, layout, configuration), FACTORY);
+ }
+
+ public Boolean isEndOfBatch() {
+ return isEndOfBatch.get();
+ }
+
+ public void setEndOfBatch(final boolean endOfBatch) {
+ this.isEndOfBatch.set(Boolean.valueOf(endOfBatch));
+ }
+
+ // override to make visible for unit tests
+ @Override
+ protected synchronized void write(final byte[] bytes, final int offset, final int length,
+ final boolean immediateFlush) {
+ super.write(bytes, offset, length, immediateFlush);
+ }
+
+ @Override
+ protected synchronized void writeToDestination(final byte[] bytes, final int offset, final int length) {
+ try {
+ randomAccessFile.write(bytes, offset, length);
+ size += length;
+ } catch (final IOException ex) {
+ final String msg = "Error writing to RandomAccessFile " + getName();
+ throw new AppenderLoggingException(msg, ex);
+ }
+ }
+
+ @Override
+ protected void createFileAfterRollover() throws IOException {
+ this.randomAccessFile = new RandomAccessFile(getFileName(), "rw");
+ if (isAppend()) {
+ randomAccessFile.seek(randomAccessFile.length());
+ }
+ writeHeader();
+ }
+
+ @Override
+ public synchronized void flush() {
+ flushBuffer(byteBuffer);
+ }
+
+ @Override
+ public synchronized boolean closeOutputStream() {
+ flush();
+ try {
+ randomAccessFile.close();
+ return true;
+ } catch (final IOException e) {
+ logError("Unable to close RandomAccessFile", e);
+ return false;
+ }
+ }
+
+ /**
+ * Returns the buffer capacity.
+ *
+ * @return the buffer size
+ */
+ @Override
+ public int getBufferSize() {
+ return byteBuffer.capacity();
+ }
+
+ /**
+ * Factory to create a RollingRandomAccessFileManager.
+ */
+ private static class RollingRandomAccessFileManagerFactory implements
+ ManagerFactory<RollingRandomAccessFileManager, FactoryData> {
+
+ /**
+ * Create the RollingRandomAccessFileManager.
+ *
+ * @param name The name of the entity to manage.
+ * @param data The data required to create the entity.
+ * @return a RollingFileManager.
+ */
+ @Override
+ public RollingRandomAccessFileManager createManager(final String name, final FactoryData data) {
+ final File file = new File(name);
+ final File parent = file.getParentFile();
+ if (null != parent && !parent.exists()) {
+ parent.mkdirs();
+ }
+
+ if (!data.append) {
+ file.delete();
+ }
+ final long size = data.append ? file.length() : 0;
+ final long time = file.exists() ? file.lastModified() : System.currentTimeMillis();
+
+ final boolean writeHeader = !data.append || !file.exists();
+ RandomAccessFile raf = null;
+ try {
+ raf = new RandomAccessFile(name, "rw");
+ if (data.append) {
+ final long length = raf.length();
+ LOGGER.trace("RandomAccessFile {} seek to {}", name, length);
+ raf.seek(length);
+ } else {
+ LOGGER.trace("RandomAccessFile {} set length to 0", name);
+ raf.setLength(0);
+ }
+ return new RollingRandomAccessFileManager(data.getLoggerContext(), raf, name, data.pattern,
+ NullOutputStream.NULL_OUTPUT_STREAM, data.append, data.immediateFlush, data.bufferSize, size, time, data.policy,
+ data.strategy, data.advertiseURI, data.layout, writeHeader);
+ } catch (final IOException ex) {
+ LOGGER.error("Cannot access RandomAccessFile " + ex, ex);
+ if (raf != null) {
+ try {
+ raf.close();
+ } catch (final IOException e) {
+ LOGGER.error("Cannot close RandomAccessFile {}", name, e);
+ }
+ }
+ }
+ return null;
+ }
+ }
+
+ /**
+ * Factory data.
+ */
+ private static class FactoryData extends ConfigurationFactoryData {
+ private final String pattern;
+ private final boolean append;
+ private final boolean immediateFlush;
+ private final int bufferSize;
+ private final TriggeringPolicy policy;
+ private final RolloverStrategy strategy;
+ private final String advertiseURI;
+ private final Layout<? extends Serializable> layout;
+
+ /**
+ * Create the data for the factory.
+ *
+ * @param pattern The pattern.
+ * @param append The append flag.
+ * @param immediateFlush
+ * @param bufferSize
+ * @param policy
+ * @param strategy
+ * @param advertiseURI
+ * @param layout
+ * @param configuration
+ */
+ public FactoryData(final String pattern, final boolean append, final boolean immediateFlush,
+ final int bufferSize, final TriggeringPolicy policy, final RolloverStrategy strategy,
+ final String advertiseURI, final Layout<? extends Serializable> layout, final Configuration configuration) {
+ super(configuration);
+ this.pattern = pattern;
+ this.append = append;
+ this.immediateFlush = immediateFlush;
+ this.bufferSize = bufferSize;
+ this.policy = policy;
+ this.strategy = strategy;
+ this.advertiseURI = advertiseURI;
+ this.layout = layout;
+ }
+
+ public TriggeringPolicy getTriggeringPolicy()
+ {
+ return this.policy;
+ }
+
+ public RolloverStrategy getRolloverStrategy()
+ {
+ return this.strategy;
+ }
+ }
+
+ @Override
+ public void updateData(final Object data) {
+ final FactoryData factoryData = (FactoryData) data;
+ setRolloverStrategy(factoryData.getRolloverStrategy());
+ setTriggeringPolicy(factoryData.getTriggeringPolicy());
+ }
+}
http://git-wip-us.apache.org/repos/asf/logging-log4j2/blob/97e45ed6/log4j-core/src/main/java/org/apache/logging/log4j/core/net/TcpSocketManager.java
----------------------------------------------------------------------
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/net/TcpSocketManager.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/net/TcpSocketManager.java
index e5af484..855b99a 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/net/TcpSocketManager.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/net/TcpSocketManager.java
@@ -1,325 +1,327 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache license, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the license for the specific language governing permissions and
- * limitations under the license.
- */
-package org.apache.logging.log4j.core.net;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.io.Serializable;
-import java.net.ConnectException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.UnknownHostException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-
-import org.apache.logging.log4j.core.Layout;
-import org.apache.logging.log4j.core.appender.AppenderLoggingException;
-import org.apache.logging.log4j.core.appender.ManagerFactory;
-import org.apache.logging.log4j.core.appender.OutputStreamManager;
-import org.apache.logging.log4j.core.util.Log4jThread;
-import org.apache.logging.log4j.core.util.NullOutputStream;
-import org.apache.logging.log4j.util.Strings;
-
-/**
- * Manager of TCP Socket connections.
- */
-public class TcpSocketManager extends AbstractSocketManager {
- /**
- The default reconnection delay (30000 milliseconds or 30 seconds).
- */
- public static final int DEFAULT_RECONNECTION_DELAY_MILLIS = 30000;
- /**
- The default port number of remote logging server (4560).
- */
- private static final int DEFAULT_PORT = 4560;
-
- private static final TcpSocketManagerFactory FACTORY = new TcpSocketManagerFactory();
-
- private final int reconnectionDelay;
-
- private Reconnector reconnector;
-
- private Socket socket;
-
- private final boolean retry;
-
- private final boolean immediateFail;
-
- private final int connectTimeoutMillis;
-
- /**
- * The Constructor.
- * @param name The unique name of this connection.
- * @param os The OutputStream.
- * @param socket The Socket.
- * @param inetAddress The Internet address of the host.
- * @param host The name of the host.
- * @param port The port number on the host.
- * @param connectTimeoutMillis the connect timeout in milliseconds.
- * @param delay Reconnection interval.
- * @param immediateFail True if the write should fail if no socket is immediately available.
- * @param layout The Layout.
- * @param bufferSize The buffer size.
- */
- public TcpSocketManager(final String name, final OutputStream os, final Socket socket, final InetAddress inetAddress,
- final String host, final int port, final int connectTimeoutMillis, final int delay,
- final boolean immediateFail, final Layout<? extends Serializable> layout, final int bufferSize) {
- super(name, os, inetAddress, host, port, layout, true, bufferSize);
- this.connectTimeoutMillis = connectTimeoutMillis;
- this.reconnectionDelay = delay;
- this.socket = socket;
- this.immediateFail = immediateFail;
- retry = delay > 0;
- if (socket == null) {
- reconnector = createReconnector();
- reconnector.start();
- }
- }
-
- /**
- * Obtain a TcpSocketManager.
- * @param host The host to connect to.
- * @param port The port on the host.
- * @param connectTimeoutMillis the connect timeout in milliseconds
- * @param reconnectDelayMillis The interval to pause between retries.
- * @param bufferSize The buffer size.
- * @return A TcpSocketManager.
- */
- public static TcpSocketManager getSocketManager(final String host, int port, final int connectTimeoutMillis,
- int reconnectDelayMillis, final boolean immediateFail, final Layout<? extends Serializable> layout, final int bufferSize) {
- if (Strings.isEmpty(host)) {
- throw new IllegalArgumentException("A host name is required");
- }
- if (port <= 0) {
- port = DEFAULT_PORT;
- }
- if (reconnectDelayMillis == 0) {
- reconnectDelayMillis = DEFAULT_RECONNECTION_DELAY_MILLIS;
- }
- return (TcpSocketManager) getManager("TCP:" + host + ':' + port, new FactoryData(
- host, port, connectTimeoutMillis, reconnectDelayMillis, immediateFail, layout, bufferSize), FACTORY);
- }
-
- @Override
- protected void write(final byte[] bytes, final int offset, final int length, final boolean immediateFlush) {
- if (socket == null) {
- if (reconnector != null && !immediateFail) {
- reconnector.latch();
- }
- if (socket == null) {
- final String msg = "Error writing to " + getName() + " socket not available";
- throw new AppenderLoggingException(msg);
- }
- }
- synchronized (this) {
- try {
- final OutputStream outputStream = getOutputStream();
- outputStream.write(bytes, offset, length);
- if (immediateFlush) {
- outputStream.flush();
- }
- } catch (final IOException ex) {
- if (retry && reconnector == null) {
- reconnector = createReconnector();
- reconnector.start();
- }
- final String msg = "Error writing to " + getName();
- throw new AppenderLoggingException(msg, ex);
- }
- }
- }
-
- @Override
- protected synchronized void closeOutputStream() {
- super.closeOutputStream();
- if (reconnector != null) {
- reconnector.shutdown();
- reconnector.interrupt();
- reconnector = null;
- }
- final Socket oldSocket = socket;
- socket = null;
- if (oldSocket != null) {
- try {
- oldSocket.close();
- } catch (final IOException e) {
- LOGGER.error("Could not close socket {}", socket);
- }
- }
- }
-
- public int getConnectTimeoutMillis() {
- return connectTimeoutMillis;
- }
-
- /**
- * Gets this TcpSocketManager's content format. Specified by:
- * <ul>
- * <li>Key: "protocol" Value: "tcp"</li>
- * <li>Key: "direction" Value: "out"</li>
- * </ul>
- *
- * @return Map of content format keys supporting TcpSocketManager
- */
- @Override
- public Map<String, String> getContentFormat() {
- final Map<String, String> result = new HashMap<>(super.getContentFormat());
- result.put("protocol", "tcp");
- result.put("direction", "out");
- return result;
- }
-
- /**
- * Handles reconnecting to a Thread.
- */
- private class Reconnector extends Log4jThread {
-
- private final CountDownLatch latch = new CountDownLatch(1);
-
- private boolean shutdown = false;
-
- private final Object owner;
-
- public Reconnector(final OutputStreamManager owner) {
- super("TcpSocketManager-Reconnector");
- this.owner = owner;
- }
-
- public void latch() {
- try {
- latch.await();
- } catch (final InterruptedException ex) {
- // Ignore the exception.
- }
- }
-
- public void shutdown() {
- shutdown = true;
- }
-
- @Override
- public void run() {
- while (!shutdown) {
- try {
- sleep(reconnectionDelay);
- final Socket sock = createSocket(inetAddress, port);
- final OutputStream newOS = sock.getOutputStream();
- synchronized (owner) {
- try {
- getOutputStream().close();
- } catch (final IOException ioe) {
- // Ignore this.
- }
-
- setOutputStream(newOS);
- socket = sock;
- reconnector = null;
- shutdown = true;
- }
- LOGGER.debug("Connection to " + host + ':' + port + " reestablished.");
- } catch (final InterruptedException ie) {
- LOGGER.debug("Reconnection interrupted.");
- } catch (final ConnectException ex) {
- LOGGER.debug(host + ':' + port + " refused connection");
- } catch (final IOException ioe) {
- LOGGER.debug("Unable to reconnect to " + host + ':' + port);
- } finally {
- latch.countDown();
- }
- }
- }
- }
-
- private Reconnector createReconnector() {
- final Reconnector recon = new Reconnector(this);
- recon.setDaemon(true);
- recon.setPriority(Thread.MIN_PRIORITY);
- return recon;
- }
-
- protected Socket createSocket(final InetAddress host, final int port) throws IOException {
- return createSocket(host.getHostName(), port);
- }
-
- protected Socket createSocket(final String host, final int port) throws IOException {
- final InetSocketAddress address = new InetSocketAddress(host, port);
- final Socket newSocket = new Socket();
- newSocket.connect(address, connectTimeoutMillis);
- return newSocket;
- }
-
- /**
- * Data for the factory.
- */
- private static class FactoryData {
- private final String host;
- private final int port;
- private final int connectTimeoutMillis;
- private final int reconnectDelayMillis;
- private final boolean immediateFail;
- private final Layout<? extends Serializable> layout;
- private final int bufferSize;
-
- public FactoryData(final String host, final int port, final int connectTimeoutMillis, final int reconnectDelayMillis,
- final boolean immediateFail, final Layout<? extends Serializable> layout, final int bufferSize) {
- this.host = host;
- this.port = port;
- this.connectTimeoutMillis = connectTimeoutMillis;
- this.reconnectDelayMillis = reconnectDelayMillis;
- this.immediateFail = immediateFail;
- this.layout = layout;
- this.bufferSize = bufferSize;
- }
- }
-
- /**
- * Factory to create a TcpSocketManager.
- */
- protected static class TcpSocketManagerFactory implements ManagerFactory<TcpSocketManager, FactoryData> {
- @Override
- public TcpSocketManager createManager(final String name, final FactoryData data) {
-
- InetAddress inetAddress;
- OutputStream os;
- try {
- inetAddress = InetAddress.getByName(data.host);
- } catch (final UnknownHostException ex) {
- LOGGER.error("Could not find address of " + data.host, ex, ex);
- return null;
- }
- try {
- // LOG4J2-1042
- final Socket socket = new Socket();
- socket.connect(new InetSocketAddress(data.host, data.port), data.connectTimeoutMillis);
- os = socket.getOutputStream();
- return new TcpSocketManager(name, os, socket, inetAddress, data.host, data.port,
- data.connectTimeoutMillis, data.reconnectDelayMillis, data.immediateFail, data.layout, data.bufferSize);
- } catch (final IOException ex) {
- LOGGER.error("TcpSocketManager (" + name + ") " + ex, ex);
- os = NullOutputStream.NULL_OUTPUT_STREAM;
- }
- if (data.reconnectDelayMillis == 0) {
- return null;
- }
- return new TcpSocketManager(name, os, null, inetAddress, data.host, data.port, data.connectTimeoutMillis,
- data.reconnectDelayMillis, data.immediateFail, data.layout, data.bufferSize);
- }
- }
-
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache license, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the license for the specific language governing permissions and
+ * limitations under the license.
+ */
+package org.apache.logging.log4j.core.net;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.net.ConnectException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.UnknownHostException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.logging.log4j.core.Layout;
+import org.apache.logging.log4j.core.appender.AppenderLoggingException;
+import org.apache.logging.log4j.core.appender.ManagerFactory;
+import org.apache.logging.log4j.core.appender.OutputStreamManager;
+import org.apache.logging.log4j.core.util.Log4jThread;
+import org.apache.logging.log4j.core.util.NullOutputStream;
+import org.apache.logging.log4j.util.Strings;
+
+/**
+ * Manager of TCP Socket connections.
+ */
+public class TcpSocketManager extends AbstractSocketManager {
+ /**
+ The default reconnection delay (30000 milliseconds or 30 seconds).
+ */
+ public static final int DEFAULT_RECONNECTION_DELAY_MILLIS = 30000;
+ /**
+ The default port number of remote logging server (4560).
+ */
+ private static final int DEFAULT_PORT = 4560;
+
+ private static final TcpSocketManagerFactory FACTORY = new TcpSocketManagerFactory();
+
+ private final int reconnectionDelay;
+
+ private Reconnector reconnector;
+
+ private Socket socket;
+
+ private final boolean retry;
+
+ private final boolean immediateFail;
+
+ private final int connectTimeoutMillis;
+
+ /**
+ * The Constructor.
+ * @param name The unique name of this connection.
+ * @param os The OutputStream.
+ * @param socket The Socket.
+ * @param inetAddress The Internet address of the host.
+ * @param host The name of the host.
+ * @param port The port number on the host.
+ * @param connectTimeoutMillis the connect timeout in milliseconds.
+ * @param delay Reconnection interval.
+ * @param immediateFail True if the write should fail if no socket is immediately available.
+ * @param layout The Layout.
+ * @param bufferSize The buffer size.
+ */
+ public TcpSocketManager(final String name, final OutputStream os, final Socket socket, final InetAddress inetAddress,
+ final String host, final int port, final int connectTimeoutMillis, final int delay,
+ final boolean immediateFail, final Layout<? extends Serializable> layout, final int bufferSize) {
+ super(name, os, inetAddress, host, port, layout, true, bufferSize);
+ this.connectTimeoutMillis = connectTimeoutMillis;
+ this.reconnectionDelay = delay;
+ this.socket = socket;
+ this.immediateFail = immediateFail;
+ retry = delay > 0;
+ if (socket == null) {
+ reconnector = createReconnector();
+ reconnector.start();
+ }
+ }
+
+ /**
+ * Obtain a TcpSocketManager.
+ * @param host The host to connect to.
+ * @param port The port on the host.
+ * @param connectTimeoutMillis the connect timeout in milliseconds
+ * @param reconnectDelayMillis The interval to pause between retries.
+ * @param bufferSize The buffer size.
+ * @return A TcpSocketManager.
+ */
+ public static TcpSocketManager getSocketManager(final String host, int port, final int connectTimeoutMillis,
+ int reconnectDelayMillis, final boolean immediateFail, final Layout<? extends Serializable> layout, final int bufferSize) {
+ if (Strings.isEmpty(host)) {
+ throw new IllegalArgumentException("A host name is required");
+ }
+ if (port <= 0) {
+ port = DEFAULT_PORT;
+ }
+ if (reconnectDelayMillis == 0) {
+ reconnectDelayMillis = DEFAULT_RECONNECTION_DELAY_MILLIS;
+ }
+ return (TcpSocketManager) getManager("TCP:" + host + ':' + port, new FactoryData(
+ host, port, connectTimeoutMillis, reconnectDelayMillis, immediateFail, layout, bufferSize), FACTORY);
+ }
+
+ @Override
+ protected void write(final byte[] bytes, final int offset, final int length, final boolean immediateFlush) {
+ if (socket == null) {
+ if (reconnector != null && !immediateFail) {
+ reconnector.latch();
+ }
+ if (socket == null) {
+ final String msg = "Error writing to " + getName() + " socket not available";
+ throw new AppenderLoggingException(msg);
+ }
+ }
+ synchronized (this) {
+ try {
+ final OutputStream outputStream = getOutputStream();
+ outputStream.write(bytes, offset, length);
+ if (immediateFlush) {
+ outputStream.flush();
+ }
+ } catch (final IOException ex) {
+ if (retry && reconnector == null) {
+ reconnector = createReconnector();
+ reconnector.start();
+ }
+ final String msg = "Error writing to " + getName();
+ throw new AppenderLoggingException(msg, ex);
+ }
+ }
+ }
+
+ @Override
+ protected synchronized boolean closeOutputStream() {
+ boolean closed = super.closeOutputStream();
+ if (reconnector != null) {
+ reconnector.shutdown();
+ reconnector.interrupt();
+ reconnector = null;
+ }
+ final Socket oldSocket = socket;
+ socket = null;
+ if (oldSocket != null) {
+ try {
+ oldSocket.close();
+ } catch (final IOException e) {
+ LOGGER.error("Could not close socket {}", socket);
+ return false;
+ }
+ }
+ return closed;
+ }
+
+ public int getConnectTimeoutMillis() {
+ return connectTimeoutMillis;
+ }
+
+ /**
+ * Gets this TcpSocketManager's content format. Specified by:
+ * <ul>
+ * <li>Key: "protocol" Value: "tcp"</li>
+ * <li>Key: "direction" Value: "out"</li>
+ * </ul>
+ *
+ * @return Map of content format keys supporting TcpSocketManager
+ */
+ @Override
+ public Map<String, String> getContentFormat() {
+ final Map<String, String> result = new HashMap<>(super.getContentFormat());
+ result.put("protocol", "tcp");
+ result.put("direction", "out");
+ return result;
+ }
+
+ /**
+ * Handles reconnecting to a Thread.
+ */
+ private class Reconnector extends Log4jThread {
+
+ private final CountDownLatch latch = new CountDownLatch(1);
+
+ private boolean shutdown = false;
+
+ private final Object owner;
+
+ public Reconnector(final OutputStreamManager owner) {
+ super("TcpSocketManager-Reconnector");
+ this.owner = owner;
+ }
+
+ public void latch() {
+ try {
+ latch.await();
+ } catch (final InterruptedException ex) {
+ // Ignore the exception.
+ }
+ }
+
+ public void shutdown() {
+ shutdown = true;
+ }
+
+ @Override
+ public void run() {
+ while (!shutdown) {
+ try {
+ sleep(reconnectionDelay);
+ final Socket sock = createSocket(inetAddress, port);
+ final OutputStream newOS = sock.getOutputStream();
+ synchronized (owner) {
+ try {
+ getOutputStream().close();
+ } catch (final IOException ioe) {
+ // Ignore this.
+ }
+
+ setOutputStream(newOS);
+ socket = sock;
+ reconnector = null;
+ shutdown = true;
+ }
+ LOGGER.debug("Connection to " + host + ':' + port + " reestablished.");
+ } catch (final InterruptedException ie) {
+ LOGGER.debug("Reconnection interrupted.");
+ } catch (final ConnectException ex) {
+ LOGGER.debug(host + ':' + port + " refused connection");
+ } catch (final IOException ioe) {
+ LOGGER.debug("Unable to reconnect to " + host + ':' + port);
+ } finally {
+ latch.countDown();
+ }
+ }
+ }
+ }
+
+ private Reconnector createReconnector() {
+ final Reconnector recon = new Reconnector(this);
+ recon.setDaemon(true);
+ recon.setPriority(Thread.MIN_PRIORITY);
+ return recon;
+ }
+
+ protected Socket createSocket(final InetAddress host, final int port) throws IOException {
+ return createSocket(host.getHostName(), port);
+ }
+
+ protected Socket createSocket(final String host, final int port) throws IOException {
+ final InetSocketAddress address = new InetSocketAddress(host, port);
+ final Socket newSocket = new Socket();
+ newSocket.connect(address, connectTimeoutMillis);
+ return newSocket;
+ }
+
+ /**
+ * Data for the factory.
+ */
+ private static class FactoryData {
+ private final String host;
+ private final int port;
+ private final int connectTimeoutMillis;
+ private final int reconnectDelayMillis;
+ private final boolean immediateFail;
+ private final Layout<? extends Serializable> layout;
+ private final int bufferSize;
+
+ public FactoryData(final String host, final int port, final int connectTimeoutMillis, final int reconnectDelayMillis,
+ final boolean immediateFail, final Layout<? extends Serializable> layout, final int bufferSize) {
+ this.host = host;
+ this.port = port;
+ this.connectTimeoutMillis = connectTimeoutMillis;
+ this.reconnectDelayMillis = reconnectDelayMillis;
+ this.immediateFail = immediateFail;
+ this.layout = layout;
+ this.bufferSize = bufferSize;
+ }
+ }
+
+ /**
+ * Factory to create a TcpSocketManager.
+ */
+ protected static class TcpSocketManagerFactory implements ManagerFactory<TcpSocketManager, FactoryData> {
+ @Override
+ public TcpSocketManager createManager(final String name, final FactoryData data) {
+
+ InetAddress inetAddress;
+ OutputStream os;
+ try {
+ inetAddress = InetAddress.getByName(data.host);
+ } catch (final UnknownHostException ex) {
+ LOGGER.error("Could not find address of " + data.host, ex, ex);
+ return null;
+ }
+ try {
+ // LOG4J2-1042
+ final Socket socket = new Socket();
+ socket.connect(new InetSocketAddress(data.host, data.port), data.connectTimeoutMillis);
+ os = socket.getOutputStream();
+ return new TcpSocketManager(name, os, socket, inetAddress, data.host, data.port,
+ data.connectTimeoutMillis, data.reconnectDelayMillis, data.immediateFail, data.layout, data.bufferSize);
+ } catch (final IOException ex) {
+ LOGGER.error("TcpSocketManager (" + name + ") " + ex, ex);
+ os = NullOutputStream.NULL_OUTPUT_STREAM;
+ }
+ if (data.reconnectDelayMillis == 0) {
+ return null;
+ }
+ return new TcpSocketManager(name, os, null, inetAddress, data.host, data.port, data.connectTimeoutMillis,
+ data.reconnectDelayMillis, data.immediateFail, data.layout, data.bufferSize);
+ }
+ }
+
+}