The idiomatic way to manage shared state with Akka Streams

user_534632 0
Viewed 353

I need to filter my flow based on some blacklist which can be changed outside of the flow execution. So, I see two option to do that:

Encapsulate blacklist in a separate service

    class Blacklist(init: Set[String]) {
      @volatile private var set: Set[String] = init

      def get: Set[String] = set
      def update(newSet: Set[String]): Unit = {
        set = newSet
      }
    }

    val blacklist = new Blacklist(Set.empty)

    Flow[String]
      .filterNot(blacklist.get)

Encapsulate blacklist in an actor

    class Blacklist extends Actor {
      import Blacklist._
      private var set = Set.empty[String]
    
      override def receive: Receive = {
        case UpdateBlacklist(newset: Set[String]) =>
          set = newset
        case GetBlacklist =>
          sender ! set
      }
    }
     
    object Blacklist {
      case class UpdateBlacklist(set: Set[String])
      case object GetBlacklist
    }

    val parallelism: Int = ???
    val blacklist = system.actorOf(Props(new Blacklist()))

    Flow[String]
    .mapAsync(parallelism) { str =>
      val ask = blacklist ? Blacklist.GetBlacklist
      ask.mapTo[Set[String]] map { str -> _ }
    } filterNot { case (str, exclude) =>
      exclude(str)
    }

I’m afraid that actor holder solution with mapAsync introduce new asynchronous boundary which prevents operators fusing. So, which one should I prefer? Or is there a more idiomatic way?

akka-stream akka scala
1 Answers

I think in your case the separate service solution is more reasonable. It is a clean solution and avoids the actor overhead.

However, your code does not work as intended. You are passing an immutable Set to filterNot instead of a function. You should consider using a synchronized set directly (based on a ConcurrentHashMap as described in this answer) or implement apply on your blacklist like so:

class Blacklist(init: Set[String]) {
    val blacklist = new AtomicReference[Set[String]](init)

    def apply(s: String) = (blacklist.get) (s)

    def update(newSet: Set[String]): Unit = {
        blacklist.set(newSet)
    }
}

val blacklist = new Blacklist(Set.empty)

Flow[String].filterNot(blacklist)

I recommend always using AtomicReference instead of @volatile since it is more explicit and comprehensible.

user_3027930
Related