Server-sent Events (SSE)

The SSE connector provides a continuous source of server-sent events recovering from connection failure.

Project Info: Alpakka Server-sent events (SSE)
Artifact
com.lightbend.akka
akka-stream-alpakka-sse
1.0-M2
JDK versions
OpenJDK 8
Scala versions2.12.7, 2.11.12
JPMS module nameakka.stream.alpakka.sse
License
Readiness level
Community-driven
Since 0.6, 2017-02-13
Home pagehttps://doc.akka.io/docs/alpakka/current/
API documentation
Forums
Release notesIn the documentation
IssuesGithub issues
Sourceshttps://github.com/akka/alpakka

Artifacts

sbt
libraryDependencies += "com.lightbend.akka" %% "akka-stream-alpakka-sse" % "1.0-M2"
Maven
<dependency>
  <groupId>com.lightbend.akka</groupId>
  <artifactId>akka-stream-alpakka-sse_2.12</artifactId>
  <version>1.0-M2</version>
</dependency>
Gradle
dependencies {
  compile group: 'com.lightbend.akka', name: 'akka-stream-alpakka-sse_2.12', version: '1.0-M2'
}

The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.

Direct dependencies
OrganizationArtifactVersionLicense
com.typesafe.akkaakka-http_2.1210.1.7Apache-2.0
com.typesafe.akkaakka-stream_2.122.5.19Apache License, Version 2.0
org.scala-langscala-library2.12.7BSD 3-Clause
Dependency tree
com.typesafe.akka    akka-http_2.12    10.1.7    Apache-2.0
    com.typesafe.akka    akka-http-core_2.12    10.1.7    Apache-2.0
        com.typesafe.akka    akka-parsing_2.12    10.1.7    Apache-2.0
            org.scala-lang    scala-library    2.12.7    BSD 3-Clause
        org.scala-lang    scala-library    2.12.7    BSD 3-Clause
    org.scala-lang    scala-library    2.12.7    BSD 3-Clause
com.typesafe.akka    akka-stream_2.12    2.5.19    Apache License, Version 2.0
    com.typesafe.akka    akka-actor_2.12    2.5.19    Apache License, Version 2.0
        com.typesafe    config    1.3.3    Apache License, Version 2.0
        org.scala-lang.modules    scala-java8-compat_2.12    0.8.0    BSD 3-clause
            org.scala-lang    scala-library    2.12.7    BSD 3-Clause
        org.scala-lang    scala-library    2.12.7    BSD 3-Clause
    com.typesafe.akka    akka-protobuf_2.12    2.5.19    Apache License, Version 2.0
        org.scala-lang    scala-library    2.12.7    BSD 3-Clause
    com.typesafe    ssl-config-core_2.12    0.3.6    Apache-2.0
        com.typesafe    config    1.3.3    Apache License, Version 2.0
        org.scala-lang.modules    scala-parser-combinators_2.12    1.1.1    BSD 3-clause
            org.scala-lang    scala-library    2.12.7    BSD 3-Clause
        org.scala-lang    scala-library    2.12.7    BSD 3-Clause
    org.reactivestreams    reactive-streams    1.0.2    CC0
    org.scala-lang    scala-library    2.12.7    BSD 3-Clause
org.scala-lang    scala-library    2.12.7    BSD 3-Clause

Usage

Define an EventSource by giving a URI, a function for sending HTTP requests, and an optional initial value for Last-Evend-ID header:

Scala
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.sse.ServerSentEvent
import akka.http.scaladsl.model.{HttpEntity, HttpRequest, HttpResponse, Uri}
import akka.stream.alpakka.sse.scaladsl.EventSource

val send: HttpRequest => Future[HttpResponse] = Http().singleRequest(_)

val eventSource: Source[ServerSentEvent, NotUsed] =
  EventSource(
    uri = Uri(s"http://$host:$port"),
    send,
    initialLastEventId = Some("2"),
    retryDelay = 1.second
  )
Java
import java.util.function.Function;
import java.util.concurrent.CompletionStage;

import akka.http.javadsl.Http;
import akka.http.javadsl.model.*;
import akka.http.javadsl.model.sse.ServerSentEvent;
import akka.stream.alpakka.sse.javadsl.EventSource;

final Http http = Http.get(system);
Function<HttpRequest, CompletionStage<HttpResponse>> send =
    (request) -> http.singleRequest(request);

final Uri targetUri = Uri.create(String.format("http://%s:%d", host, port));
final Optional<String> lastEventId = Optional.of("2");
Source<ServerSentEvent, NotUsed> eventSource =
    EventSource.create(targetUri, send, lastEventId, materializer);

Then happily consume ServerSentEvents:

Scala
val events: Future[immutable.Seq[ServerSentEvent]] =
  eventSource
    .throttle(elements = 1, per = 500.milliseconds, maximumBurst = 1, ThrottleMode.Shaping)
    .take(nrOfSamples)
    .runWith(Sink.seq)
Java
int elements = 1;
Duration per = Duration.ofMillis(500);
int maximumBurst = 1;

eventSource
    .throttle(elements, per, maximumBurst, ThrottleMode.shaping())
    .take(nrOfSamples)
    .runWith(Sink.seq(), materializer);
Found an error in this documentation? The source code for this page can be found here. Please feel free to edit and contribute a pull request.