alifirat alifirat - 1 year ago 69
Scala Question

Akka Actor - Asking message gives me a Empty Set

In order to avoid using a mutable variable inside an

, I made the following things :

import akka._
import play.api.Logger
import scala.language.implicitConversions

object Sessions_Buffer {

case class Sessions_Buffer_Data(
data: scala.collection.immutable.Map[String, List[Array[Byte]]]

// var sessions_buffer: Map[String, List[Array[Byte]]] = Map()
def props(): Props = Props(Sessions_Buffer())

case class Sessions_Buffer() extends Actor {

import Sessions_Buffer._
def receive = active(Map())

def mergeMaps(m1: Map[String, List[Array[Byte]]], m2: Map[String, List[Array[Byte]]]) =
for ((key, values) <- m2) yield {
val existingData = m1.getOrElse(key, List())
(key -> (existingData ++ values))

def active(dataFromKafka: Map[String, List[Array[Byte]]]): Receive = {
case sessionData: Sessions_Buffer_Data =>
for ((key, values) <- dataFromKafka) {
Logger.debug("key = " + key + " values =" + values.size)
case "get_session_data" =>
sender() ! dataFromKafka

class Test_Kakfa_To_Buffer
extends TestKit(ActorSystem("test-kafka-buffer"))
with ImplicitSender
with WordSpecLike
with Matchers
with BeforeAndAfterAll {

var kafkaProps: Properties = _

override def beforeAll = {
kafkaProps = KafkaConfigService.securedDeviceKafkaProps

override def afterAll = {

val buffer_Dispatcher = system.actorOf(Sessions_Buffer.props())
... // some process here
val futureSessionsData = buffer_Dispatcher ? "get_session_data"
val sessionsData: Map[String, List[Array[Byte]]] =

Await.result(futureSessionsData,timeout.duration).asInstanceOf[Map[String, List[Array[Byte]]]]
assert(sessionsData.keys.size == (TestUtils.NUMBER_OF_DEVICES + 1))

The problem is that in one unit test that I wrote, when I'm asking the session_data, the resulting Map is empty.The output error from scalatest :

Set() had size 0 instead of expected size 4

During the process, the logs shows that the Map is not empty but when I'm asking this map, it's empty ! What I'm missing ?

Answer Source

In general, your code is structured correctly, but there are two problems that is causing this "weird" behaviour:

  1. Your merge function has a bug - it does not consider keys which are in m1 but not on m2.

  2. You are logging the content of the sessionData message, but what you are returning as a result of get_session_data is the merge of the sessionData content with the previous content.

I believe if you fix these two issues you will have what you are looking for!

Recommended from our users: Dynamic Network Monitoring from WhatsUp Gold from IPSwitch. Free Download