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
2.0.2
|
JDK versions | Adopt OpenJDK 8 Adopt OpenJDK 11 |
Scala versions | 2.12.11, 2.11.12, 2.13.3 |
JPMS module name | akka.stream.alpakka.sse |
License | |
Readiness level |
Since 0.6, 2017-02-13
|
Home page | https://doc.akka.io/docs/alpakka/current |
API documentation | |
Forums | |
Release notes | In the documentation |
Issues | Github issues |
Sources | https://github.com/akka/alpakka |
Artifacts
- sbt
val AkkaVersion = "2.5.31" val AkkaHttpVersion = "10.1.11" libraryDependencies ++= Seq( "com.lightbend.akka" %% "akka-stream-alpakka-sse" % "2.0.2", "com.typesafe.akka" %% "akka-stream" % AkkaVersion, "com.typesafe.akka" %% "akka-http" % AkkaHttpVersion )
- Maven
<properties> <akka.version>2.5.31</akka.version> <akka.http.version>10.1.11</akka.http.version> <scala.binary.version>2.12</scala.binary.version> </properties> <dependency> <groupId>com.lightbend.akka</groupId> <artifactId>akka-stream-alpakka-sse_${scala.binary.version}</artifactId> <version>2.0.2</version> </dependency> <dependency> <groupId>com.typesafe.akka</groupId> <artifactId>akka-stream_${scala.binary.version}</artifactId> <version>${akka.version}</version> </dependency> <dependency> <groupId>com.typesafe.akka</groupId> <artifactId>akka-http_${scala.binary.version}</artifactId> <version>${akka.http.version}</version> </dependency>
- Gradle
versions += [ AkkaVersion: "2.5.31", AkkaHttpVersion: "10.1.11", ScalaBinary: "2.12" ] dependencies { compile group: 'com.lightbend.akka', name: "akka-stream-alpakka-sse_${versions.ScalaBinary}", version: '2.0.2', compile group: 'com.typesafe.akka', name: "akka-stream_${versions.ScalaBinary}", version: versions.AkkaVersion, compile group: 'com.typesafe.akka', name: "akka-http_${versions.ScalaBinary}", version: versions.AkkaHttpVersion }
The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.
- Direct dependencies
Organization Artifact Version com.typesafe.akka akka-http_2.12 10.1.11 com.typesafe.akka akka-stream_2.12 2.5.31 org.scala-lang scala-library 2.12.11 - Dependency tree
com.typesafe.akka akka-http_2.12 10.1.11 com.typesafe.akka akka-http-core_2.12 10.1.11 com.typesafe.akka akka-parsing_2.12 10.1.11 org.scala-lang scala-library 2.12.11 org.scala-lang scala-library 2.12.11 org.scala-lang scala-library 2.12.11 com.typesafe.akka akka-stream_2.12 2.5.31 com.typesafe.akka akka-actor_2.12 2.5.31 com.typesafe config 1.3.3 org.scala-lang.modules scala-java8-compat_2.12 0.8.0 org.scala-lang scala-library 2.12.11 org.scala-lang scala-library 2.12.11 com.typesafe.akka akka-protobuf_2.12 2.5.31 org.scala-lang scala-library 2.12.11 com.typesafe ssl-config-core_2.12 0.3.8 com.typesafe config 1.3.3 org.scala-lang.modules scala-parser-combinators_2.12 1.1.2 org.scala-lang scala-library 2.12.11 org.scala-lang scala-library 2.12.11 org.reactivestreams reactive-streams 1.0.2 org.scala-lang scala-library 2.12.11 org.scala-lang scala-library 2.12.11
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 import org.scalatest.matchers.should.Matchers import org.scalatest.wordspec.AsyncWordSpec 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 ServerSentEvent
s:
- 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);