Custom router and dead routees

[Banduk]

Hi guys,

I have some questions about the following custom router. I still a little bit new on 

scala and akka so maybe you'd do that in a completly different way.

Please let me know if it's the case!!!

So, I'm receiving messages of the type T and I want to route them by the given field (of the type Node) to actors/routers (defined by p:Props)

here's the codehere's the code

My questions/problems are:

  1. How to implement the commented (line 17) in a better/cleanner/safest way? If I don't do it, I'll have only one router for all the actors (that's not what I want).

  2. I don't know if it's related to the first question, but it's working "fine" to the first messages, but after some time I receive the following message:

      [05/10/2012 16:46:33.848] [testsystem-akka.actor.default-dispatcher-3] [akka://testsystem/user/route/purl_nsAutomobile] received dead letter from Actor[akka://testsystem/deadLetters]: tests.TEST_Routing$$anonfun$1$$anonfun$apply$mcV$sp$1$$anon$1@69ea25aa

ps. I already tried Camel, but I'd prefer to do that by my way so I can have some more freedom and learn more scala/akka!

Thank you,

Banduk

[Björn Antonsson]

Hi Banduk,

On Thursday, 10 May 2012 at 17:45, Banduk wrote:

Hi guys,

I have some questions about the following custom router. I still a little bit new on 

scala and akka so maybe you'd do that in a completly different way.

Please let me know if it's the case!!!

So, I'm receiving messages of the type T and I want to route them by the given field (of the type Node) to actors/routers (defined by p:Props)

here's the codehere's the code

My questions/problems are:

  1. How to implement the commented (line 17) in a better/cleanner/safest way? If I don't do it, I'll have only one router for all the actors (that's not what I want).

Your code in line 17 is just a

p.routerConfig match { case r: SimpleRouter[_] ⇒ whatever }

I'm not exactly sure what your code is supposed to do since the parameterized types (T in your code) are erased by the JVM and will match everything. You need a manifest to reify them, but that is a discussion for the scala-user mailing list.

  2. I don't know if it's related to the first question, but it's working "fine" to the first messages, but after some time I receive the following message:

      [05/10/2012 16:46:33.848] [testsystem-akka.actor.default-dispatcher-3] [akka://testsystem/user/route/purl_nsAutomobile] received dead letter from Actor[akka://testsystem/deadLetters]: tests.TEST_Routing$$anonfun$1$$anonfun$apply$mcV$sp$1$$anon$1@69ea25aa

I'm not sure exactly what happens with the error, but a comment about Routers. The API documentation http://doc.akka.io/api/akka/2.0/#akka.routing.RouterConfig states that the route function can (and will) be called concurrently by multiple threads at the same time, and your code is using unprotected mutable maps, that will definitely miss updates.

I've added this caution to the main docs.

The easiest way out of this would be to turn your router into an actor with the same functionality, that just forwards messages to the child actors. The receive method in an actor is guaranteed to be single threaded.

B/

 

ps. I already tried Camel, but I'd prefer to do that by my way so I can have some more freedom and learn more scala/akka!

Thank you,

Banduk

--
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To view this discussion on the web visit https://groups.google.com/d/msg/akka-user/-/eoHmiNJ10kEJ.
To post to this group, send email to akka...@googlegroups.com.
To unsubscribe from this group, send email to akka-user+...@googlegroups.com.
For more options, visit this group at http://groups.google.com/group/akka-user?hl=en.

-- 

Björn Antonsson

Typesafe - The software stack for applications that scale

[Banduk]

Hi Björn,

Sorry about the delay on answering, I did not have access to my files until today.

Thank you for the explanation and for the hints, I think I got to what I want using an actor as you proposed.

PS:

I noticed that I set the code at pastebin to be available only for 30 days, so here's the code If someone need it later.

class SimpleRouter[T](val filter: T => Node, val p: Props) extends RouterConfig {

  def routerDispatcher: String = Dispatchers.DefaultDispatcherId

  def supervisorStrategy: SupervisorStrategy = SupervisorStrategy.defaultStrategy

  override val resizer = Some(DefaultResizer(lowerBound = 0))

  val routees = collection.mutable.Map[Node, ActorRef]()

  def createRoute(props: Props, routeeProvider: RouteeProvider) = {

    {

      case (sender, message) =>

        message match {

          case msg: T =>

            val fieldVal = filter.apply(msg)

            val actor    = routees.get(fieldVal)

            List(Destination(sender, routees.get(fieldVal).getOrElse({

              // How to better implement this?

              val prop = if (p.routerConfig.isInstanceOf[SimpleRouter[T]]) {

                val rout = p.routerConfig.asInstanceOf[SimpleRouter[T]]

                p.withRouter(new SimpleRouter(rout.filter, rout.p))

              } else p

              val routee = routeeProvider.context.actorOf(p2, (fieldVal.encoded))

              routeeProvider.registerRoutees(IndexedSeq(routee))

              routees += (fieldVal -> routee)

              

              routee

            })))

          case _ => {

            println("UNHANDLED MESSAGE: " + message)

            null

          }

        }

    }

  }

}

To unsubscribe from this group, send email to akka-user+unsubscribe@googlegroups.com.

For more options, visit this group at http://groups.google.com/group/akka-user?hl=en.