BlogWebSocketActor.scala

import akka.actor.{Actor, ActorRef, ActorSystem, PoisonPill, Props}
import akka.event.Logging
import Models.{SubscriptionMessage, _}
import JsonUtils.stringify
import play.api.libs.json.{JsString, Json}

object WebSocketActor {
  def props(out: ActorRef, redisSubActorProps: Props, actorSystem: ActorSystem): Props =
    Props(new WebSocketActor(out, redisSubActorProps, actorSystem))
}

class WebSocketActor(client: ActorRef, redisSubActorProps: Props, actorSystem: ActorSystem) extends Actor {
  val log = Logging(context.system, this)
  // Cannot use context to create child actor because the child will be terminated before postStop happens
  private val redisSubActor = actorSystem.actorOf(redisSubActorProps)
  private var subscriptionMessage: SubscriptionMessage = _

  override def receive: Receive = {
    case ThrowableOrSubscriptionsMessage(disjunction) =>
      disjunction foreach { (message: SubscriptionMessage) =>
        subscriptionMessage = message
        redisSubActor ! RedisSubscribeMessage(Seq(message.userId))
        client ! stringify(SuccessResponse(payload =
          JsString(s"Subscribed to events for User: ${message.userId}")))
      }

      disjunction.swap foreach { err =>
        client ! stringify(Json.toJson(ErrorResponse(s"Invalid subscription message format: ${err.getMessage}")))
      }
    
    case NotificationEvent(eventSubscription, payload) =>
      Some(eventSubscription).filter { sub =>
        subscriptionMessage.userId == sub.userId 
        // && ... 
        // it's up to you to define subscription mechanism and do fine grained filtering here
      }.foreach { _ => client ! stringify(SuccessResponse(payload))}
  }

  override def postStop(): Unit = {
    Some(subscriptionMessage).foreach { message =>
      redisSubActor ! RedisUnsubscribeMessage(Seq(message.userId))
      redisSubActor ! PoisonPill
    }
  }
}