Как проверить издателя на DistributedPubSub в кластере Akka?

Question

У меня есть актер, единственная ответственность которого заключается в пересылке сообщений, которые он получает от внешних интерфейсов (командной строки, пользователя и т. Д.), В соответствующие темы. Я хочу проверить, правильно ли он публикует эти сообщения.

Мне нужно будет создать фиктивного подписчика, который будет ожидать сообщений, опубликованных на определенную тему, и делать утверждения о сообщениях, которые он получает.

Вот мой код, который я пытался понять, что:

Messages.scala

case class Foo(foo: String)

InterfaceForwardingActor.scala

import akka.actor.{Actor, ActorLogging}
import akka.cluster.pubsub.{DistributedPubSub, DistributedPubSubMediator}
import com.typesafe.config.Config

/** Actor responsible for forwarding stimuli external to the system.
  * For instance, messages from the command-line interface or from a UI.
  *
  */
class InterfaceForwardingActor extends Actor with ActorLogging {
  import DistributedPubSubMediator.Publish

  protected val mediator = DistributedPubSub(context.system).mediator

  log.info(s"Hello from interface forwarder.")

  final val topic = "info"

  def receive = {
    case foo: Foo => {
      log.info("Forwarding a Foo message")
      mediator ! Publish(topic, foo)
    }
  }  
}

и тестовый код

InterfaceForwardingActorTest.scala

import akka.actor.{ActorSystem, Props}
import akka.cluster.client.ClusterClient.Publish
import akka.cluster.pubsub.DistributedPubSub
import akka.testkit.{ImplicitSender, TestKit, TestProbe}
import com.typesafe.config.ConfigFactory
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}

class InterfaceForwardingActorTest extends 
    TestKit(ActorSystem("InterfaceForwardingActorSpec")) with ImplicitSender with 
    WordSpecLike with Matchers with BeforeAndAfterAll {

  override def afterAll {
    TestKit.shutdownActorSystem(system)
  }

  "An InterfaceForwardingActor" must {
    val interfaceForwardingActor = system.actorOf(Props[InterfaceForwardingActor])

    val probe = TestProbe()
    val mediator = DistributedPubSub(system).mediator

    // subscribe the test probe to the "info" topic
    mediator ! Publish("info", probe.ref)

    "publish a Foo message" in {
      val msg = Foo("test")
      interfaceForwardingActor ! msg
      probe.expectMsg(msg)
    }
  }
}

Я обнаружил, что probe, который подписан на info тема не получает сообщение в течение установленного по умолчанию периода времени, равного 3 секундам, и утверждение не выполняется. Интересная часть, однако, заключается в том, что я вижу сообщение в журнале, в котором говорится, что субъект перенаправления интерфейса действительно пересылает сообщение Foo.

Что я делаю не так в моем тесте?

Answer 1

TestProbe следует подписаться на тему в тестовом коде:

mediator ! Subscribe("info", probe.ref)

вместо

mediator ! Publish("info", probe.ref)

Страница документации распределенного pub-sub здесь, для справки.