!remoting Update classic transport from Netty 3 to netty4

.scalafix.conf

@@ -48,4 +48,5 @@ SortImports.blocks = [
   "com.sun."
   "org.apache.pekko."
   "org.reactivestreams."
+  "io.netty."
 ]

project/Dependencies.scala

@@ -28,7 +28,6 @@ object Dependencies {
   // needs to be inline with the aeron version, check
   // https://github.com/real-logic/aeron/blob/1.x.y/build.gradle
   val agronaVersion = "1.19.2"
-  val nettyVersion = "3.10.6.Final"
   val netty4Version = "4.1.96.Final"
   val protobufJavaVersion = "3.19.6"
   val logbackVersion = "1.2.11"
@@ -60,7 +59,6 @@ object Dependencies {
     // Compile
 
     val config = "com.typesafe" % "config" % "1.4.2"
-    val netty = "io.netty" % "netty" % nettyVersion
     val `netty-transport` = "io.netty" % "netty-transport" % netty4Version
     val `netty-handler` = "io.netty" % "netty-handler" % netty4Version
 
@@ -278,7 +276,7 @@ object Dependencies {
       Compile.slf4jApi,
       TestDependencies.scalatest.value)
 
-  val remoteDependencies = Seq(netty, aeronDriver, aeronClient)
+  val remoteDependencies = Seq(`netty-transport`, `netty-handler`, aeronDriver, aeronClient)
   val remoteOptionalDependencies = remoteDependencies.map(_ % "optional")
 
   val remote = l ++= Seq(

project/Paradox.scala

@@ -66,7 +66,7 @@ object Paradox {
       "scalatest.version" -> Dependencies.scalaTestVersion,
       "sigar_loader.version" -> "1.6.6-rev002",
       "aeron_version" -> Dependencies.aeronVersion,
-      "netty_version" -> Dependencies.nettyVersion,
+      "netty_version" -> Dependencies.netty4Version,
       "logback_version" -> Dependencies.logbackVersion))
 
   val rootsSettings = Seq(

remote-tests/src/test/scala/org/apache/pekko/remote/classic/RemotingFailedToBindSpec.scala

@@ -14,14 +14,14 @@
 package org.apache.pekko.remote.classic
 
 import com.typesafe.config.ConfigFactory
-import org.jboss.netty.channel.ChannelException
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpec
-
 import org.apache.pekko
 import pekko.actor.ActorSystem
 import pekko.testkit.SocketUtil
 
+import java.net.BindException
+
 class RemotingFailedToBindSpec extends AnyWordSpec with Matchers {
 
   "an ActorSystem" must {
@@ -43,10 +43,10 @@ class RemotingFailedToBindSpec extends AnyWordSpec with Matchers {
        """.stripMargin)
       val as = ActorSystem("RemotingFailedToBindSpec", config)
       try {
-        val ex = intercept[ChannelException] {
+        val ex = intercept[BindException] {
           ActorSystem("BindTest2", config)
         }
-        ex.getMessage should startWith("Failed to bind")
+        ex.getMessage should startWith("Address already in use")
       } finally {
         as.terminate()
       }

remote/src/main/mima-filters/1.1.x.backwards.excludes/netty4.backwards.excludes

@@ -0,0 +1,2 @@
+#migrate the classic transport to Netty4
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.remote.transport.netty.NettyFutureBridge.apply")

remote/src/main/scala/org/apache/pekko/remote/RemoteActorRefProvider.scala

@@ -281,7 +281,7 @@ private[pekko] class RemoteActorRefProvider(
   private def checkNettyOnClassPath(system: ActorSystemImpl): Unit = {
     checkClassOrThrow(
       system,
-      "org.jboss.netty.channel.Channel",
+      "io.netty.channel.Channel",
       "Classic",
       "Netty",
       "https://pekko.apache.org/docs/pekko/current/remoting.html")

remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyHelpers.scala

@@ -13,36 +13,37 @@
 
 package org.apache.pekko.remote.transport.netty
 
-import java.nio.channels.ClosedChannelException
+import io.netty.buffer.ByteBuf
+import io.netty.channel.{ ChannelHandlerContext, SimpleChannelInboundHandler }
 
+import java.nio.channels.ClosedChannelException
 import scala.util.control.NonFatal
-
-import org.jboss.netty.channel._
-
 import org.apache.pekko
 import pekko.PekkoException
 import pekko.util.unused
 
+import scala.annotation.nowarn
+
 /**
  * INTERNAL API
  */
 private[netty] trait NettyHelpers {
 
-  protected def onConnect(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
+  protected def onConnect(@unused ctx: ChannelHandlerContext): Unit = ()
 
-  protected def onDisconnect(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
+  protected def onDisconnect(@unused ctx: ChannelHandlerContext): Unit = ()
 
-  protected def onOpen(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
+  protected def onOpen(@unused ctx: ChannelHandlerContext): Unit = ()
 
-  protected def onMessage(@unused ctx: ChannelHandlerContext, @unused e: MessageEvent): Unit = ()
+  protected def onMessage(@unused ctx: ChannelHandlerContext, @unused msg: ByteBuf): Unit = ()
 
-  protected def onException(@unused ctx: ChannelHandlerContext, @unused e: ExceptionEvent): Unit = ()
+  protected def onException(@unused ctx: ChannelHandlerContext, @unused e: Throwable): Unit = ()
 
-  final protected def transformException(ctx: ChannelHandlerContext, ev: ExceptionEvent): Unit = {
-    val cause = if (ev.getCause ne null) ev.getCause else new PekkoException("Unknown cause")
+  final protected def transformException(ctx: ChannelHandlerContext, ex: Throwable): Unit = {
+    val cause = if (ex ne null) ex else new PekkoException("Unknown cause")
     cause match {
       case _: ClosedChannelException => // Ignore
-      case null | NonFatal(_)        => onException(ctx, ev)
+      case null | NonFatal(_)        => onException(ctx, ex)
       case e: Throwable              => throw e // Rethrow fatals
     }
   }
@@ -51,54 +52,46 @@ private[netty] trait NettyHelpers {
 /**
  * INTERNAL API
  */
-private[netty] trait NettyServerHelpers extends SimpleChannelUpstreamHandler with NettyHelpers {
+private[netty] trait NettyServerHelpers extends SimpleChannelInboundHandler[ByteBuf] with NettyHelpers {
 
-  final override def messageReceived(ctx: ChannelHandlerContext, e: MessageEvent): Unit = {
-    super.messageReceived(ctx, e)
-    onMessage(ctx, e)
+  final override def channelRead0(ctx: ChannelHandlerContext, msg: ByteBuf): Unit = {
+    onMessage(ctx, msg)
   }
 
-  final override def exceptionCaught(ctx: ChannelHandlerContext, e: ExceptionEvent): Unit = transformException(ctx, e)
+  @nowarn("msg=deprecated")
+  final override def exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable): Unit =
+    transformException(ctx, cause)
 
-  final override def channelConnected(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelConnected(ctx, e)
-    onConnect(ctx, e)
+  final override def channelActive(ctx: ChannelHandlerContext): Unit = {
+    onOpen(ctx)
+    onConnect(ctx)
   }
 
-  final override def channelOpen(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelOpen(ctx, e)
-    onOpen(ctx, e)
-  }
-
-  final override def channelDisconnected(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelDisconnected(ctx, e)
-    onDisconnect(ctx, e)
+  final override def channelInactive(ctx: ChannelHandlerContext): Unit = {
+    onDisconnect(ctx)
   }
 }
 
 /**
  * INTERNAL API
  */
-private[netty] trait NettyClientHelpers extends SimpleChannelHandler with NettyHelpers {
-  final override def messageReceived(ctx: ChannelHandlerContext, e: MessageEvent): Unit = {
-    super.messageReceived(ctx, e)
-    onMessage(ctx, e)
-  }
+private[netty] trait NettyClientHelpers extends SimpleChannelInboundHandler[ByteBuf] with NettyHelpers {
 
-  final override def exceptionCaught(ctx: ChannelHandlerContext, e: ExceptionEvent): Unit = transformException(ctx, e)
+  final override def channelRead0(ctx: ChannelHandlerContext, msg: ByteBuf): Unit = {
+    onMessage(ctx, msg)
+  }
 
-  final override def channelConnected(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelConnected(ctx, e)
-    onConnect(ctx, e)
+  @nowarn("msg=deprecated")
+  final override def exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable): Unit = {
+    transformException(ctx, cause)
   }
 
-  final override def channelOpen(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelOpen(ctx, e)
-    onOpen(ctx, e)
+  final override def channelActive(ctx: ChannelHandlerContext): Unit = {
+    onOpen(ctx)
+    onConnect(ctx)
   }
 
-  final override def channelDisconnected(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelDisconnected(ctx, e)
-    onDisconnect(ctx, e)
+  final override def channelInactive(ctx: ChannelHandlerContext): Unit = {
+    onDisconnect(ctx)
   }
 }

remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettySSLSupport.scala

@@ -15,8 +15,9 @@ package org.apache.pekko.remote.transport.netty
 
 import scala.annotation.nowarn
 import com.typesafe.config.Config
-import org.jboss.netty.handler.ssl.SslHandler
-
+import io.netty.channel.Channel
+import io.netty.handler.ssl.SslHandler
+import io.netty.util.concurrent.{ Future, GenericFutureListener }
 import org.apache.pekko
 import pekko.japi.Util._
 import pekko.util.ccompat._
@@ -64,6 +65,14 @@ private[pekko] object NettySSLSupport {
     val sslEngine =
       if (isClient) sslEngineProvider.createClientSSLEngine()
       else sslEngineProvider.createServerSSLEngine()
-    new SslHandler(sslEngine)
+    val handler = new SslHandler(sslEngine)
+    handler.handshakeFuture().addListener(new GenericFutureListener[Future[Channel]] {
+      override def operationComplete(future: Future[Channel]): Unit = {
+        if (!future.isSuccess) {
+          handler.closeOutbound().channel().close()
+        }
+      }
+    })
+    handler
   }
 }