diff --git a/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/sliding.md b/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/sliding.md index aca224c22d..83b17ecddb 100644 --- a/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/sliding.md +++ b/akka-docs/src/main/paradox/stream/operators/Source-or-Flow/sliding.md @@ -4,13 +4,9 @@ Provide a sliding window over the incoming stream and pass the windows as groups @ref[Simple operators](../index.md#simple-operators) -@@@div { .group-scala } - ## Signature -@@signature [Flow.scala](/akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala) { #sliding } - -@@@ +@apidoc[Flow.sliding](Flow) { scala="#sliding(n:Int,step:Int):FlowOps.this.Repr[scala.collection.immutable.Seq[Out]]" java="#sliding(int,int)" } ## Description @@ -18,6 +14,37 @@ Provide a sliding window over the incoming stream and pass the windows as groups Note: the last window might be smaller than the requested size due to end of stream. +## Examples + +In this first sample we just see the behavior of the operator itself, first with a window of 2 elements and @scala[the default +`step` which is 1]@java[a step value of 1]. + +Scala +: @@snip [Sliding.scala](/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala) { #sliding-1 } + +Java +: @@snip [Sliding.java](/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java) { #sliding-1 } + +If the stream stops without having seen enough elements to fill a window, the last window will have as many elements +was emitted before the stream ended. Here we also provide a step to move two elements forward for each window: + +Scala +: @@snip [Sliding.scala](/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala) { #sliding-2 } + +Java +: @@snip [Sliding.java](/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java) { #sliding-2 } + +One use case for sliding is to implement a moving average, here we do that with a "period" of `5`: + +Scala +: @@snip [Sliding.scala](/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala) { #moving-average } + +Java +: @@snip [Sliding.java](/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java) { #moving-average } + +Sliding can also be used to do simple windowing, see @ref[splitAfter](splitAfter.md). + + ## Reactive Streams semantics @@@div { .callout } diff --git a/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java b/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java new file mode 100644 index 0000000000..2f8cc3060f --- /dev/null +++ b/akka-docs/src/test/java/jdocs/stream/operators/sourceorflow/Sliding.java @@ -0,0 +1,56 @@ +/* + * Copyright (C) 2009-2020 Lightbend Inc. + */ + +package jdocs.stream.operators.sourceorflow; + +import akka.NotUsed; +import akka.actor.typed.ActorSystem; +import akka.stream.javadsl.Source; + +import java.util.Arrays; +import java.util.stream.Collectors; + +public class Sliding { + + private final ActorSystem system = null; + + public void slidingExample1() { + // #sliding-1 + Source source = Source.range(1, 4); + source.sliding(2, 1).runForeach(n -> System.out.println(n), system); + // prints: + // [1, 2] + // [2, 3] + // [3, 4] + // #sliding-1 + } + + public void slidingExample2() { + // #sliding-2 + Source source = Source.range(1, 4); + source.sliding(3, 2).runForeach(n -> System.out.println(n), system); + // prints: + // Vector(1, 2, 3) + // [1, 2, 3] + // [3, 4] - shorter because stream ended before we got 3 elements + // #sliding-2 + } + + public void slidingExample3() { + // #moving-average + Source numbers = Source.from(Arrays.asList(1, 3, 10, 2, 3, 4, 2, 10, 11)); + Source movingAverage = + numbers + .sliding(5, 1) + .map(window -> ((float) window.stream().mapToInt(i -> i).sum()) / window.size()); + movingAverage.runForeach(n -> System.out.println(n), system); + // prints + // 3.8 = average of 1, 3, 10, 2, 3 + // 4.4 = average of 3, 10, 2, 3, 4 + // 4.2 = average of 10, 2, 3, 4, 2 + // 4.2 = average of 2, 3, 4, 2, 10 + // 6.0 = average of 3, 4, 2, 10, 11 + // #moving-average + } +} diff --git a/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala b/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala new file mode 100644 index 0000000000..a07d5c0d1a --- /dev/null +++ b/akka-docs/src/test/scala/docs/stream/operators/sourceorflow/Sliding.scala @@ -0,0 +1,48 @@ +/* + * Copyright (C) 2019-2020 Lightbend Inc. + */ + +package docs.stream.operators.sourceorflow + +import akka.stream.scaladsl.Source +import akka.actor.ActorSystem + +object Sliding { + implicit val system: ActorSystem = ??? + + def slidingExample1(): Unit = { + //#sliding-1 + val source = Source(1 to 4) + source.sliding(2).runForeach(println) + // prints: + // Vector(1, 2) + // Vector(2, 3) + // Vector(3, 4) + //#sliding-1 + } + + def slidingExample2(): Unit = { + //#sliding-2 + val source = Source(1 to 4) + source.sliding(n = 3, step = 2).runForeach(println) + // prints: + // Vector(1, 2, 3) + // Vector(3, 4) - shorter because stream ended before we got 3 elements + //#sliding-2 + } + + def slidingExample3(): Unit = { + //#moving-average + val numbers = Source(1 :: 3 :: 10 :: 2 :: 3 :: 4 :: 2 :: 10 :: 11 :: Nil) + val movingAverage = numbers.sliding(5).map(window => window.sum.toFloat / window.size) + movingAverage.runForeach(println) + // prints + // 3.8 = average of 1, 3, 10, 2, 3 + // 4.4 = average of 3, 10, 2, 3, 4 + // 4.2 = average of 10, 2, 3, 4, 2 + // 4.2 = average of 2, 3, 4, 2, 10 + // 6.0 = average of 3, 4, 2, 10, 11 + //#moving-average + } + +}