Allow LogRotatorSink be a Sink[T] instead of Sink[ByteString]

file/src/main/mima-filters/2.0.0-RC2.backwards.excludes/PR2323-log-rotator-sink.excludes

@@ -0,0 +1,8 @@
+ProblemFilters.exclude[IncompatibleSignatureProblem]("akka.stream.alpakka.file.scaladsl.LogRotatorSink.in")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("akka.stream.alpakka.file.scaladsl.LogRotatorSink.shape")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("akka.stream.alpakka.file.scaladsl.LogRotatorSink.this")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("akka.stream.alpakka.file.scaladsl.LogRotatorSink#Logic.checkTrigger")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("akka.stream.alpakka.file.scaladsl.LogRotatorSink#Logic.rotate")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("akka.stream.alpakka.file.scaladsl.LogRotatorSink#Logic.sourceOut")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("akka.stream.alpakka.file.scaladsl.LogRotatorSink#Logic.sourceOut_=")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("akka.stream.alpakka.file.scaladsl.LogRotatorSink#Logic.triggerGenerator")

file/src/main/scala/akka/stream/alpakka/file/scaladsl/LogRotatorSink.scala

@@ -35,7 +35,8 @@ object LogRotatorSink {
       fileOpenOptions: Set[OpenOption] = Set(StandardOpenOption.APPEND, StandardOpenOption.CREATE)
   ): Sink[ByteString, Future[Done]] =
     Sink.fromGraph(
-      new LogRotatorSink[Path, IOResult](triggerGeneratorCreator, sinkFactory = FileIO.toPath(_, fileOpenOptions))
+      new LogRotatorSink[ByteString, Path, IOResult](triggerGeneratorCreator,
+                                                     sinkFactory = FileIO.toPath(_, fileOpenOptions))
     )
 
   /**
@@ -50,7 +51,22 @@ object LogRotatorSink {
       triggerGeneratorCreator: () => ByteString => Option[C],
       sinkFactory: C => Sink[ByteString, Future[R]]
   ): Sink[ByteString, Future[Done]] =
-    Sink.fromGraph(new LogRotatorSink[C, R](triggerGeneratorCreator, sinkFactory))
+    Sink.fromGraph(new LogRotatorSink[ByteString, C, R](triggerGeneratorCreator, sinkFactory))
+
+  /**
+   * Sink directing the incoming `T`s to a new `Sink` created by `sinkFactory` whenever `triggerGenerator` returns a value.
+   *
+   * @param triggerGeneratorCreator creates a function that triggers rotation by returning a value
+   * @param sinkFactory creates sinks for `T`s from the value returned by `triggerGenerator`
+   * @tparam T stream and sink data type
+   * @tparam C criterion type (for files a `Path`)
+   * @tparam R result type in materialized futures of `sinkFactory`
+   **/
+  def withTypedSinkFactory[T, C, R](
+      triggerGeneratorCreator: () => T => Option[C],
+      sinkFactory: C => Sink[T, Future[R]]
+  ): Sink[T, Future[Done]] =
+    Sink.fromGraph(new LogRotatorSink[T, C, R](triggerGeneratorCreator, sinkFactory))
 }
 
 /**
@@ -61,11 +77,11 @@ object LogRotatorSink {
  * @tparam C criterion type (for files a `Path`)
  * @tparam R result type in materialized futures of `sinkFactory`
  */
-final private class LogRotatorSink[C, R](triggerGeneratorCreator: () => ByteString => Option[C],
-                                         sinkFactory: C => Sink[ByteString, Future[R]])
-    extends GraphStageWithMaterializedValue[SinkShape[ByteString], Future[Done]] {
+final private class LogRotatorSink[T, C, R](triggerGeneratorCreator: () => T => Option[C],
+                                            sinkFactory: C => Sink[T, Future[R]])
+    extends GraphStageWithMaterializedValue[SinkShape[T], Future[Done]] {
 
-  val in = Inlet[ByteString]("LogRotatorSink.in")
+  val in = Inlet[T]("LogRotatorSink.in")
   override val shape = SinkShape.of(in)
 
   override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Done]) = {
@@ -77,8 +93,8 @@ final private class LogRotatorSink[C, R](triggerGeneratorCreator: () => ByteStri
   }
 
   private final class Logic(promise: Promise[Done], decider: Decider) extends GraphStageLogic(shape) {
-    val triggerGenerator: ByteString => Option[C] = triggerGeneratorCreator()
-    var sourceOut: SubSourceOutlet[ByteString] = _
+    val triggerGenerator: T => Option[C] = triggerGeneratorCreator()
+    var sourceOut: SubSourceOutlet[T] = _
     var sinkCompletions: immutable.Seq[Future[R]] = immutable.Seq.empty
 
     def failThisStage(ex: Throwable): Unit =
@@ -90,7 +106,7 @@ final private class LogRotatorSink[C, R](triggerGeneratorCreator: () => ByteStri
         promise.failure(ex)
       }
 
-    def checkTrigger(data: ByteString): Option[C] =
+    def checkTrigger(data: T): Option[C] =
       try {
         triggerGenerator(data)
       } catch {
@@ -144,10 +160,10 @@ final private class LogRotatorSink[C, R](triggerGeneratorCreator: () => ByteStri
       getAsyncCallback[Holder[R]](sinkCompletionCallbackHandler(newFuture))
 
     //we recreate the tail of the stream, and emit the data for the next req
-    def rotate(triggerValue: C, data: ByteString): Unit = {
+    def rotate(triggerValue: C, data: T): Unit = {
       val prevOut = Option(sourceOut)
 
-      sourceOut = new SubSourceOutlet[ByteString]("LogRotatorSink.sub-out")
+      sourceOut = new SubSourceOutlet[T]("LogRotatorSink.sub-out")
       sourceOut.setHandler(new OutHandler {
         override def onPull(): Unit = {
           sourceOut.push(data)

file/src/test/scala/docs/scaladsl/LogRotatorSinkSpec.scala

@@ -170,6 +170,48 @@ class LogRotatorSinkSpec
 
     }
 
+    "work for stream-based rotation " in assertAllStagesStopped {
+      // #stream
+      val destinationDir = FileSystems.getDefault.getPath("/tmp")
+
+      val streamBasedTriggerCreator: () => ((String, String)) => Option[Path] = () => {
+        var currentFilename: Option[String] = None
+        (element: (String, String)) => {
+          if (currentFilename.contains(element._1)) {
+            None
+          } else {
+            currentFilename = Some(element._1)
+            Some(destinationDir.resolve(element._1))
+          }
+        }
+      }
+
+      val timeBasedSink: Sink[(String, String), Future[Done]] =
+        LogRotatorSink.withTypedSinkFactory(
+          streamBasedTriggerCreator,
+          (path: Path) =>
+            Flow[(String, String)]
+              .map { case (_, data) => ByteString(data) }
+              .via(Compression.gzip)
+              .toMat(FileIO.toPath(path))(Keep.right)
+        )
+      // #stream
+
+      val timeBaseCompletion = Source(
+        immutable.Seq(
+          ("stream1", "test1"),
+          ("stream1", "test2"),
+          ("stream1", "test3"),
+          ("stream2", "test4"),
+          ("stream2", "test5"),
+          ("stream2", "test6")
+        )
+      ).runWith(timeBasedSink)
+
+      timeBaseCompletion.futureValue shouldBe Done
+
+    }
+
     "write lines to a single file" in assertAllStagesStopped {
       var files = Seq.empty[Path]
       val triggerFunctionCreator = () => {