compile issues with watchTermination()

Author: pjfanning

docs/src/test/scala/docs/http/scaladsl/HttpServerExampleSpec.scala

@@ -111,7 +111,7 @@ class HttpServerExampleSpec extends AnyWordSpec with Matchers
     val failureMonitor: ActorRef = system.actorOf(MyExampleMonitoringActor.props)
 
     val reactToTopLevelFailures = Flow[IncomingConnection]
-      .watchTermination()((_, termination) =>
+      .watchTermination((_, termination) =>
         termination.failed.foreach {
           cause => failureMonitor ! cause
         })

http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala

@@ -106,7 +106,7 @@ private[http] final class Http2Ext(implicit val system: ActorSystem)
         (incoming: Tcp.IncomingConnection) =>
           try {
             httpPlusSwitching(http1, http2).addAttributes(prepareServerAttributes(settings, incoming))
-              .watchTermination() {
+              .watchTermination {
                 case (connectionTerminatorF, future) =>
                   connectionTerminatorF.foreach { connectionTerminator =>
                     masterTerminator.registerConnection(connectionTerminator)(fm.executionContext)
@@ -170,7 +170,7 @@ private[http] final class Http2Ext(implicit val system: ActorSystem)
 
             val serverLayer: Flow[ByteString, ByteString, Future[Done]] = Flow.fromGraph(
               Flow[HttpRequest]
-                .watchTermination()(Keep.right)
+                .watchTermination(Keep.right)
                 .prepend(injectedRequest)
                 .via(Http2Blueprint.handleWithStreamIdHeader(settings.http2Settings.maxConcurrentStreams)(handler)(
                   system.dispatcher))

http-core/src/main/scala/org/apache/pekko/http/scaladsl/Http.scala

@@ -121,9 +121,9 @@ class HttpExt @InternalStableApi /* constructor signature is hardcoded in Teleme
       handler: Flow[HttpRequest, HttpResponse, Any]): ServerLayerFlow =
     Flow.fromGraph(
       Flow[HttpRequest]
-        .watchTermination()(Keep.right)
+        .watchTermination(Keep.right)
         .via(handler)
-        .watchTermination() { (termWatchBefore, termWatchAfter) =>
+        .watchTermination { (termWatchBefore, termWatchAfter) =>
           // flag termination when the user handler has gotten (or has emitted) termination
           // signals in both directions
           termWatchBefore.flatMap(_ => termWatchAfter)(ExecutionContext.parasitic)
@@ -234,7 +234,7 @@ class HttpExt @InternalStableApi /* constructor signature is hardcoded in Teleme
       .mapAsyncUnordered(settings.maxConnections) { incoming =>
         try {
           fullLayer
-            .watchTermination() {
+            .watchTermination {
               case ((done, connectionTerminator), whenTerminates) =>
                 whenTerminates.onComplete { _ =>
                   masterTerminator.removeConnection(connectionTerminator)

http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HostConnectionPoolSpec.scala

@@ -700,7 +700,7 @@ class HostConnectionPoolSpec extends PekkoSpecWithMaterializer(
               Sink.fromSubscriber(serverRequests),
               Source.fromPublisher(serverResponses))
               .joinMat(clientServerImplementation.get(killSwitch))(Keep.right)
-              .watchTermination()(Keep.both)
+              .watchTermination(Keep.both)
               .join(
                 Flow.fromSinkAndSource(
                   Sink.fromSubscriber(responseSubscriber),

http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WSClientAutobahnTest.scala

@@ -173,7 +173,7 @@ object WSClientAutobahnTest extends App {
     Http().singleWebSocketRequest(uri, clientFlow)._2
 
   def completionSignal[T]: Flow[T, T, Future[Done]] =
-    Flow[T].watchTermination()((_, res) => res)
+    Flow[T].watchTermination((_, res) => res)
 
   /**
    * The autobahn tests define a weird API where every request must be a WebSocket request and

http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketIntegrationSpec.scala

@@ -210,7 +210,7 @@ class WebSocketIntegrationSpec extends PekkoSpecWithMaterializer(
       val handlerTermination = Promise[Done]()
 
       val handler = Flow[Message]
-        .watchTermination()(Keep.right)
+        .watchTermination(Keep.right)
         .mapMaterializedValue(handlerTermination.completeWith(_))
         .map(m => TextMessage.Strict(s"Echo [${m.asTextMessage.getStrictText}]"))
 

http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/TelemetrySpiSpec.scala

@@ -126,7 +126,7 @@ abstract class TelemetrySpiSpec(useTls: Boolean) extends PekkoSpecWithMaterializ
               attrs.get[TelemetryAttributes.ClientMeta].foreach(probe.ref ! _)
               probe.ref ! requestId
               request.addAttribute(requestIdAttr, requestId).addHeader(headers.RawHeader("request-id", requestId.id))
-            }.watchTermination() { (_, done) =>
+            }.watchTermination { (_, done) =>
               done.onComplete {
                 case Success(_) => probe.ref ! "close-seen" // this is the expected case
                 case Failure(t) => probe.ref ! t.getMessage // useful to diagnose cases where there's a failure
@@ -193,7 +193,7 @@ abstract class TelemetrySpiSpec(useTls: Boolean) extends PekkoSpecWithMaterializ
               telemetryProbe.ref ! "connection-seen"
               telemetryProbe.ref ! connId
               conn.copy(flow = conn.flow.addAttributes(Attributes(connId)))
-            }.watchTermination() { (notUsed, done) =>
+            }.watchTermination { (notUsed, done) =>
               done.onComplete(_ => telemetryProbe.ref ! "unbind-seen")(system.dispatcher)
               notUsed
             }
@@ -203,7 +203,7 @@ abstract class TelemetrySpiSpec(useTls: Boolean) extends PekkoSpecWithMaterializ
             Flow[HttpResponse].map { response =>
               telemetryProbe.ref ! "response-seen"
               response
-            }.watchTermination() { (_, done) =>
+            }.watchTermination { (_, done) =>
               done.foreach(_ => telemetryProbe.ref ! "close-seen")(system.dispatcher)
             },
             StreamUtils.statefulAttrsMap[HttpRequest, HttpRequest](attributes => { request =>