akka-stream-with-springboot-api/src/main/scala/fr/natan/akkastreamfileprocessingapi/service/AkkaStreamComponents.scala

package fr.natan.akkastreamfileprocessingapi.service

import akka.actor.ActorSystem
import akka.stream.javadsl.Framing
import akka.stream.scaladsl.{Compression, FileIO, Flow, Sink, Source}
import akka.util.ByteString
import akka.{Done, NotUsed}
import com.typesafe.scalalogging.slf4j.Logger
import fr.natan.akkastreamfileprocessingapi.businessexceptions.FileNotFoundException
import fr.natan.akkastreamfileprocessingapi.models.Movie
import fr.natan.akkastreamfileprocessingapi.valitator.Validators.fileExists

import java.io.File
import java.nio.file.Paths
import scala.concurrent.Future

object AkkaStreamComponents {

  implicit val actor: ActorSystem = ActorSystem("AkkaStreamActor")

  private def convertToClassMovie(array: Array[String]): Movie = {
    val movie: Movie = Movie(array(0), array(1), array(2), array(3), array(4).toShort, array(5).toInt, array(6), array(7), array(8))
    movie
  }

  //flows

  def buildMovieFlow(movieID: String): Flow[String, Movie, NotUsed] = {

    val flow: Flow[String, Movie, NotUsed] =
      Flow[String].filter(rows => !rows.contains(movieID))
        .map(row => {
          val movie: Array[String] = row.split("\t")
          convertToClassMovie(movie)
        })
    flow

  }

  def filterByMoviePrimaryTitleFlow(moviePrimaryTitle: String): Flow[String, Movie, NotUsed] = {
    val filterFlow: Flow[String, Movie, NotUsed] =
      Flow[String]
        .filter((rows: String) => {
          rows.contains(moviePrimaryTitle)
        })
        .map(row => {
          val movie: Array[String] = row.split("\t")
          convertToClassMovie(movie)
        })

    filterFlow
  }

  //source
  def buildSource(inputFile: File): Source[String, NotUsed] = {

    var source: Source[String, NotUsed] = null

    if (!fileExists(inputFile.getPath)) {
      return null
    }
    source = Source
      .single(inputFile)
      .flatMapConcat(
        (file: File) =>
          FileIO.fromPath(Paths.get(inputFile.getPath)
          )
      )
      .via(Compression.gunzip())
      .via(
        Framing.delimiter(ByteString("\n"), 4096)
        .map(byteString => byteString.utf8String)
      )

    source
  }

  def buildAndValidateSource(inputFile: File): Source[String, NotUsed] = {

    val source: Source[String, NotUsed] = buildSource(inputFile = inputFile)
    if (source == null) {
      throw new FileNotFoundException(filename = inputFile.getPath)
    }

    source
  }

  //sink
  def buildSink(logger: Logger): Sink[Movie, Future[Done]] = {
    val sink = Sink.foreach[Movie](
      (movie: Movie) => logger.info(s"${movie.toString}")
    )
    sink
  }
}