//设置package包名称以及导入依赖的类
package com.stacktrace.yo.scrapeline.imdb.pipelines

import java.nio.file.Paths

import akka.NotUsed
import akka.stream.scaladsl.{FileIO, Flow, Keep, Sink, Source}
import akka.stream.{ActorMaterializer, IOResult}
import akka.util.ByteString
import com.stacktrace.yo.scrapeline.core.ScrapeClient.jsoup
import com.stacktrace.yo.scrapeline.core._
import com.stacktrace.yo.scrapeline.imdb.Domain.MovieNameAndDetailUrl
import net.ruippeixotog.scalascraper.dsl.DSL._
import net.ruippeixotog.scalascraper.model.Document
import net.ruippeixotog.scalascraper.scraper.ContentExtractors.elementList

import scala.concurrent.Future

class MovieListPipeline(implicit val m: ActorMaterializer) {

  def getPipelineSource: Source[jsoup.DocumentType, NotUsed] = Source.single(ScrapeClient.scrape("http://www.the-numbers.com/movie/budgets/all"))

  def getParseFlow: Flow[Document, MovieNameAndDetailUrl, NotUsed] = {
    Flow[Document]
      .mapConcat(doc => {
        val table = doc >> elementList("table tr")
        val movieLinkTuples = table.flatMap(tr => {
          val name = tr >> elementList("tr b a")
          name.map(
            link => {
              MovieNameAndDetailUrl(link.text, "http://www.the-numbers.com/" + link.attr("href"))
            }
          )
        })
        movieLinkTuples
      })
  }

  def getPipeOut: Sink[MovieNameAndDetailUrl, Future[IOResult]] = Flow[MovieNameAndDetailUrl]
    .map(s => ByteString(s.name + "\n"))
    .toMat(FileIO.toPath(Paths.get("movie.txt")))(Keep.right)

  def buildAndRun: Future[IOResult] = {
    getPipelineSource
      .via(getParseFlow)
      .runWith(getPipeOut)
  }

} 

//设置package包名称以及导入依赖的类
package com.martinseeler.dtf

import java.io.File
import java.nio.file.StandardOpenOption._

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Framing}
import akka.util.ByteString
import com.martinseeler.dtf.stages.{DeltaToByteStringStage, TickToDeltaStage}

object SerializationDemo extends App {

  implicit val system = ActorSystem("serialization")
  implicit val mat = ActorMaterializer()

  val csvSource = FileIO
    .fromPath(new File("ticks_1MM.csv").toPath)
    .via(Framing.delimiter(ByteString("\n"), 1024))
    .map(_.utf8String.split(','))
    .map(xs => Tick(xs(0).toLong, xs(1).toDouble, xs(2).toDouble))

  val dtffSink = FileIO
    .toPath(new File("ticks_1MM.dtff").toPath, options = Set(CREATE, WRITE, TRUNCATE_EXISTING))

  csvSource
    .via(new TickToDeltaStage())
    .via(new DeltaToByteStringStage())
    .runWith(dtffSink)
    .onComplete(_ => system.terminate())(scala.concurrent.ExecutionContext.global)
} 

//设置package包名称以及导入依赖的类
package com.martinseeler.dtf

import java.io.File

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Source}
import com.martinseeler.dtf.stages.{ByteStringToDeltaStage, DeltaToTickStage}

object DeserializationDemo extends App {

  implicit val system = ActorSystem("deserialization")
  implicit val mat = ActorMaterializer()

  val dtffSource = FileIO.fromPath(new File("ticks_1MM.dtff").toPath)

  dtffSource
    .via(new ByteStringToDeltaStage())
    .via(new DeltaToTickStage())
    .runForeach(println)
    .onComplete(_ => system.terminate())(
        scala.concurrent.ExecutionContext.global)
} 

//设置package包名称以及导入依赖的类
package scaladays.akka.stream

import java.io.File

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Source}
import akka.util.ByteString

import scaladays.akka.domain.Tweet
import scaladays.akka.http.MyJsonProtocol
import scaladays.akka.support.MakingUpData

object Step0GenerateStatuses extends App
  with MakingUpData with MyJsonProtocol {

  implicit val system = ActorSystem()
  implicit val ec = system.dispatcher
  implicit val mat = ActorMaterializer()

  val n = 10000

  Source.repeat(NotUsed).take(n)
    .map(_ => Tweet.random)
    .mapAsync(1)(t => Marshal(t).to[ByteString])
    .intersperse(ByteString("\n"))
    .runWith(FileIO.toPath(new File("tweets.json").toPath))
    .onComplete { res =>
      println(s"Generated $n tweets. ($res)")
      system.terminate()
    }

} 

//设置package包名称以及导入依赖的类
package im.actor.util.http

import java.nio.file.{ Files, Path }

import akka.stream.scaladsl.FileIO

import scala.concurrent._

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpRequest
import akka.stream.Materializer

import scala.util.{ Success, Failure }

class DownloadManager(implicit system: ActorSystem, materializer: Materializer) {
  implicit val ec: ExecutionContext = system.dispatcher

  val http = Http()

  def download(url: String): Future[(Path, Long)] = {
    val tempFileFuture = createTempFile()
    val responseFuture = http.singleRequest(HttpRequest(uri = url))

    for {
      filePath ? tempFileFuture
      response ? responseFuture
      ioRes ? response.entity.dataBytes.runWith(FileIO.toPath(filePath))
    } yield {
      ioRes.status match {
        case Success(_)     ? (filePath, ioRes.count)
        case Failure(cause) ? throw cause
      }
    }
  }

  // FIXME: dispatcher for this
  private def createTempFile(): Future[Path] = {
    Future {
      blocking {
        Files.createTempFile("ActorDownloadManager", "")
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.stacktrace.yo.scrapeline.imdb.pipelines

import java.net.URLEncoder
import java.nio.file.Paths

import akka.stream.scaladsl.{FileIO, Flow, Framing, Source}
import akka.stream.{ActorMaterializer, IOResult}
import akka.util.ByteString
import akka.{Done, NotUsed}
import com.stacktrace.yo.scrapeline.imdb.Domain.MovieNameAndDetailUrl

import scala.concurrent.{ExecutionContext, Future}

class MovieNameToIMDBPipeline(implicit val m: ActorMaterializer, implicit val ec: ExecutionContext) {

  private def getPipelineSource: Source[String, Future[IOResult]] = {
    FileIO.fromPath(Paths.get("movie.txt"))
      .via(Framing.delimiter(ByteString("\n"), 256)
        .map(_.utf8String)
      )
  }

  private def getDetailUrlFlow: Flow[String, MovieNameAndDetailUrl, NotUsed] = {
    Flow[String]
      .mapAsyncUnordered(100)(mapPipeToImdbSearch)
  }


  private def mapPipeToImdbSearch(in: String): Future[MovieNameAndDetailUrl] = Future {
    val encodedString: String = URLEncoder.encode(in, "UTF-8")
    MovieNameAndDetailUrl(in, "http://www.imdb.com/find?ref_=nv_sr_fn&q=" + encodedString + "&s=tt")
  }


  def getOutput: Future[Done] = {
    getPipelineSource
      .via(getDetailUrlFlow)
      .via(new IMDBSearchSubPipe().getSubFlow)
      .runForeach(x => println(x.name + ":" + x.url))
  }

} 

//设置package包名称以及导入依赖的类
package csw.services.config.server.files

import java.nio.file.{Path, Paths}

import akka.stream.scaladsl.{FileIO, Keep}
import csw.services.config.api.models.ConfigData
import csw.services.config.server.commons.ConfigServerLogger
import csw.services.config.server.{ActorRuntime, Settings}

import scala.async.Async._
import scala.concurrent.Future


  def validate(id: String, path: Path): Future[Boolean] = async {
    id == await(Sha1.fromPath(path))
  }

  def saveAndSha(configData: ConfigData): Future[(Path, String)] = async {
    val path = await(fileRepo.createTempFile("config-service-overize-", ".tmp"))
    val (resultF, shaF) = configData.source
      .alsoToMat(FileIO.toPath(path))(Keep.right)
      .toMat(Sha1.sink)(Keep.both)
      .run()
    await(resultF).status.get
    (path, await(shaF))
  }

} 

//设置package包名称以及导入依赖的类
package csw.services.config.server.files

import java.nio.file.Path
import java.security.MessageDigest

import akka.stream.Materializer
import akka.stream.scaladsl.{FileIO, Flow, Keep, Sink, Source}
import akka.util.ByteString
import csw.services.config.api.models.ConfigData

import scala.concurrent.Future


object Sha1 {

  private def fromSource(source: Source[ByteString, Any])(implicit mat: Materializer): Future[String] =
    source.runWith(sink)

  def fromConfigData(configData: ConfigData)(implicit mat: Materializer): Future[String] =
    fromSource(configData.source)

  def fromPath(path: Path)(implicit mat: Materializer): Future[String] =
    fromSource(FileIO.fromPath(path))

  //Keep this a def so that the digester is created anew each time.
  def sink: Sink[ByteString, Future[String]] = {
    val sha1Digester = MessageDigest.getInstance("SHA-1")
    Flow[ByteString]
      .fold(sha1Digester) { (digester, bs) =>
        digester.update(bs.toArray)
        digester
      }
      .mapConcat(_.digest().toList)
      .map(_ & 0xFF)
      .map("%02x" format _)
      .toMat(Sink.fold("")(_ + _))(Keep.right)
  }

} 

//设置package包名称以及导入依赖的类
package eu.svez.backpressuredemo.B_http

import java.nio.file.Paths

import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.scaladsl.FileIO
import eu.svez.backpressuredemo.Flows._
import eu.svez.backpressuredemo.StreamDemo

object FileClient extends StreamDemo {

  val byteSource = FileIO
    .fromPath(Paths.get("/tmp/bigfile.zip"))
    .via(valve(sourceRate.get))
    .via(meter("sourceHttp"))

  val host = "localhost"
  val port = 8080

  val request = HttpRequest(
    uri = Uri(s"http://$host:$port/file"),
    entity = HttpEntity(ContentTypes.`application/octet-stream`, byteSource)
  )

  sourceRate.send(5)

  Http().singleRequest(request).onComplete{ _ =>
    println("All sent!")
  }

  readRatesFromStdIn()
} 

//设置package包名称以及导入依赖的类
package com.flipkart.connekt.receptors.directives

import java.io.File

import akka.Done
import akka.http.scaladsl.model.{HttpEntity, MediaTypes, Multipart}
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.server._
import akka.stream.Materializer
import akka.stream.scaladsl.FileIO
import com.flipkart.connekt.commons.utils.StringUtils._
import com.flipkart.connekt.receptors.directives.MultiPartFormData._

import scala.concurrent.{ExecutionContext, Future}
import scala.util.Try


  def extractFormData: Directive1[Map[Name,  Either[Value, FileInfo]]] = {
    Directive[Tuple1[Map[Name,  Either[Value, FileInfo]]]] { inner =>
      extractMaterializer { implicit mat =>
        extractExecutionContext { implicit ec =>
          uploadFileImpl(mat, ec) { filesFuture =>
            ctx => {
              filesFuture.map(map => inner(Tuple1(map))).flatMap(route => route(ctx))
            }
          }
        }
      }
    }
  }

  def downloadFile(file: String): Route = {
    val f = new File(file)
    val responseEntity = HttpEntity(
      MediaTypes.`application/octet-stream`,
      f.length,
      FileIO.fromFile(f, chunkSize = 262144))
    complete(responseEntity)
  }
}


object MultiPartFormData {

  //form field name
  type Name = String
  type Value = String

  case class FileInfo(fileName: String, tmpFilePath: String, status: Try[Done])
} 

//设置package包名称以及导入依赖的类
package pl.touk.nussknacker.ui.api

import java.io.File

import akka.stream.Materializer
import akka.stream.scaladsl.{FileIO, Source}
import akka.util.ByteString
import com.typesafe.scalalogging.LazyLogging
import pl.touk.nussknacker.ui.api.ProcessAttachmentService.AttachmentToAdd
import pl.touk.nussknacker.ui.db.entity.AttachmentEntity.AttachmentEntityData
import pl.touk.nussknacker.ui.process.repository.ProcessActivityRepository
import pl.touk.nussknacker.ui.security.LoggedUser
import pl.touk.nussknacker.ui.util.CatsSyntax

import scala.concurrent.{ExecutionContext, Future}
import scala.util.control.NonFatal

class ProcessAttachmentService(attachmentsBasePath: String, processActivityRepository: ProcessActivityRepository) extends LazyLogging {

  def saveAttachment(processId: String, processVersionId: Long, originalFileName: String, byteSource: Source[ByteString, Any])
                    (implicit ec: ExecutionContext, loggedUser: LoggedUser, mat: Materializer): Future[Unit] = {
    val relativeFilePath = s"$processId/${s"${System.currentTimeMillis()}-$originalFileName"}"
    val attachmentFile = getAttachmentFile(relativeFilePath)
    attachmentFile.getParentFile.mkdirs()
    val fileSink = FileIO.toFile(attachmentFile)
    byteSource.runWith(fileSink).flatMap { _ =>
      val attachmentToAdd = AttachmentToAdd(processId, processVersionId, originalFileName, relativeFilePath)
      processActivityRepository.addAttachment(attachmentToAdd).recoverWith { case NonFatal(ex) =>
        logger.warn(s"Failure during writing attachment to db. Removing file ${attachmentFile}", ex)
        attachmentFile.delete()
        Future.failed(ex)
      }
    }
  }

  def readAttachment(attachmentId: Long)
                    (implicit ec: ExecutionContext, loggedUser: LoggedUser): Future[Option[(File, AttachmentEntityData)]] = {
    val attachmentFutureOpt = processActivityRepository.findAttachment(attachmentId)
    CatsSyntax.futureOpt.map(attachmentFutureOpt) { attachment =>
      (getAttachmentFile(attachment.filePath), attachment)
    }
  }

  private def getAttachmentFile(attachmentRelativePath: String): File = {
    new File(attachmentsBasePath, attachmentRelativePath)
  }

}

object ProcessAttachmentService {

  case class AttachmentToAdd(processId: String,
                             processVersionId: Long,
                             fileName: String,
                             relativeFilePath: String
                            )

} 

//设置package包名称以及导入依赖的类
package controllers

import javax.inject._

import akka.stream.scaladsl.{FileIO, Source}
import akka.util.ByteString
import play.api.http.HttpEntity
import play.api.mvc._


@Singleton
class HomeController @Inject()(webJarAssets: WebJarAssets) extends Controller {

  def index = {
    Action {
      Ok(views.html.index("Your new application is ready.", webJarAssets))
    }
  }

  def streaming = {
    Action {
      val file = new java.io.File("conf/vid4.mp4")
      val path: java.nio.file.Path = file.toPath
      val source: Source[ByteString, _] = FileIO.fromPath(path)

      Result(
        header = ResponseHeader(200, Map.empty),
        body = HttpEntity.Streamed(source, Some(file.length), Some("application/mp4"))
      )

      
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.knoldus

import java.io.File

import akka.actor.ActorSystem
import akka.http.scaladsl.model.Multipart.BodyPart
import akka.http.scaladsl.model.{HttpResponse, Multipart, StatusCodes}
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.server.Route
import akka.stream.Materializer
import akka.stream.scaladsl.FileIO

import scala.concurrent.duration._
import scala.concurrent.{ExecutionContextExecutor, Future}

trait MultipartFormDataHandler {

  implicit val system: ActorSystem

  implicit def executor: ExecutionContextExecutor

  implicit val materializer: Materializer

  val routes = processMultiPartData

  def processMultiPartData: Route = path("process" / "multipart" / "data") {
    (post & entity(as[Multipart.FormData])) { formData =>
    complete {
    val extractedData: Future[Map[String, Any]] = formData.parts.mapAsync[(String, Any)](1) {

      case file: BodyPart if file.name == "file" => val tempFile = File.createTempFile("process", "file")
        file.entity.dataBytes.runWith(FileIO.toPath(tempFile.toPath)).map { ioResult=>
        s"file ${file.filename.fold("Unknown")(identity)}" -> s"${ioResult.count} bytes"}

      case data: BodyPart => data.toStrict(2.seconds).map(strict =>data.name -> strict.entity.data.utf8String)
    }.runFold(Map.empty[String, Any])((map, tuple) => map + tuple)

     extractedData.map { data => HttpResponse(StatusCodes.OK, entity = s"Data : ${data.mkString(", ")} has been successfully saved.")}
     .recover {
      case ex: Exception =>HttpResponse(StatusCodes.InternalServerError,entity = s"Error in processing multipart form data due to ${ex.getMessage}")
     }
   }
    }
  }
} 

//设置package包名称以及导入依赖的类
package lew.bing.akka.stream

import java.io.ByteArrayInputStream
import java.nio.file.Paths

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, StreamConverters}


object Demo5 {

  def main(args: Array[String]): Unit = {
    //????
    implicit val system = ActorSystem("QuickStart")
    implicit val materializer = ActorMaterializer()
    import scala.concurrent.ExecutionContext.Implicits.global
    //??is
    val source = StreamConverters.fromInputStream(() => new ByteArrayInputStream("I am better man!".getBytes))
    //??os
    val sink = StreamConverters.fromOutputStream(() => System.out)
    val result = source.runWith(sink)
    val result2 =FileIO.fromPath(Paths.get("factorial2.txt")).to(FileIO.toPath(Paths.get("factorial3.txt"))).run()
    result.flatMap(f => result2).onComplete(_ => system.terminate())
  }

} 

//设置package包名称以及导入依赖的类
package com.karasiq.nanoboard.server.utils

import java.nio.file.{Files, Path, Paths}

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.headers.Location
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, Uri}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Source}
import com.karasiq.nanoboard.captcha.storage.{NanoboardCaptchaFileSource, NanoboardCaptchaSource}
import com.typesafe.config.{Config, ConfigFactory}

import scala.concurrent.{ExecutionContext, Future}

object CaptchaLoader {
  def apply(config: Config = ConfigFactory.load())(implicit am: ActorMaterializer, as: ActorSystem, ec: ExecutionContext): CaptchaLoader = {
    new CaptchaLoader(Paths.get(config.getString("nanoboard.captcha.storage")))
  }

  def load(config: Config = ConfigFactory.load())(implicit am: ActorMaterializer, as: ActorSystem, ec: ExecutionContext): Future[NanoboardCaptchaFileSource] = {
    apply(config).forUrl(config.getString("nanoboard.captcha.download-url"))
  }
}

final class CaptchaLoader(baseDir: Path)(implicit am: ActorMaterializer, as: ActorSystem, ec: ExecutionContext) {
  private val http = Http()

  // TODO: https://github.com/akka/akka/issues/15990
  private def requestWithRedirects(uri: Uri): Future[HttpResponse] = {
    http.singleRequest(HttpRequest(uri = uri)).flatMap { response ?
      val location = response.header[Location]
      if (response.status.isRedirection() && location.isDefined) {
        requestWithRedirects(location.get.uri)
      } else {
        Future.successful(response)
      }
    }
  }

  def forUrl(url: String): Future[NanoboardCaptchaFileSource] = {
    val fileName = baseDir.resolve(s"${Integer.toHexString(url.hashCode)}.nbc")
    if (Files.exists(fileName)) {
      assert(Files.isRegularFile(fileName), s"Not a file: $fileName")
      Future.successful(NanoboardCaptchaSource.fromFile(fileName.toString))
    } else {
      Source
        .fromFuture(requestWithRedirects(url))
        .flatMapConcat(_.entity.dataBytes)
        .toMat(FileIO.toFile(fileName.toFile))((_, r) ? r.map { ioResult ?
          if (ioResult.wasSuccessful) NanoboardCaptchaSource.fromFile(fileName.toString)
          else throw ioResult.getError
        })
        .run()
    }
  }
} 

//设置package包名称以及导入依赖的类
package sandbox

import org.scalacheck.Gen
import sandbox.util.DataGen
import java.nio.file.Files
import java.time.LocalDateTime
import java.io.File
import java.io.PrintWriter
import akka.stream.scaladsl.Source
import akka.actor.ActorSystem
import scala.concurrent.Await
import scala.concurrent.duration._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Keep
import scala.concurrent.Future
import akka.stream.scaladsl.FileIO
import java.nio.file.Path

object StreamFromFilesInDirectory extends App {

  implicit val system = ActorSystem("StreamFromFilesInDirectory")
  implicit val ec = system.dispatcher
  implicit val mat = ActorMaterializer()

  val dataDir = Files.createTempDirectory(s"").toFile()
  dataDir.deleteOnExit()

  println(dataDir.getAbsolutePath)

  try {
    val names = Gen.listOfN(100, DataGen.name("US")).sample.get

    names.zipWithIndex.foreach{ case (contents,idx) =>
      val file = new File(dataDir,s"File$idx.txt")
      file.deleteOnExit()

      val pw = new PrintWriter(file)
      pw.write(contents)
      pw.close
    }

    val source = Source(dataDir.list().toList)

    val done =
      source
        .map(fn => new File(dataDir,fn))
        .flatMapConcat(fileSource)
        .toMat(Sink.seq)(Keep.right).run

    waitFor(done).foreach(println)
  }
  finally {
    waitFor(system.terminate())
  }

  def fileSource(file: File) = FileIO.fromPath(file.toPath).map(_.decodeString("UTF-8"))

  def waitFor[T](f: Future[T],d: Duration= 10.seconds): T = Await.result(f,d)
} 

//设置package包名称以及导入依赖的类
package services

import java.io.File
import java.nio.file.{Files, Path}

import akka.stream.scaladsl.FileIO
import play.api.http.{HeaderNames, HttpEntity, HttpProtocol, MimeTypes}
import play.api.mvc.{Request, ResponseHeader, Result, Results}


trait FileUploadingService {

  def sendFile(name: String, filePath: Path)(implicit request: Request[_]): Result = {
    val file = new File(filePath.toString)

    if (file.exists()) {
      val source = FileIO.fromPath(file.toPath)
      val contentType = Option(Files.probeContentType(filePath)).getOrElse(MimeTypes.BINARY)

      val headers = Seq(
        HeaderNames.CONTENT_DISPOSITION -> s"attachment; filename=$name",
        HeaderNames.CONTENT_TYPE -> contentType
      )

      request.version match {
        case HttpProtocol.HTTP_1_0 =>
          val size = file.length()
          val httpEntity = HttpEntity.Streamed(source, Some(size), Some(contentType))

          Result(
            header = ResponseHeader(200, Map(HeaderNames.CONTENT_LENGTH -> size.toString) ++ headers.toMap),
            body = httpEntity
          )
        case HttpProtocol.HTTP_1_1 =>
          Results.Ok.chunked(source).withHeaders(headers: _*)
      }
    } else {
      Results.NotFound
    }
  }

} 

//设置package包名称以及导入依赖的类
package io.scalac.newspaper.crawler.urls

import java.nio.file.Path

import akka.NotUsed
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Source}
import akka.util.ByteString

import scala.collection.immutable
import scala.concurrent.{ExecutionContext, Future}
import scala.io

class FileURLsStore(urlsFilePath: Path)(implicit ec: ExecutionContext, materializer: ActorMaterializer) extends URLsStore {

  private var urls: immutable.Seq[String] = readData(urlsFilePath)

  private def readData(path: Path) =
    io.Source.fromFile(urlsFilePath.toFile).getLines().to[immutable.Seq]

  override def getURLs: Source[String, NotUsed] =
    Source(urls)

  def removeURL(url: String): Future[Boolean] = {
    urls = urls.filterNot(_ == url)

    Source(urls)
     .map(s => ByteString(s + "\n"))
      .runWith(FileIO.toPath(urlsFilePath))
      .map(_ => true)
  }
} 

//设置package包名称以及导入依赖的类
package io.scalac.newspaper.crawler.urls

import java.nio.file.Paths

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Framing, Sink, Source}
import akka.testkit.TestKit
import akka.util.ByteString
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}

import scala.collection.immutable
import scala.concurrent.Future


class FileURLsStoreSpec extends TestKit(ActorSystem("test-system")) with WordSpecLike with BeforeAndAfterAll with Matchers with ScalaFutures {

  override def afterAll(): Unit = TestKit.shutdownActorSystem(system)

  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()

  "FileURLsStore" should {
    "be able to delete url from file" in {
      val path = Paths.get("src/test/resources/urls")
      val google = "http://google.pl"
      val scalac = "http://blog.scalac.io"

      val urls = immutable.Seq(google, scalac)

      val resultF = for {
        _ <- Source(urls)
          .map(s => ByteString(s + "\n"))
          .runWith(FileIO.toPath(path))
        cut <- Future.successful(new FileURLsStore(path))
        _ <- cut.removeURL(scalac)
        urlsFromStore <- cut.getURLs.runWith(Sink.seq)
        urlsFromFile <- FileIO.fromPath(path).via(Framing.delimiter(ByteString("\n"), 1024, true)).map(_.utf8String).runWith(Sink.seq)
      } yield(urlsFromFile, urlsFromStore)

      val (urlsFromFile, urlsFromStore) = resultF.futureValue

      urlsFromFile shouldEqual urlsFromStore
      urlsFromFile shouldEqual Vector(google)

    }
  }
} 

//设置package包名称以及导入依赖的类
package code

import java.nio.file.FileSystems

import akka.NotUsed
import akka.stream.IOResult
import akka.stream.scaladsl.{FileIO, Flow, Framing, Keep, RunnableGraph}
import akka.util.ByteString

import scala.concurrent.Future

object BlueprintExample extends AkkaStreamsApp {
  val file = this.getClass.getClassLoader.getResource("current_inventory.csv")
  val inPath = FileSystems.getDefault.getPath(file.getPath)
  val outPath = FileSystems.getDefault.getPath("no_inventory.csv")
  val fileSource = FileIO.fromPath(inPath)
  val fileSink = FileIO.toPath(outPath)

  val csvHandler: Flow[String, List[String], NotUsed] =
    Flow[String]
      .drop(1)
      .map(_.split(",").toList)
      .log("csvHandler")

  val lowInventoryFlow: RunnableGraph[Future[IOResult]] =
    fileSource
      .via(Framing.delimiter(ByteString("\n"), Integer.MAX_VALUE))
      .map(_.utf8String)
      .log("Before CSV Handler")
      .via(csvHandler)
      .filter(list => list(2).toInt == 0)
      .log("After filter")
      .map { list =>
        ByteString(list.mkString(",")) ++ ByteString("\n")
      }.toMat(fileSink)(Keep.right)

  override def akkaStreamsExample: Future[_] =
    lowInventoryFlow.run()

  runExample
}