akka_trigger.scala

val rateLimiter =
  Flow[(Integer, String)]
    .statefulMapConcat { () =>
      val hashMap = new mutable.HashMap[Integer, (Long, String)]()
      (el) => el match {
        case (userId, alert) => {
          val current = System.currentTimeMillis()
          hashMap.get(userId) match {
            case Some((oldTime, oldAlert)) => {
              if (oldTime + 15.seconds.toMillis < current) {
                hashMap.put(userId, (current, alert))
                List[(Integer, String)]((userId, alert))
              } else
                List[(Integer, String)]()
            }
            case None => {
              hashMap.put(userId, (current, alert))
              List[(Integer, String)]((userId, alert))
            }
          }
        }
      }
    }