From 4b6bac15e52908d6d482aeba27146ef025cca156 Mon Sep 17 00:00:00 2001
From: Nitesh Kant <nitesh1706@gmail.com>
Date: Sat, 25 Jun 2016 19:25:40 -0700
Subject: [PATCH] Ignoring `ClosedChannelException` from the error stream.

Problem

Since `ReactiveSocket` reads and writes from a transport connection, it is completely expected that at some point the underlying connection may have been severed. In such a case, both the read and write will fail on the connection. This pollutes logs of applications that directly log the error stream (as expected).

Modification

Since, this is kind of an "expected" exception, there is no point to send this to the error stream.
This change adds a filter to the error stream to filter any known exceptions (as of today, it is only `ClosedChannelException`)

PS: We currently do not unsubscribe from the input (and stop writing) when `ReactiveSocket` is explicitly closed. This change does not mean, we do not clean that up.
---
 .../reactivesocket/DefaultReactiveSocket.java |  2 +-
 .../io/reactivesocket/KnownErrorFilter.java   | 39 +++++++++++++++++++
 2 files changed, 40 insertions(+), 1 deletion(-)
 create mode 100644 reactivesocket-core/src/main/java/io/reactivesocket/KnownErrorFilter.java

diff --git a/reactivesocket-core/src/main/java/io/reactivesocket/DefaultReactiveSocket.java b/reactivesocket-core/src/main/java/io/reactivesocket/DefaultReactiveSocket.java
index 1ea884980..1eb3a48c2 100644
--- a/reactivesocket-core/src/main/java/io/reactivesocket/DefaultReactiveSocket.java
+++ b/reactivesocket-core/src/main/java/io/reactivesocket/DefaultReactiveSocket.java
@@ -73,7 +73,7 @@ private DefaultReactiveSocket(
         this.clientRequestHandler = clientRequestHandler;
         this.responderConnectionHandler = responderConnectionHandler;
         this.leaseGovernor = leaseGovernor;
-        this.errorStream = errorStream;
+        this.errorStream = new KnownErrorFilter(errorStream);
         this.shutdownListeners = new CopyOnWriteArrayList<>();
     }
 
diff --git a/reactivesocket-core/src/main/java/io/reactivesocket/KnownErrorFilter.java b/reactivesocket-core/src/main/java/io/reactivesocket/KnownErrorFilter.java
new file mode 100644
index 000000000..9c7500598
--- /dev/null
+++ b/reactivesocket-core/src/main/java/io/reactivesocket/KnownErrorFilter.java
@@ -0,0 +1,39 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed 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 io.reactivesocket;
+
+import java.nio.channels.ClosedChannelException;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Consumer;
+
+final class KnownErrorFilter implements Consumer<Throwable> {
+
+    private static final List<Class<? extends Throwable>> knownErrors =
+            Collections.singletonList(ClosedChannelException.class);
+    private final Consumer<Throwable> delegate;
+
+    KnownErrorFilter(Consumer<Throwable> delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void accept(Throwable throwable) {
+        if (!knownErrors.contains(throwable.getClass())) {
+            delegate.accept(throwable);
+        }
+    }
+}