scala - Akka Streams Covarience for SourceQueueWithComplete

Question

I have an Actor which is created from within a SupervisorActor and this Actor is responsible for pushing the messages that it gets into a stream. Here is the Actor:

class KafkaPublisher[T <: KafkaMessage: ClassTag] extends Actor {

  implicit val system = context.system
  val log = Logging(system, this.getClass.getName)

  override final def receive = {
    case ProducerStreamActivated(_, stream: SourceQueueWithComplete[T]) =>
      log.info(s"Activated stream for Kafka Producer with ActorName >> ${self.path.name} << ActorPath >> ${self.path} <<")
      context.become(active(stream))

    case other =>
      log.warning("KafkaPublisher got some unknown message while producing: " + other)
  }

  def active(stream: SourceQueueWithComplete[T]): Receive = {
    case msg: T =>
      log.info(s"Got Message >> $msg << Pushing it to stream $stream")
      stream.offer(msg)

    case other =>
      log.warning("KafkaPublisher got the unknown message while producing: " + other)
  }
}
object KafkaPublisher {

  def props[T <: KafkaMessage: ClassTag] =
    Props(new KafkaPublisher[T])
}

As it can be seen that this Actor works on a generic message type as the SupervisorActor is responsible for creating concrete instances of this generic Actor.

trait Event
object Event {
  case class ProducerStreamActivated[T <: KafkaMessage](kafkaTopic: String, stream: SourceQueueWithComplete[T]) extends Event
}

trait KafkaMessage
object KafkaMessage {

  case class DefaultMessage(message: String, timestamp: DateTime) extends KafkaMessage {
    def this() = this("DEFAULT-EMPTY-MESSAGE", DateTime.now(DateTimeZone.UTC))
  }

  case class DefaultMessageBundle(messages: Seq[DefaultMessage], timeStamp: DateTime) extends KafkaMessage {
    def this() = this(Seq.empty, DateTime.now(DateTimeZone.UTC))
  }
}

Obviously, I get the following error:

Error:(17, 45) pattern type is incompatible with expected type;
 found   : akka.stream.scaladsl.SourceQueueWithComplete[T]
 required: akka.stream.scaladsl.SourceQueueWithComplete[com.eon.pm.messages.KafkaMessage]
Note: T <: com.my.project.messages.KafkaMessage, but trait SourceQueueWithComplete is invariant in type T.
You may wish to define T as +T instead. (SLS 4.5)
    case ProducerStreamActivated(_, stream: SourceQueueWithComplete[T]) =>

The SourceQueueWithComplete is not Covariant. How can I mitigate this? Any suggestions?

Answer 1

like:

class KafkaPublisher extends Actor {

  implicit val system = context.system
  val log = Logging(system, this.getClass.getName)

  override final def receive = {
    case ProducerStreamActivated(_, stream) =>
      log.info(s"Activated stream for Kafka Producer with ActorName >> ${self.path.name} << ActorPath >> ${self.path} <<")
      context.become(active(stream))

    case other =>
      log.warning("KafkaPublisher got some unknown message while producing: " + other)
  }

  def active(stream: SourceQueueWithComplete[KafkaMessage]): Receive = {
    case msg: DefaultMessage =>
      log.info(s"Got Message >> $msg << Pushing it to stream $stream")
      stream.offer(msg)

    case msg: DefaultMessageBundle =>
      log.info(s"Got Message >> $msg << Pushing it to stream $stream")
      stream.offer(msg)

    case other =>
      log.warning("KafkaPublisher got the unknown message while producing: " + other)
  }
}
object KafkaPublisher {

  def props = Props(new KafkaPublisher)
}

trait Event
object Event {
  case class ProducerStreamActivated(kafkaTopic: String, stream: SourceQueueWithComplete[KafkaMessage]) extends Event
}

trait KafkaMessage
object KafkaMessage {

  case class DefaultMessage(message: String, timestamp: DateTime) extends KafkaMessage {
    def this() = this("DEFAULT-EMPTY-MESSAGE", DateTime.now(DateTimeZone.UTC))
  }

  case class DefaultMessageBundle(messages: Seq[DefaultMessage], timeStamp: DateTime) extends KafkaMessage {
    def this() = this(Seq.empty, DateTime.now(DateTimeZone.UTC))
  }
}