title: Maintaining state within a stream
body: <p>I have a heavy load flow of users data. I want to determine if this is a new user by it's id. In order to reduce calls to the db I rather maintain a state in memory of previous users.</p>
  
  <pre><code>val users = mutable.set[String]()
  //init the state from db
  user = db.getAllUsersIds()
  val source: Source[User, NotUsed]
  val dbSink: Sink[User, NotUsed] //goes to db
  //if the user is added to the set it will return true
  val usersFilter = Flow[User].filter(user =&gt; users.add(user.id))
  </code></pre>
  
  <p>now I can create a graph</p>
  
  <pre><code>source ~&gt; usersFilter ~&gt; dbSink
  </code></pre>
  
  <p>my problem is that the mutable state is shared and unsafe. Is there an option to maintain the state within the flow ?</p>