From 057e03ebe2d820a722b6052fcd6a13488fa1326d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Fri, 13 Oct 2017 14:50:00 +0200
Subject: [PATCH 1/9] Make TCP server unbinding observable #23798

---
 .../test/scala/akka/stream/io/TcpSpec.scala   |  4 ++++
 .../scala/akka/stream/impl/io/TcpStages.scala | 24 ++++++++++---------
 .../main/scala/akka/stream/javadsl/Tcp.scala  | 15 ++++++++++--
 .../main/scala/akka/stream/scaladsl/Tcp.scala | 19 ++++++++++-----
 4 files changed, 43 insertions(+), 19 deletions(-)

diff --git a/akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala b/akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala
index 00dad374f0..955ab1b60e 100644
--- a/akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala
+++ b/akka-stream-tests/src/test/scala/akka/stream/io/TcpSpec.scala
@@ -469,9 +469,11 @@ class TcpSpec extends StreamSpec("akka.stream.materializer.subscription-timeout.
       val resultFuture =
         Source(testInput).via(Tcp().outgoingConnection(serverAddress)).runFold(ByteString.empty)((acc, in) ⇒ acc ++ in)
 
+      binding.whenUnbound.value should be(None)
       resultFuture.futureValue should be(expectedOutput)
       binding.unbind().futureValue
       echoServerFinish.futureValue
+      binding.whenUnbound.futureValue should be(Done)
     }
 
     "work with a chain of echoes" in {
@@ -484,6 +486,7 @@ class TcpSpec extends StreamSpec("akka.stream.materializer.subscription-timeout.
 
       // make sure that the server has bound to the socket
       val binding = bindingFuture.futureValue
+      binding.whenUnbound.value should be(None)
 
       val echoConnection = Tcp().outgoingConnection(serverAddress)
 
@@ -501,6 +504,7 @@ class TcpSpec extends StreamSpec("akka.stream.materializer.subscription-timeout.
       resultFuture.futureValue should be(expectedOutput)
       binding.unbind().futureValue
       echoServerFinish.futureValue
+      binding.whenUnbound.futureValue should be(Done)
     }
 
     "bind and unbind correctly" in EventFilter[BindException](occurrences = 2).intercept {
diff --git a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
index 8c06148b55..89d585deff 100644
--- a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
+++ b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
@@ -5,10 +5,10 @@ package akka.stream.impl.io
 
 import java.net.InetSocketAddress
 import java.util.concurrent.TimeoutException
-import java.util.concurrent.atomic.{ AtomicBoolean, AtomicLong }
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
 
-import akka.NotUsed
-import akka.actor.{ ActorRef, Terminated }
+import akka.{Done, NotUsed}
+import akka.actor.{ActorRef, Terminated}
 import akka.annotation.InternalApi
 import akka.dispatch.ExecutionContexts
 import akka.io.Inet.SocketOption
@@ -17,14 +17,14 @@ import akka.io.Tcp._
 import akka.stream._
 import akka.stream.impl.ReactiveStreamsCompliance
 import akka.stream.impl.fusing.GraphStages.detacher
-import akka.stream.scaladsl.Tcp.{ OutgoingConnection, ServerBinding }
-import akka.stream.scaladsl.{ BidiFlow, Flow, TcpIdleTimeoutException, Tcp ⇒ StreamTcp }
+import akka.stream.scaladsl.Tcp.{OutgoingConnection, ServerBinding}
+import akka.stream.scaladsl.{BidiFlow, Flow, TcpIdleTimeoutException, Tcp => StreamTcp}
 import akka.stream.stage._
 import akka.util.ByteString
 
 import scala.collection.immutable
-import scala.concurrent.duration.{ Duration, FiniteDuration }
-import scala.concurrent.{ Future, Promise }
+import scala.concurrent.duration.{Duration, FiniteDuration}
+import scala.concurrent.{Future, Promise}
 
 /**
  * INTERNAL API
@@ -76,7 +76,7 @@ import scala.concurrent.{ Future, Promise }
               // stopped.
               thisStage.tell(Unbind, thisStage)
               unbindPromise.future
-            }))
+            }, unbindPromise.future.map(_ => Done)(ExecutionContexts.sameThreadExecutionContext)))
           case f: CommandFailed ⇒
             val ex = new BindFailedException {
               // cannot modify the actual exception class for compatibility reasons
@@ -84,7 +84,7 @@ import scala.concurrent.{ Future, Promise }
             }
             f.cause.foreach(ex.initCause)
             bindingPromise.failure(ex)
-            unbindPromise.success(() ⇒ Future.successful(()))
+            unbindPromise.failure(ex)
             failStage(ex)
           case c: Connected ⇒
             push(out, connectionFor(c, sender))
@@ -96,8 +96,10 @@ import scala.concurrent.{ Future, Promise }
             if (unbindStarted) {
               unbindCompleted()
             } else {
-              failStage(new IllegalStateException("IO Listener actor terminated unexpectedly for remote endpoint [" +
-                endpoint.getHostString + ":" + endpoint.getPort + "]"))
+              val ex = new IllegalStateException("IO Listener actor terminated unexpectedly for remote endpoint [" +
+                endpoint.getHostString + ":" + endpoint.getPort + "]")
+              unbindPromise.failure(ex)
+              failStage(ex)
             }
         }
       }
diff --git a/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
index e6b13cf0ce..ccd0f4d47e 100644
--- a/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
@@ -3,11 +3,14 @@
  */
 package akka.stream.javadsl
 
-import java.lang.{ Iterable ⇒ JIterable }
+import java.lang.{Iterable => JIterable}
 import java.util.Optional
-import akka.NotUsed
+
+import akka.{Done, NotUsed}
+
 import scala.concurrent.duration._
 import java.net.InetSocketAddress
+
 import akka.actor.ActorSystem
 import akka.actor.ExtendedActorSystem
 import akka.actor.ExtensionId
@@ -17,10 +20,13 @@ import akka.stream.scaladsl
 import akka.util.ByteString
 import akka.japi.Util.immutableSeq
 import akka.io.Inet.SocketOption
+
 import scala.compat.java8.OptionConverters._
 import scala.compat.java8.FutureConverters._
 import java.util.concurrent.CompletionStage
 
+import scala.concurrent.Future
+
 object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
 
   /**
@@ -39,6 +45,11 @@ object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
      * The produced [[java.util.concurrent.CompletionStage]] is fulfilled when the unbinding has been completed.
      */
     def unbind(): CompletionStage[Unit] = delegate.unbind().toJava
+
+    /**
+     * @return A completion stage that is completed when manually unbound, or failed if the server fails
+     */
+    def whenUnbound(): CompletionStage[Done] = delegate.whenUnbound.toJava
   }
 
   /**
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
index c7770ac05a..60d4119718 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
@@ -6,26 +6,33 @@ package akka.stream.scaladsl
 import java.net.InetSocketAddress
 import java.util.concurrent.TimeoutException
 
-import akka.NotUsed
+import akka.{Done, NotUsed}
 import akka.actor._
 import akka.io.Inet.SocketOption
-import akka.io.{ IO, Tcp ⇒ IoTcp }
+import akka.io.{IO, Tcp => IoTcp}
 import akka.stream._
 import akka.stream.impl.fusing.GraphStages.detacher
-import akka.stream.impl.io.{ ConnectionSourceStage, OutgoingConnectionStage, TcpIdleTimeout }
+import akka.stream.impl.io.{ConnectionSourceStage, OutgoingConnectionStage, TcpIdleTimeout}
 import akka.util.ByteString
 
 import scala.collection.immutable
 import scala.concurrent.Future
-import scala.concurrent.duration.{ Duration, FiniteDuration }
+import scala.concurrent.duration.{Duration, FiniteDuration}
 import scala.util.control.NoStackTrace
 
 object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
 
   /**
-   * * Represents a successful TCP server binding.
+   * Represents a successful TCP server binding.
+   *
+   * @param localAddress The address the server was bound to
+   * @param unbindAction a function that will trigger unbind of the server
+   * @param whenUnbound A future that is completed when the server is unbound, or failed if the server binding fails
    */
-  final case class ServerBinding(localAddress: InetSocketAddress)(private val unbindAction: () ⇒ Future[Unit]) {
+  final case class ServerBinding(localAddress: InetSocketAddress)(
+    private val unbindAction: () ⇒ Future[Unit],
+    val whenUnbound: Future[Done]
+  ) {
     def unbind(): Future[Unit] = unbindAction()
   }
 

From 1c35429a9b70eba1694d14c756193b724d14a8f4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Fri, 13 Oct 2017 15:17:46 +0200
Subject: [PATCH 2/9] Formatting

---
 .../scala/akka/stream/impl/io/TcpStages.scala    | 16 ++++++++--------
 .../src/main/scala/akka/stream/javadsl/Tcp.scala |  4 ++--
 .../main/scala/akka/stream/scaladsl/Tcp.scala    | 10 +++++-----
 3 files changed, 15 insertions(+), 15 deletions(-)

diff --git a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
index 89d585deff..468c948df3 100644
--- a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
+++ b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
@@ -5,10 +5,10 @@ package akka.stream.impl.io
 
 import java.net.InetSocketAddress
 import java.util.concurrent.TimeoutException
-import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicLong }
 
-import akka.{Done, NotUsed}
-import akka.actor.{ActorRef, Terminated}
+import akka.{ Done, NotUsed }
+import akka.actor.{ ActorRef, Terminated }
 import akka.annotation.InternalApi
 import akka.dispatch.ExecutionContexts
 import akka.io.Inet.SocketOption
@@ -17,14 +17,14 @@ import akka.io.Tcp._
 import akka.stream._
 import akka.stream.impl.ReactiveStreamsCompliance
 import akka.stream.impl.fusing.GraphStages.detacher
-import akka.stream.scaladsl.Tcp.{OutgoingConnection, ServerBinding}
-import akka.stream.scaladsl.{BidiFlow, Flow, TcpIdleTimeoutException, Tcp => StreamTcp}
+import akka.stream.scaladsl.Tcp.{ OutgoingConnection, ServerBinding }
+import akka.stream.scaladsl.{ BidiFlow, Flow, TcpIdleTimeoutException, Tcp ⇒ StreamTcp }
 import akka.stream.stage._
 import akka.util.ByteString
 
 import scala.collection.immutable
-import scala.concurrent.duration.{Duration, FiniteDuration}
-import scala.concurrent.{Future, Promise}
+import scala.concurrent.duration.{ Duration, FiniteDuration }
+import scala.concurrent.{ Future, Promise }
 
 /**
  * INTERNAL API
@@ -76,7 +76,7 @@ import scala.concurrent.{Future, Promise}
               // stopped.
               thisStage.tell(Unbind, thisStage)
               unbindPromise.future
-            }, unbindPromise.future.map(_ => Done)(ExecutionContexts.sameThreadExecutionContext)))
+            }, unbindPromise.future.map(_ ⇒ Done)(ExecutionContexts.sameThreadExecutionContext)))
           case f: CommandFailed ⇒
             val ex = new BindFailedException {
               // cannot modify the actual exception class for compatibility reasons
diff --git a/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
index ccd0f4d47e..7919f848a0 100644
--- a/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
@@ -3,10 +3,10 @@
  */
 package akka.stream.javadsl
 
-import java.lang.{Iterable => JIterable}
+import java.lang.{ Iterable ⇒ JIterable }
 import java.util.Optional
 
-import akka.{Done, NotUsed}
+import akka.{ Done, NotUsed }
 
 import scala.concurrent.duration._
 import java.net.InetSocketAddress
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
index 60d4119718..b1f83c57f4 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
@@ -6,18 +6,18 @@ package akka.stream.scaladsl
 import java.net.InetSocketAddress
 import java.util.concurrent.TimeoutException
 
-import akka.{Done, NotUsed}
+import akka.{ Done, NotUsed }
 import akka.actor._
 import akka.io.Inet.SocketOption
-import akka.io.{IO, Tcp => IoTcp}
+import akka.io.{ IO, Tcp ⇒ IoTcp }
 import akka.stream._
 import akka.stream.impl.fusing.GraphStages.detacher
-import akka.stream.impl.io.{ConnectionSourceStage, OutgoingConnectionStage, TcpIdleTimeout}
+import akka.stream.impl.io.{ ConnectionSourceStage, OutgoingConnectionStage, TcpIdleTimeout }
 import akka.util.ByteString
 
 import scala.collection.immutable
 import scala.concurrent.Future
-import scala.concurrent.duration.{Duration, FiniteDuration}
+import scala.concurrent.duration.{ Duration, FiniteDuration }
 import scala.util.control.NoStackTrace
 
 object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
@@ -31,7 +31,7 @@ object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
    */
   final case class ServerBinding(localAddress: InetSocketAddress)(
     private val unbindAction: () ⇒ Future[Unit],
-    val whenUnbound: Future[Done]
+    val whenUnbound:          Future[Done]
   ) {
     def unbind(): Future[Unit] = unbindAction()
   }

From c830a75bd049d6f5d33bbd48fdf44e2c6b3c3da8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Fri, 13 Oct 2017 17:18:28 +0200
Subject: [PATCH 3/9] Mima and internalizing things that should have already
 been internal

---
 akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala  | 6 ++++--
 akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala | 5 ++++-
 2 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
index 7919f848a0..eec6b899a8 100644
--- a/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/javadsl/Tcp.scala
@@ -25,14 +25,16 @@ import scala.compat.java8.OptionConverters._
 import scala.compat.java8.FutureConverters._
 import java.util.concurrent.CompletionStage
 
-import scala.concurrent.Future
+import akka.annotation.InternalApi
 
 object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
 
   /**
    * Represents a prospective TCP server binding.
+   *
+   * Not indented for user construction
    */
-  class ServerBinding private[akka] (delegate: scaladsl.Tcp.ServerBinding) {
+  final class ServerBinding @InternalApi private[akka] (delegate: scaladsl.Tcp.ServerBinding) {
     /**
      * The local address of the endpoint bound by the materialization of the `connections` [[Source]].
      */
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
index b1f83c57f4..323221d539 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
@@ -8,6 +8,7 @@ import java.util.concurrent.TimeoutException
 
 import akka.{ Done, NotUsed }
 import akka.actor._
+import akka.annotation.InternalApi
 import akka.io.Inet.SocketOption
 import akka.io.{ IO, Tcp ⇒ IoTcp }
 import akka.stream._
@@ -28,8 +29,10 @@ object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
    * @param localAddress The address the server was bound to
    * @param unbindAction a function that will trigger unbind of the server
    * @param whenUnbound A future that is completed when the server is unbound, or failed if the server binding fails
+   *
+   * Not indented for user construction
    */
-  final case class ServerBinding(localAddress: InetSocketAddress)(
+  final case class ServerBinding @InternalApi private[akka] (localAddress: InetSocketAddress)(
     private val unbindAction: () ⇒ Future[Unit],
     val whenUnbound:          Future[Done]
   ) {

From 4c1410c3b55398631d4d8ec38707c9823a6bd9a5 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Tue, 31 Oct 2017 14:19:34 +0100
Subject: [PATCH 4/9] Complete unbind promise earlier if possible and always in
 postStop

---
 .../src/main/scala/akka/stream/impl/io/TcpStages.scala       | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
index 468c948df3..5c001bf9ef 100644
--- a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
+++ b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
@@ -146,6 +146,7 @@ import scala.concurrent.{ Future, Promise }
 
       private def unbindCompleted(): Unit = {
         stageActor.unwatch(listener)
+        unbindPromise.trySuccess(Done)
         if (connectionFlowsAwaitingInitialization.get() == 0) completeStage()
         else scheduleOnce(BindShutdownTimer, bindShutdownTimeout)
       }
@@ -156,7 +157,9 @@ import scala.concurrent.{ Future, Promise }
       }
 
       override def postStop(): Unit = {
-        unbindPromise.trySuccess(())
+        // a bit unexpected to succeed here rather than fail with abrupt stage termination
+        // but there was an existing test case covering this behavior
+        unbindPromise.trySuccess(Done)
         bindingPromise.tryFailure(new NoSuchElementException("Binding was unbound before it was completely finished"))
       }
     }

From 7bdd6c893412cac6b86feb5d3a1069c7770c0f33 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Wed, 1 Nov 2017 20:51:06 +0100
Subject: [PATCH 5/9] Put those @params last

---
 akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
index 323221d539..68bd0a5ce2 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
+++ b/akka-stream/src/main/scala/akka/stream/scaladsl/Tcp.scala
@@ -26,11 +26,11 @@ object Tcp extends ExtensionId[Tcp] with ExtensionIdProvider {
   /**
    * Represents a successful TCP server binding.
    *
+   * Not indented for user construction
+   *
    * @param localAddress The address the server was bound to
    * @param unbindAction a function that will trigger unbind of the server
    * @param whenUnbound A future that is completed when the server is unbound, or failed if the server binding fails
-   *
-   * Not indented for user construction
    */
   final case class ServerBinding @InternalApi private[akka] (localAddress: InetSocketAddress)(
     private val unbindAction: () ⇒ Future[Unit],

From bdfcbc28da14b6ccf103d58a248334adbe65d854 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Tue, 14 Nov 2017 11:32:48 +0100
Subject: [PATCH 6/9] Try fail in case the unbind promise was already completed

---
 akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
index 5c001bf9ef..afdb151981 100644
--- a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
+++ b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
@@ -98,7 +98,7 @@ import scala.concurrent.{ Future, Promise }
             } else {
               val ex = new IllegalStateException("IO Listener actor terminated unexpectedly for remote endpoint [" +
                 endpoint.getHostString + ":" + endpoint.getPort + "]")
-              unbindPromise.failure(ex)
+              unbindPromise.tryFailure(ex)
               failStage(ex)
             }
         }

From acdabeff7447ca05fa12e182921bdd023e7108c1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Tue, 14 Nov 2017 11:36:55 +0100
Subject: [PATCH 7/9] Mima

---
 akka-stream/src/main/mima-filters/2.5.7.backwards.excludes | 5 +++++
 1 file changed, 5 insertions(+)
 create mode 100644 akka-stream/src/main/mima-filters/2.5.7.backwards.excludes

diff --git a/akka-stream/src/main/mima-filters/2.5.7.backwards.excludes b/akka-stream/src/main/mima-filters/2.5.7.backwards.excludes
new file mode 100644
index 0000000000..f34727e353
--- /dev/null
+++ b/akka-stream/src/main/mima-filters/2.5.7.backwards.excludes
@@ -0,0 +1,5 @@
+# 23798 observable unbind on stream Tcp server
+ProblemFilters.exclude[FinalClassProblem]("akka.stream.javadsl.Tcp$ServerBinding")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.copy")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.this")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.apply")
\ No newline at end of file

From ea76a0c4d33d62749ff99f39f7bd8e95e79d4a13 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Tue, 14 Nov 2017 11:51:02 +0100
Subject: [PATCH 8/9] One more tryFailure

---
 akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
index afdb151981..9e621f8da5 100644
--- a/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
+++ b/akka-stream/src/main/scala/akka/stream/impl/io/TcpStages.scala
@@ -84,7 +84,7 @@ import scala.concurrent.{ Future, Promise }
             }
             f.cause.foreach(ex.initCause)
             bindingPromise.failure(ex)
-            unbindPromise.failure(ex)
+            unbindPromise.tryFailure(ex)
             failStage(ex)
           case c: Connected ⇒
             push(out, connectionFor(c, sender))

From b315cb787ceab5ed0c8fafa95beccce42117efe7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Johan=20Andr=C3=A9n?= <johan@markatta.com>
Date: Fri, 17 Nov 2017 11:01:37 +0100
Subject: [PATCH 9/9] The right place for MiMa excludes

---
 .../src/main/mima-filters/2.5.6.backwards.excludes        | 8 +++++++-
 .../src/main/mima-filters/2.5.7.backwards.excludes        | 5 -----
 2 files changed, 7 insertions(+), 6 deletions(-)
 delete mode 100644 akka-stream/src/main/mima-filters/2.5.7.backwards.excludes

diff --git a/akka-stream/src/main/mima-filters/2.5.6.backwards.excludes b/akka-stream/src/main/mima-filters/2.5.6.backwards.excludes
index fe984a3f4a..d5c50598e6 100644
--- a/akka-stream/src/main/mima-filters/2.5.6.backwards.excludes
+++ b/akka-stream/src/main/mima-filters/2.5.6.backwards.excludes
@@ -30,4 +30,10 @@ ProblemFilters.exclude[IncompatibleMethTypeProblem]("akka.stream.impl.fusing.Gra
 ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.fusing.GraphInterpreterShell#AsyncInput.apply")
 ProblemFilters.exclude[IncompatibleResultTypeProblem]("akka.stream.impl.fusing.GraphInterpreterShell#AsyncInput.copy$default$4")
 ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.fusing.GraphInterpreterShell#AsyncInput.this")
-ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.fusing.GraphInterpreter.runAsyncInput")
\ No newline at end of file
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.fusing.GraphInterpreter.runAsyncInput")
+
+# 23798 observable unbind on stream Tcp server
+ProblemFilters.exclude[FinalClassProblem]("akka.stream.javadsl.Tcp$ServerBinding")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.copy")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.this")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.apply")
diff --git a/akka-stream/src/main/mima-filters/2.5.7.backwards.excludes b/akka-stream/src/main/mima-filters/2.5.7.backwards.excludes
deleted file mode 100644
index f34727e353..0000000000
--- a/akka-stream/src/main/mima-filters/2.5.7.backwards.excludes
+++ /dev/null
@@ -1,5 +0,0 @@
-# 23798 observable unbind on stream Tcp server
-ProblemFilters.exclude[FinalClassProblem]("akka.stream.javadsl.Tcp$ServerBinding")
-ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.copy")
-ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.this")
-ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.scaladsl.Tcp#ServerBinding.apply")
\ No newline at end of file