markusthoemmes commented on a change in pull request #2425: Throttle message
bus consumption.
URL:
https://github.com/apache/incubator-openwhisk/pull/2425#discussion_r124457802
##########
File path:
core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
##########
@@ -151,13 +141,23 @@ class InvokerPool(
}
/** Receive Ping messages from invokers. */
- pingConsumer.onMessage((topic, _, _, bytes) => {
+ val pingPollDuration = 1.second
+ val invokerPingFeed = context.system.actorOf(Props {
+ new MessageFeed("ping", logging, pingConsumer, pingConsumer.maxPeek,
pingPollDuration, processInvokerPing, logHandoff = false)
+ })
+
+ def processInvokerPing(bytes: Array[Byte]): Future[Unit] = Future {
val raw = new String(bytes, StandardCharsets.UTF_8)
PingMessage.parse(raw) match {
- case Success(p: PingMessage) => self ! p
- case Failure(t) => logging.error(this, s"failed
processing message: $raw with $t")
+ case Success(p: PingMessage) =>
+ self ! p
+ invokerPingFeed ! MessageFeed.Processed
+
+ case Failure(t) =>
+ invokerPingFeed ! MessageFeed.Processed
Review comment:
Send this message after the `match`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
