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

Fixing ticket #652 - Reaping expired futures

Browse source
This commit is contained in:
Viktor Klang 2011-02-08 15:06:13 +01:00
parent bc423fcc76
commit 83f9c4982f
3 changed files with 19 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

1
akka-remote/src/main/scala/akka/remote/RemoteShared.scala
View file

@ -16,6 +16,7 @@ object RemoteClientSettings {
val RECONNECTION_TIME_WINDOW = Duration(config.getInt("akka.remote.client.reconnection-time-window", 600), TIME_UNIT).toMillis
val READ_TIMEOUT = Duration(config.getInt("akka.remote.client.read-timeout", 1), TIME_UNIT)
val RECONNECT_DELAY = Duration(config.getInt("akka.remote.client.reconnect-delay", 5), TIME_UNIT)
val REAP_FUTURES_DELAY = Duration(config.getInt("akka.remote.client.reap-futures-delay", 5), TIME_UNIT)
val MESSAGE_FRAME_SIZE = config.getInt("akka.remote.client.message-frame-size", 1048576)
}

19
akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala
View file

@ -284,6 +284,19 @@ class ActiveRemoteClient private[akka] (
log.slf4j.debug("Remote client connection failed", connection.getCause)
false
} else {
timer.newTimeout(new TimerTask() {
def run(timeout: Timeout) = {
if(isRunning) {
log.slf4j.debug("Reaping expired futures awaiting completion from [{}]", remoteAddress)
val i = futures.entrySet.iterator
while(i.hasNext) {
val e = i.next
if (e.getValue.isExpired)
futures.remove(e.getKey)
}
}
}
}, RemoteClientSettings.REAP_FUTURES_DELAY.length, RemoteClientSettings.REAP_FUTURES_DELAY.unit)
notifyListeners(RemoteClientStarted(module, remoteAddress))
true
}
@ -440,8 +453,10 @@ class ActiveRemoteClientHandler(
if (client.isWithinReconnectionTimeWindow) {
timer.newTimeout(new TimerTask() {
def run(timeout: Timeout) = {
client.openChannels.remove(event.getChannel)
client.connect(reconnectIfAlreadyConnected = true)
if (client.isRunning) {
client.openChannels.remove(event.getChannel)
client.connect(reconnectIfAlreadyConnected = true)
}
}
}, RemoteClientSettings.RECONNECT_DELAY.toMillis, TimeUnit.MILLISECONDS)
} else spawn { client.shutdown }

1
config/akka-reference.conf
View file

@ -135,6 +135,7 @@ akka {
reconnect-delay = 5
read-timeout = 10
message-frame-size = 1048576
reap-futures-delay = 5
reconnection-time-window = 600 # Maximum time window that a client should try to reconnect for
}
}