raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Fixed issue #190: RemoteClient shutdown ends up in endless loop

Browse source
This commit is contained in:
Jonas Bonér 2010-05-21 11:37:08 +02:00
parent 86a8fd0d3b
commit 2fce1676b2
2 changed files with 5 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-core/src/main/scala/dispatch/ThreadPoolBuilder.scala
View file

@ -30,7 +30,7 @@ trait ThreadPoolBuilder {
def isShutdown = executor.isShutdown def isShutdown = executor.isShutdown
def buildThreadPool = synchronized { def buildThreadPool: Unit = synchronized {
ensureNotActive ensureNotActive
inProcessOfBuilding = false inProcessOfBuilding = false
if (boundedExecutorBound > 0) { if (boundedExecutorBound > 0) {

9
akka-core/src/main/scala/remote/RemoteClient.scala
View file

@ -139,11 +139,11 @@ class RemoteClient(val hostname: String, val port: Int) extends Logging {
Executors.newCachedThreadPool) Executors.newCachedThreadPool)
private val bootstrap = new ClientBootstrap(channelFactory) private val bootstrap = new ClientBootstrap(channelFactory)
private val openChannels = new DefaultChannelGroup(classOf[RemoteClient].getName);
private val timer = new HashedWheelTimer private val timer = new HashedWheelTimer
private val remoteAddress = new InetSocketAddress(hostname, port) private val remoteAddress = new InetSocketAddress(hostname, port)
private[remote] var connection: ChannelFuture = _ private[remote] var connection: ChannelFuture = _
private[remote] val openChannels = new DefaultChannelGroup(classOf[RemoteClient].getName);
bootstrap.setPipelineFactory(new RemoteClientPipelineFactory(name, futures, supervisors, bootstrap, remoteAddress, timer, this)) bootstrap.setPipelineFactory(new RemoteClientPipelineFactory(name, futures, supervisors, bootstrap, remoteAddress, timer, this))
bootstrap.setOption("tcpNoDelay", true) bootstrap.setOption("tcpNoDelay", true)
@ -296,11 +296,10 @@ class RemoteClientHandler(val name: String,
override def channelClosed(ctx: ChannelHandlerContext, event: ChannelStateEvent) = if (client.isRunning) { override def channelClosed(ctx: ChannelHandlerContext, event: ChannelStateEvent) = if (client.isRunning) {
timer.newTimeout(new TimerTask() { timer.newTimeout(new TimerTask() {
def run(timeout: Timeout) = { def run(timeout: Timeout) = {
client.openChannels.remove(event.getChannel)
log.debug("Remote client reconnecting to [%s]", remoteAddress) log.debug("Remote client reconnecting to [%s]", remoteAddress)
client.connection = bootstrap.connect(remoteAddress) client.connection = bootstrap.connect(remoteAddress)
client.connection.awaitUninterruptibly // Wait until the connection attempt succeeds or fails.
// Wait until the connection attempt succeeds or fails.
client.connection.awaitUninterruptibly
if (!client.connection.isSuccess) { if (!client.connection.isSuccess) {
client.listeners.toArray.foreach(l => l.asInstanceOf[ActorRef] ! RemoteClientError(client.connection.getCause)) client.listeners.toArray.foreach(l => l.asInstanceOf[ActorRef] ! RemoteClientError(client.connection.getCause))
log.error(client.connection.getCause, "Reconnection to [%s] has failed", remoteAddress) log.error(client.connection.getCause, "Reconnection to [%s] has failed", remoteAddress)