FTP

The FTP connector provides Akka Stream sources to connect to FTP, FTPs and SFTP servers. Currently, two kinds of sources are provided:

  • one for browsing or traversing the server recursively and,
  • another for retrieving files as a stream of bytes.

Reported issues

Tagged issues at Github

Artifacts

sbt
libraryDependencies += "com.lightbend.akka" %% "akka-stream-alpakka-ftp" % "0.20"
Maven
<dependency>
  <groupId>com.lightbend.akka</groupId>
  <artifactId>akka-stream-alpakka-ftp_2.12</artifactId>
  <version>0.20</version>
</dependency>
Gradle
dependencies {
  compile group: 'com.lightbend.akka', name: 'akka-stream-alpakka-ftp_2.12', version: '0.20'
}

Usage

Configuring the connection settings

In order to establish a connection with the remote server, you need to provide a specialized version of a RemoteFileSettings instance. It’s specialized as it depends on the kind of server you’re connecting to: FTP, FTPs or SFTP.

Scala
import akka.stream.alpakka.ftp.FtpSettings
import akka.stream.alpakka.ftp.FtpCredentials.AnonFtpCredentials
import org.apache.commons.net.PrintCommandListener
import org.apache.commons.net.ftp.FTPClient
import java.io.PrintWriter
import java.net.InetAddress

val settings = FtpSettings(
  InetAddress.getByName("localhost"),
  credentials = AnonFtpCredentials,
  binary = true,
  passiveMode = true,
  configureConnection = (ftpClient: FTPClient) => {
    ftpClient.addProtocolCommandListener(new PrintCommandListener(new PrintWriter(System.out), true))
  }
)
Full source at GitHub
Java
import akka.stream.alpakka.ftp.FtpCredentials;
import akka.stream.alpakka.ftp.FtpSettings;
import org.apache.commons.net.PrintCommandListener;
import org.apache.commons.net.ftp.FTPClient;

FtpSettings.create(InetAddress.getByName("localhost"))
    .withPort(FtpSettings.DefaultFtpPort())
    .withCredentials(FtpCredentials.createAnonCredentials())
    .withBinary(false)
    .withPassiveMode(false)
    .withConfigureConnectionConsumer(
        (FTPClient ftpClient) -> {
          ftpClient.addProtocolCommandListener(
              new PrintCommandListener(new PrintWriter(System.out), true));
        });
create-settings
Full source at GitHub

The configuration above will create an anonymous connection with a remote FTP server in passive mode. For both FTPs and SFTP servers, you will need to provide the specialized versions of these settings: FtpsSettings or SftpSettings respectively.

The example demonstrates optional use of configureConnection option available on FTP and FTPs clients. Use it to configure any custom parameters the server may require, such as explicit or implicit data transfer encryption.

For non-anonymous connection, please provide an instance of NonAnonFtpCredentials instead.

For connection using a private key, please provide an instance of SftpIdentity to SftpSettings.

In order to use a custom SSH client for SFTP please provide an instance of SSHClient.

Scala
import akka.stream.alpakka.ftp.scaladsl.{Sftp, SftpApi}
import net.schmizz.sshj.DefaultConfig
import net.schmizz.sshj.SSHClient

val sshClient: SSHClient = new SSHClient(new DefaultConfig)
val configuredClient: SftpApi = Sftp(sshClient)
Full source at GitHub
Java

import akka.stream.alpakka.ftp.javadsl.Sftp; import akka.stream.alpakka.ftp.javadsl.SftpApi; import net.schmizz.sshj.DefaultConfig; import net.schmizz.sshj.SSHClient; public class ConfigureCustomSSHClient { public ConfigureCustomSSHClient() { SSHClient sshClient = new SSHClient(new DefaultConfig()); SftpApi sftp = Sftp.create(sshClient); } }
Full source at GitHub

Traversing a remote FTP folder recursively

In order to traverse a remote folder recursively, you need to use the ls method in the FTP API:

Scala
import akka.NotUsed
import akka.stream.alpakka.ftp.scaladsl.Ftp
import akka.stream.scaladsl.Source

def listFiles(basePath: String, settings: FtpSettings): Source[FtpFile, NotUsed] =
  Ftp.ls(basePath, settings)
Full source at GitHub
Java
import akka.NotUsed;
import akka.stream.alpakka.ftp.FtpFile;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.javadsl.Source;

public class FtpTraversingExample {

  public Source<FtpFile, NotUsed> listFiles(String basePath, FtpSettings settings)
      throws Exception {
    return Ftp.ls(basePath, settings);
  }
}
Full source at GitHub

This source will emit FtpFile elements with no significant materialization.

For both FTPs and SFTP servers, you will need to use the FTPs and SFTP API respectively.

Retrieving files

In order to retrieve a remote file as a stream of bytes, you need to use the fromPath method in the FTP API:

Scala
import akka.stream.IOResult
import akka.stream.alpakka.ftp.scaladsl.Ftp
import akka.stream.scaladsl.Source
import akka.util.ByteString
import scala.concurrent.Future

def retrieveFromPath(path: String, settings: FtpSettings): Source[ByteString, Future[IOResult]] =
  Ftp.fromPath(path, settings)
Full source at GitHub
Java
import akka.stream.IOResult;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.javadsl.Source;
import akka.util.ByteString;
import java.util.concurrent.CompletionStage;

public class FtpRetrievingExample {

  public Source<ByteString, CompletionStage<IOResult>> retrieveFromPath(
      String path, FtpSettings settings) throws Exception {
    return Ftp.fromPath(path, settings);
  }
}
Full source at GitHub

This source will emit ByteString elements and materializes to Future in Scala API and CompletionStage in Java API of IOResult when the stream finishes.

For both FTPs and SFTP servers, you will need to use the FTPs and SFTP API respectively.

Writing files

In order to store a remote file from a stream of bytes, you need to use the toPath method in the FTP API:

Scala
import akka.stream.IOResult
import akka.stream.alpakka.ftp.scaladsl.Ftp
import akka.stream.scaladsl.Sink
import akka.util.ByteString
import scala.concurrent.Future

def storeToPath(path: String, settings: FtpSettings, append: Boolean): Sink[ByteString, Future[IOResult]] =
  Ftp.toPath(path, settings, append)
Full source at GitHub
Java
import akka.stream.IOResult;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.javadsl.Sink;
import akka.util.ByteString;

import java.util.concurrent.CompletionStage;

public class FtpWritingExample {

  public Sink<ByteString, CompletionStage<IOResult>> storeToPath(String path, FtpSettings settings)
      throws Exception {
    return Ftp.toPath(path, settings);
  }
}
Full source at GitHub

This sink will consume ByteString elements and materializes to Future in Scala API and CompletionStage in Java API of IOResult when the stream finishes.

For both FTPs and SFTP servers, you will need to use the FTPs and SFTP API respectively.

Removing files

In order to remove a remote file, you need to use the remove method in the FTP API:

Scala
import akka.stream.IOResult
import akka.stream.alpakka.ftp.scaladsl.Ftp
import akka.stream.scaladsl.Sink
import scala.concurrent.Future

def remove(settings: FtpSettings): Sink[FtpFile, Future[IOResult]] =
  Ftp.remove(settings)
Full source at GitHub
Java
import akka.stream.IOResult;
import akka.stream.alpakka.ftp.FtpFile;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.javadsl.Sink;
import java.util.concurrent.CompletionStage;

public class FtpRemovingExample {

  public Sink<FtpFile, CompletionStage<IOResult>> remove(FtpSettings settings) throws Exception {
    return Ftp.remove(settings);
  }
}
Full source at GitHub

This sink will consume FtpFile elements and materializes to Future in Scala API and CompletionStage in Java API of IOResult when the stream finishes.

Moving files

In order to move a remote file, you need to use the move method in the FTP API. The move method takes a function to calculate the path to which the file should be moved based on the consumed FtpFile.

Scala
import akka.stream.IOResult
import akka.stream.alpakka.ftp.scaladsl.Ftp
import akka.stream.scaladsl.Sink
import scala.concurrent.Future

def move(destinationPath: FtpFile => String, settings: FtpSettings): Sink[FtpFile, Future[IOResult]] =
  Ftp.move(destinationPath, settings)
Full source at GitHub
Java
import akka.stream.IOResult;
import akka.stream.alpakka.ftp.FtpFile;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.javadsl.Sink;
import java.util.concurrent.CompletionStage;
import java.util.function.Function;

public class FtpMovingExample {

  public Sink<FtpFile, CompletionStage<IOResult>> move(
      Function<FtpFile, String> destinationPath, FtpSettings settings) throws Exception {
    return Ftp.move(destinationPath, settings);
  }
}
Full source at GitHub

This sink will consume FtpFile elements and materializes to Future in Scala API and CompletionStage in Java API of IOResult when the stream finishes.

Typical use-case for this would be listing files from a ftp location, do some processing and move the files when done. An example of this use case can be found below.

Example: downloading files from an FTP location and move the original files

Scala
import akka.NotUsed
import akka.stream.alpakka.ftp.scaladsl.Ftp
import akka.stream.scaladsl.FileIO
import akka.stream.scaladsl.RunnableGraph
import java.nio.file.Files

def processAndMove(sourcePath: String,
                   destinationPath: FtpFile => String,
                   settings: FtpSettings): RunnableGraph[NotUsed] =
  Ftp
    .ls(sourcePath, settings)
    .flatMapConcat(ftpFile => Ftp.fromPath(ftpFile.path, settings).map((_, ftpFile)))
    .alsoTo(FileIO.toPath(Files.createTempFile("downloaded", "tmp")).contramap(_._1))
    .to(Ftp.move(destinationPath, settings).contramap(_._2))
Full source at GitHub
Java
import akka.NotUsed;
import akka.japi.Pair;
import akka.stream.alpakka.ftp.FtpFile;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.javadsl.FileIO;
import akka.stream.javadsl.RunnableGraph;

import java.nio.file.Files;
import java.util.function.Function;

public class FtpProcessAndMoveExample {

  public RunnableGraph<NotUsed> processAndMove(
      String sourcePath, Function<FtpFile, String> destinationPath, FtpSettings settings)
      throws Exception {
    return Ftp.ls(sourcePath, settings)
        .flatMapConcat(
            ftpFile ->
                Ftp.fromPath(ftpFile.path(), settings).map(data -> new Pair<>(data, ftpFile)))
        .alsoTo(FileIO.toPath(Files.createTempFile("downloaded", "tmp")).contramap(Pair::first))
        .to(Ftp.move(destinationPath, settings).contramap(Pair::second));
  }
}
Full source at GitHub

Running the example code

The code in this guide is part of runnable tests of this project. You are welcome to browse the code, edit and run it in sbt.

Scala
sbt
> ftp/test
Java
sbt
> ftp/test
Warning

When using the SFTP API, take into account that JVM relies on /dev/random for random number generation by default. This might potentially block the process on some operating systems as /dev/random waits for a certain amount of entropy to be generated on the host machine before returning a result. In such case, please consider providing the parameter -Djava.security.egd = file:/dev/./urandom into the execution context. Further information can be found here.

The source code for this page can be found here.