How do I handle a massive stream of events without crashing my server?

If you're building an analytics dashboard for something like an e-commerce site during a Black Friday sale, you can't just throw every incoming event into a List and hope for the best. You'll run out of heap space in minutes. This is where backpressure comes in. I always recommend using Pekko Streams (or Akka) for this because it handles the "producer is faster than the consumer" problem natively.

Instead of a traditional imperative loop, you define a pipeline. In our case, we're processing PurchaseEvent objects. By using a Source and a Flow, the system ensures that if your aggregation logic slows down, the source (like a Kafka topic) is told to slow down too.

case class PurchaseEvent(orderId: String, amount: Double, category: String, timestamp: Long)

// A simple stream that filters out "test" orders and maps them to a simplified metric
val analyticsFlow = Flow[PurchaseEvent]
  .filter(_.amount > 0) // Ignore zero-dollar test transactions
  .map(event => (event.category, event.amount))
  .groupedWithin(100, 1.second) // Batch them to avoid overwhelming the downstream

The groupedWithin call is a lifesaver. It prevents your backend from trying to push a WebSocket update for every single single-cent transaction, instead bundling them into manageable chunks every second.

How do I aggregate data over a sliding window instead of just counting everything since startup?

A dashboard that shows "Total Sales Since 2022" is a report; a dashboard that shows "Sales in the last 5 minutes" is real-time analytics. To do this in Scala, you need a way to maintain state that expires. While you could use a database, for a high-frequency dashboard, I prefer doing this in-memory using a stateful map or a sliding window logic.

I usually implement a StatefulMap approach where I keep a buffer of timestamps. Here is a simplified version of how you might track the rolling sum of revenue:

class RevenueAggregator {
  private var window = scala.collection.mutable.Queue[(Long, Double)]()

  def addAndCalculate(amount: Double, now: Long): Double = {
    window.enqueue((now, amount))
    
    // Slide the window: remove any events older than 5 minutes
    val fiveMinutesAgo = now - (5 * 60 * 1000)
    while (window.nonEmpty && window.head._1 < fiveMinutesAgo) {
      window.dequeue()
    }
    
    window.map(_._2).sum
  }
}

It's a simple queue, but it's effective. Just remember: if you're running this across a cluster, you'll need to move this state into something like Redis or use a distributed streaming engine. For a single-node backend, this is plenty fast.

How do I push these updates to multiple users in real-time without polling?

You don't want your frontend hitting an endpoint every 500ms—that's just an expensive way to waste CPU cycles. You want a "push" model. In the Scala ecosystem, the BroadcastHub is your best friend here. It allows you to take a single stream of analytics and "fan it out" to every connected WebSocket client.

Think of it like a radio station: the backend broadcasts the current metrics, and the users just tune in. Here is how you hook that up to a WebSocket route:

// Create a hub that broadcasts the aggregated metrics to all subscribers
val (hub, source) = BroadcastHub.sinkAndSource[DashboardMetrics](bufferSize = 256)

// The stream pipeline feeds into the hub
analyticsFlow
  .map(batch => calculateMetrics(batch))
  .to(hub)
  .run()

// In your HTTP routes:
val route = path("ws" / "metrics") {
  get {
    complete {
      HttpEntity.WebSocket { request, session =>
        // Each user gets their own stream from the hub
        val flow = Flow[String].ignore // We only push, not receive
        source.map(metrics => metrics.toJson.toString).to(session)
      }
    }
  }
}

The magic here is that source is a "hot" stream. New users joining the dashboard don't get the entire history of the app; they just start receiving the latest updates as they happen.

📋 Practical Task

Implementing a High-Value Order Filter Stream

Your analytics dashboard is getting too noisy. The stakeholders only want a specific "High-Value Alert" feed that pushes orders greater than $500 to a separate WebSocket channel.

Your Task:

• Create a new Flow[PurchaseEvent] that filters for orders where amount > 500.0.
• Implement a transformation that converts the PurchaseEvent into a HighValueAlert case class (which should contain the orderId and a formatted string: "Large purchase of $[amount] detected!").
• Connect this flow to a second BroadcastHub so that these alerts are broadcast independently from the general metrics stream.
• Ensure the stream is non-blocking and can handle at least 1,000 events per second without dropping alerts.