You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2025/03/10 22:56:45 UTC
(pekko) branch pjfanning-patch-1 updated: refactor
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch pjfanning-patch-1
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/pjfanning-patch-1 by this push:
new e3939e398e refactor
e3939e398e is described below
commit e3939e398e2f8b7f7bd5c5bd37091620c11da18b
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Mon Mar 10 23:56:33 2025 +0100
refactor
---
.../apache/pekko/remote/testconductor/RemoteConnection.scala | 11 +++++------
.../apache/pekko/remote/transport/netty/NettyTransport.scala | 10 +++++-----
2 files changed, 10 insertions(+), 11 deletions(-)
diff --git a/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala b/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala
index 7dca43b1a1..db5fc70066 100644
--- a/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala
+++ b/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala
@@ -16,14 +16,13 @@ package org.apache.pekko.remote.testconductor
import java.net.InetSocketAddress
import java.util.concurrent.TimeUnit
-import scala.annotation.nowarn
import scala.util.control.NonFatal
import io.netty.bootstrap.{ Bootstrap, ServerBootstrap }
import io.netty.buffer.{ ByteBuf, ByteBufUtil }
import io.netty.channel._
import io.netty.channel.ChannelHandler.Sharable
-import io.netty.channel.nio.NioEventLoopGroup
+import io.netty.channel.nio.NioIoHandler
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.{ NioServerSocketChannel, NioSocketChannel }
import io.netty.handler.codec.{
@@ -115,7 +114,6 @@ private[pekko] trait RemoteConnection {
/**
* INTERNAL API.
*/
-@nowarn("msg=deprecated")
private[pekko] object RemoteConnection {
def apply(
role: Role,
@@ -125,7 +123,7 @@ private[pekko] object RemoteConnection {
role match {
case Client =>
val bootstrap = new Bootstrap()
- val eventLoopGroup = new NioEventLoopGroup(poolSize)
+ val eventLoopGroup = new MultiThreadIoEventLoopGroup(poolSize, NioIoHandler.newFactory())
val cf = bootstrap
.group(eventLoopGroup)
.channel(classOf[NioSocketChannel])
@@ -150,8 +148,9 @@ private[pekko] object RemoteConnection {
case Server =>
val bootstrap = new ServerBootstrap()
- val parentEventLoopGroup = new NioEventLoopGroup(poolSize)
- val childEventLoopGroup = new NioEventLoopGroup(poolSize)
+ val ioHandlerFactory = NioIoHandler.newFactory()
+ val parentEventLoopGroup = new MultiThreadIoEventLoopGroup(poolSize, ioHandlerFactory)
+ val childEventLoopGroup = new MultiThreadIoEventLoopGroup(poolSize, ioHandlerFactory)
val cf = bootstrap
.group(parentEventLoopGroup, childEventLoopGroup)
.channel(classOf[NioServerSocketChannel])
diff --git a/remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyTransport.scala b/remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyTransport.scala
index 78e1095f34..4548c7ff89 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyTransport.scala
@@ -53,10 +53,11 @@ import io.netty.channel.{
ChannelInitializer,
ChannelOption,
ChannelPipeline,
- EventLoopGroup
+ EventLoopGroup,
+ MultiThreadIoEventLoopGroup
}
import io.netty.channel.group.{ ChannelGroup, ChannelGroupFuture, ChannelMatchers, DefaultChannelGroup }
-import io.netty.channel.nio.NioEventLoopGroup
+import io.netty.channel.nio.NioIoHandler
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.{ NioServerSocketChannel, NioSocketChannel }
import io.netty.handler.codec.{ LengthFieldBasedFrameDecoder, LengthFieldPrepender }
@@ -366,11 +367,10 @@ class NettyTransport(val settings: NettyTransportSettings, val system: ExtendedA
private val log = Logging.withMarker(system, classOf[NettyTransport])
- @nowarn("msg=deprecated")
private def createEventLoopGroup(nThreadCount: Int): EventLoopGroup =
UseDispatcherForIo.map(system.dispatchers.lookup)
- .map(executor => new NioEventLoopGroup(0, executor))
- .getOrElse(new NioEventLoopGroup(nThreadCount, system.threadFactory))
+ .map(executor => new MultiThreadIoEventLoopGroup(0, executor, NioIoHandler.newFactory()))
+ .getOrElse(new MultiThreadIoEventLoopGroup(nThreadCount, system.threadFactory, NioIoHandler.newFactory()))
/*
* Be aware, that the close() method of DefaultChannelGroup is racy, because it uses an iterator over a ConcurrentHashMap.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org