Migrate the classic transport to Netty 4.

Signed-off-by: He-Pin <[email protected]>

.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.15.1"
-  val nettyVersion = "3.10.6.Final"
   val netty4Version = "4.1.94.Final"
   val protobufJavaVersion = "3.16.3"
   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/src/main/scala/org/apache/pekko/remote/RemoteActorRefProvider.scala

@@ -13,13 +13,12 @@
 
 package org.apache.pekko.remote
 
+import scala.annotation.nowarn
 import scala.concurrent.Future
 import scala.util.Failure
 import scala.util.control.Exception.Catcher
 import scala.util.control.NonFatal
 
-import scala.annotation.nowarn
-
 import org.apache.pekko
 import pekko.ConfigurationException
 import pekko.Done
@@ -281,7 +280,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

@@ -15,34 +15,33 @@ package org.apache.pekko.remote.transport.netty
 
 import java.nio.channels.ClosedChannelException
 
+import scala.annotation.nowarn
 import scala.util.control.NonFatal
 
-import org.jboss.netty.channel._
-
 import org.apache.pekko
 import pekko.PekkoException
 import pekko.util.unused
 
+import io.netty.channel.{ ChannelHandlerContext, ChannelInboundHandlerAdapter }
+
 /**
  * INTERNAL API
  */
 private[netty] trait NettyHelpers {
 
-  protected def onConnect(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
-
-  protected def onDisconnect(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
+  protected def onActive(@unused ctx: ChannelHandlerContext): Unit = ()
 
-  protected def onOpen(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
+  protected def onInactive(@unused ctx: ChannelHandlerContext): Unit = ()
 
-  protected def onMessage(@unused ctx: ChannelHandlerContext, @unused e: MessageEvent): Unit = ()
+  protected def onMessage(@unused ctx: ChannelHandlerContext, @unused msg: Any): Unit = ()
 
-  protected def onException(@unused ctx: ChannelHandlerContext, @unused e: ExceptionEvent): Unit = ()
+  protected def onException(@unused ctx: ChannelHandlerContext, @unused cause: 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, exception: Throwable): Unit = {
+    val cause = if (exception.getCause ne null) exception.getCause else new PekkoException("Unknown cause")
     cause match {
       case _: ClosedChannelException => // Ignore
-      case null | NonFatal(_)        => onException(ctx, ev)
+      case null | NonFatal(_)        => onException(ctx, exception)
       case e: Throwable              => throw e // Rethrow fatals
     }
   }
@@ -51,54 +50,51 @@ private[netty] trait NettyHelpers {
 /**
  * INTERNAL API
  */
-private[netty] trait NettyServerHelpers extends SimpleChannelUpstreamHandler with NettyHelpers {
+private[netty] trait NettyServerHelpers extends ChannelInboundHandlerAdapter with NettyHelpers {
 
-  final override def messageReceived(ctx: ChannelHandlerContext, e: MessageEvent): Unit = {
-    super.messageReceived(ctx, e)
-    onMessage(ctx, e)
+  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = {
+    super.channelRead(ctx, msg)
+    onMessage(ctx, msg)
   }
 
-  final override def exceptionCaught(ctx: ChannelHandlerContext, e: ExceptionEvent): Unit = transformException(ctx, e)
-
-  final override def channelConnected(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelConnected(ctx, e)
-    onConnect(ctx, e)
+  @nowarn("msg=deprecated")
+  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)
+  override def channelActive(ctx: ChannelHandlerContext): Unit = {
+    super.channelActive(ctx)
+    onActive(ctx)
   }
 
-  final override def channelDisconnected(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelDisconnected(ctx, e)
-    onDisconnect(ctx, e)
+  override def channelInactive(ctx: ChannelHandlerContext): Unit = {
+    super.channelInactive(ctx)
+    onInactive(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 ChannelInboundHandlerAdapter with NettyHelpers {
 
-  final override def exceptionCaught(ctx: ChannelHandlerContext, e: ExceptionEvent): Unit = transformException(ctx, e)
+  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = {
+    super.channelRead(ctx, msg)
+    onMessage(ctx, msg)
+  }
 
-  final override def channelConnected(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelConnected(ctx, e)
-    onConnect(ctx, e)
+  @nowarn("msg=deprecated")
+  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)
+  override def channelActive(ctx: ChannelHandlerContext): Unit = {
+    super.channelActive(ctx)
+    onActive(ctx)
   }
 
-  final override def channelDisconnected(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = {
-    super.channelDisconnected(ctx, e)
-    onDisconnect(ctx, e)
+  override def channelInactive(ctx: ChannelHandlerContext): Unit = {
+    super.channelInactive(ctx)
+    onInactive(ctx)
   }
 }

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

@@ -14,13 +14,14 @@
 package org.apache.pekko.remote.transport.netty
 
 import scala.annotation.nowarn
-import com.typesafe.config.Config
-import org.jboss.netty.handler.ssl.SslHandler
 
+import com.typesafe.config.Config
 import org.apache.pekko
 import pekko.japi.Util._
 import pekko.util.ccompat._
 
+import io.netty.handler.ssl.SslHandler
+
 /**
  * INTERNAL API
  */