package uk.ac.wellcome.platform.archive.archivist.flow

import akka.NotUsed
import akka.stream.scaladsl.{Flow, Source}
import com.amazonaws.services.s3.AmazonS3
import grizzled.slf4j.Logging
import uk.ac.wellcome.platform.archive.archivist.models.{ArchiveJob, BagUploaderConfig}
import uk.ac.wellcome.platform.archive.common.models.ArchiveComplete

object ArchiveZipFileFlow extends Logging {
  def apply(config: BagUploaderConfig)(
    implicit s3Client: AmazonS3
  ): Flow[ZipFileDownloadComplete, ArchiveComplete, NotUsed] = {

    Flow[ZipFileDownloadComplete].flatMapConcat {
      case ZipFileDownloadComplete(zipFile, ingestRequest) =>
        Source
          .single(zipFile)
          .map(ArchiveJob.create(_, config))
          .collect{case Right(job) => job}
          // TODO: Log error here
          .via(ArchiveJobFlow(config.bagItConfig.digestDelimiterRegexp))
          .collect {case Right(archiveJob) => archiveJob}
          .map(job =>
            ArchiveComplete(
              job.bagLocation,
              ingestRequest
            ))
    }
  }
}
package uk.ac.wellcome.platform.archive.archivist.flow

import akka.NotUsed
import akka.stream.scaladsl.{Flow, Source}
import com.amazonaws.services.s3.AmazonS3
import grizzled.slf4j.Logging
import uk.ac.wellcome.platform.archive.archivist.models.{ArchiveJob, BagUploaderConfig}
import uk.ac.wellcome.platform.archive.common.models.ArchiveComplete

object ArchiveZipFileFlow extends Logging {
  def apply(config: BagUploaderConfig)(
    implicit s3Client: AmazonS3
  ): Flow[ZipFileDownloadComplete, ArchiveComplete, NotUsed] = {

    Flow[ZipFileDownloadComplete].flatMapConcat {
      case ZipFileDownloadComplete(zipFile, ingestRequest) =>
        // TODO report progress here
        Source
          .single(zipFile)
          .map(ArchiveJob.create(_, config))
          .collect{case Right(job) => job}
          .via(ArchiveJobFlow(config.bagItConfig.digestDelimiterRegexp))
          .collect {case Right(archiveJob) => archiveJob}
          .map(job =>
            ArchiveComplete(
              job.bagLocation,
              ingestRequest
            ))
    }
  }
}
Add a TODO to mark where we should add the reporting flow
Add a TODO to mark where we should add the reporting flow
