2

I am using Alpakka-FTP, but maybe I'm looking for a general akka-stream pattern. The FTP connector can list files or retrieve them:

def ls(host: String): Source[FtpFile, NotUsed]
def fromPath(host: String, path: Path): Source[ByteString, Future[IOResult]]

Ideally, I would like to create a stream like this:

LIST
  .FETCH_ITEM
  .FOREACH(do something)

But I'm unable to create such a stream with the two functions I wrote above. I feel like I should be able to get there using a Flow, something like

Ftp.ls
  .via(some flow that uses the Ftp.fromPath above)
  .runWith(Sink.foreach(do something))

Is this possible, given only the ls and fromPath functions above?

EDIT:

I am able to work it out using one actor and mapAsync, but I still feel it should be more straightforward.

class Downloader extends Actor {
  override def receive = {
    case ftpFile: FtpFile =>
      Ftp.fromPath(Paths.get(ftpFile.path), settings)
        .toMat(FileIO.toPath(Paths.get("testHDF.txt")))(Keep.right)
        .run() pipeTo sender
  }
}

val downloader = as.actorOf(Props(new Downloader))

Ftp.ls("test_path", settings)
  .mapAsync(1)(ftpFile => (downloader ? ftpFile) (3.seconds).mapTo[IOResult])
  .runWith(Sink.foreach(res => println("got it!" + res)))
ticofab
  • 7,551
  • 13
  • 49
  • 90

1 Answers1

1

You should be able to use flatMapConcat for this purpose. Your specific example could be re-written to

Ftp.ls("test_path", settings).flatMapConcat{ ftpFile =>
  Ftp.fromPath(Paths.get(ftpFile.path), settings)
}.runWith(FileIO.toPath(Paths.get("testHDF.txt")))

Documentation here.

Stefano Bonetti
  • 8,973
  • 1
  • 25
  • 44