title: Overcoming Type Erasure in Scala with TypeTag
body: <p>I have the following case class:</p>
  
  <pre><code>trait Event
  object Event {
    case class ProducerStreamActivated[T &lt;: 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))
    }
  }
  </code></pre>
  
  <p>In one of my Actor, I have the following method that identifies the actual Type:</p>
  
  <pre><code>class KafkaPublisher[T &lt;: KafkaMessage: TypeTag] extends Actor {
  
    def paramInfo[T](x: T)(implicit tag: TypeTag[T]): Unit = {
      val targs = typeOf[T] match { case TypeRef(_, _, args) =&gt; args }
      println(s"type of $x has type arguments $targs")
    }
  
    implicit val system = context.system
    val log = Logging(system, this.getClass.getName)
  
    override final def receive = {
      case ProducerStreamActivated(_, stream) =&gt;
        paramInfo(stream)
        log.info(s"Activated stream for Kafka Producer with ActorName &gt;&gt; ${self.path.name} &lt;&lt; ActorPath &gt;&gt; ${self.path} &lt;&lt;")
        context.become(active(stream))
  
      case other =&gt;
        log.warning("KafkaPublisher got some unknown message while producing: " + other)
    }
  
    def active(stream: SourceQueueWithComplete[KafkaMessage]): Receive = {
      case msg: T =&gt;
        stream.offer(msg)
  
      case other =&gt;
        log.warning("KafkaPublisher got the unknown message while producing: " + other)
    }
  }
  object KafkaPublisher {
  
    def props[T &lt;: KafkaMessage: TypeTag] =
      Props(new KafkaPublisher[T])
  }
  </code></pre>
  
  <p>I create an instance of the ProducerStreamActivated(...) in a parent Actor like this:</p>
  
  <pre><code>val stream = producerStream[DefaultMessage](producerProperties)
    def producerStream[T: Converter](producerProperties: Map[String, String]): SourceQueueWithComplete[T] = {
      if (Try(producerProperties("isEnabled").toBoolean).getOrElse(false)) {
        log.info(s"Kafka is enabled for topic ${producerProperties("publish-topic")}")
        val streamFlow = flowToKafka[T](producerProperties)
        val streamSink = sink(producerProperties)
        source[T].via(streamFlow).to(streamSink).run()
      } else {
        // We just Log to the Console and by pass all Kafka communication
        log.info(s"Kafka is disabled for topic ${producerProperties("publish-topic")}")
        source[T].via(flowToLog[T](log)).to(Sink.ignore).run()
      }
    }
  </code></pre>
  
  <p>When I now print the Type that is contained in the stream SourceQueueWithComplete[T] in my child actor, I get to see the base class KafkaMessage contained instead of the expected DefaultMessage. Any ideas how to mitigate this?</p>