Be more resilient to exceptions thrown by listeners when firing events

Author: brenuart

src/main/java/net/logstash/logback/appender/AbstractLogstashTcpSocketAppender.java

@@ -995,32 +995,23 @@ protected Future<?> scheduleReaderCallable(Callable<Void> readerCallable) {
     }
 
     protected void fireEventSent(Socket socket, Event event, long durationInNanos) {
-        for (Listener listener : listeners) {
-            listener.eventSent(this, socket, event, durationInNanos);
-        }
+        safelyFireEvent(l -> l.eventSent(this, socket, event, durationInNanos));
     }
+
     protected void fireEventSendFailure(Event event, Throwable reason) {
-        for (Listener listener : listeners) {
-            listener.eventSendFailure(this, event, reason);
-        }
+        safelyFireEvent(l -> l.eventSendFailure(this, event, reason));
     }
 
     protected void fireConnectionOpened(Socket socket) {
-        for (Listener listener : listeners) {
-            listener.connectionOpened(this, socket);
-        }
+        safelyFireEvent(l -> l.connectionOpened(this, socket));
     }
 
     protected void fireConnectionClosed(Socket socket) {
-        for (Listener listener : listeners) {
-            listener.connectionClosed(this, socket);
-        }
+        safelyFireEvent(l -> l.connectionClosed(this, socket));
     }
 
     protected void fireConnectionFailed(InetSocketAddress address, Throwable throwable) {
-        for (Listener listener : listeners) {
-            listener.connectionFailed(this, address, throwable);
-        }
+        safelyFireEvent(l -> l.connectionFailed(this, address, throwable));
     }
 
 

src/main/java/net/logstash/logback/appender/AsyncDisruptorAppender.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
 
 import net.logstash.logback.appender.listener.AppenderListener;
 import net.logstash.logback.status.LevelFilteringStatusListener;
@@ -486,23 +487,28 @@ protected List<Object> getThreadNameFormatParams() {
     }
 
     protected void fireAppenderStarted() {
-        for (Listener listener : listeners) {
-            listener.appenderStarted(this);
-        }
+        safelyFireEvent(l -> l.appenderStarted(this));
     }
+
     protected void fireAppenderStopped() {
-        for (Listener listener : listeners) {
-            listener.appenderStopped(this);
-        }
+        safelyFireEvent(l -> l.appenderStopped(this));
     }
+
     protected void fireEventAppended(Event event, long durationInNanos) {
-        for (Listener listener : listeners) {
-            listener.eventAppended(this, event, durationInNanos);
-        }
+        safelyFireEvent(l -> l.eventAppended(this, event, durationInNanos));
     }
+
     protected void fireEventAppendFailed(Event event, Throwable reason) {
+        safelyFireEvent(l -> l.eventAppendFailed(this, event, reason));
+    }
+
+    protected void safelyFireEvent(Consumer<Listener> callback) {
         for (Listener listener : listeners) {
-            listener.eventAppendFailed(this, event, reason);
+            try {
+                callback.accept(listener);
+            } catch (Exception e) {
+                addError("Failed to invoke listener " + listener, e);
+            }
         }
     }