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.

Artifacts

The Akka dependencies are available from Akka’s library repository. To access them there, you need to configure the URL for this repository.

sbt

resolvers += "Akka library repository".at("https://repo.akka.io/maven")

Maven

<project>
  ...
  <repositories>
    <repository>
      <id>akka-repository</id>
      <name>Akka library repository</name>
      <url>https://repo.akka.io/maven</url>
    </repository>
  </repositories>
</project>

Gradle

repositories {
    mavenCentral()
    maven {
        url "https://repo.akka.io/maven"
    }
}

Additionally, add the dependencies as below.

sbt

val AkkaVersion = "2.10.0"
libraryDependencies ++= Seq(
  "com.lightbend.akka" %% "akka-stream-alpakka-ftp" % "9.0.0",
  "com.typesafe.akka" %% "akka-stream" % AkkaVersion
)

Maven

<properties>
  <akka.version>2.10.0</akka.version>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencies>
  <dependency>
    <groupId>com.lightbend.akka</groupId>
    <artifactId>akka-stream-alpakka-ftp_${scala.binary.version}</artifactId>
    <version>9.0.0</version>
  </dependency>
  <dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-stream_${scala.binary.version}</artifactId>
    <version>${akka.version}</version>
  </dependency>
</dependencies>

Gradle

def versions = [
  AkkaVersion: "2.10.0",
  ScalaBinary: "2.13"
]
dependencies {
  implementation "com.lightbend.akka:akka-stream-alpakka-ftp_${versions.ScalaBinary}:9.0.0"
  implementation "com.typesafe.akka:akka-stream_${versions.ScalaBinary}:${versions.AkkaVersion}"
}

The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.

Direct dependencies

Organization	Artifact	Version
com.hierynomus	sshj	0.39.0
com.typesafe.akka	akka-stream_2.13	2.10.0
commons-net	commons-net	3.11.1
org.scala-lang	scala-library	2.13.12

Dependency tree

com.hierynomus    sshj    0.39.0    The Apache License, Version 2.0
    com.hierynomus    asn-one    0.6.0    The Apache License, Version 2.0
        org.slf4j    slf4j-api    2.0.13
    net.i2p.crypto    eddsa    0.3.0    CC0 1.0 Universal
    org.bouncycastle    bcpkix-jdk18on    1.78.1    Bouncy Castle Licence
        org.bouncycastle    bcprov-jdk18on    1.78.1    Bouncy Castle Licence
        org.bouncycastle    bcutil-jdk18on    1.78.1    Bouncy Castle Licence
            org.bouncycastle    bcprov-jdk18on    1.78.1    Bouncy Castle Licence
    org.bouncycastle    bcprov-jdk18on    1.78.1    Bouncy Castle Licence
    org.slf4j    slf4j-api    2.0.13
com.typesafe.akka    akka-stream_2.13    2.10.0    BUSL-1.1
    com.typesafe.akka    akka-actor_2.13    2.10.0    BUSL-1.1
        com.typesafe    config    1.4.3    Apache-2.0
        org.scala-lang    scala-library    2.13.12    Apache-2.0
    com.typesafe.akka    akka-protobuf-v3_2.13    2.10.0    BUSL-1.1
    org.reactivestreams    reactive-streams    1.0.4    MIT-0
    org.scala-lang    scala-library    2.13.12    Apache-2.0
commons-net    commons-net    3.11.1
org.scala-lang    scala-library    2.13.12    Apache-2.0

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

copysourceval ftpSettings = FtpSettings
  .create(InetAddress.getByName(HOSTNAME))
  .withPort(PORT)
  .withCredentials(CREDENTIALS)
  .withBinary(true)
  .withPassiveMode(true)
  // only useful for debugging
  .withConfigureConnection((ftpClient: FTPClient) => {
    ftpClient.addProtocolCommandListener(new PrintCommandListener(new PrintWriter(System.out), true))
  })

Java

copysourceimport akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.javadsl.Source;
import org.apache.commons.net.PrintCommandListener;
import org.apache.commons.net.ftp.FTPClient;
import java.net.InetAddress;

FtpSettings ftpSettings =
    FtpSettings.create(InetAddress.getByName(HOSTNAME))
        .withPort(PORT)
        .withCredentials(CREDENTIALS)
        .withBinary(true)
        .withPassiveMode(true)
        // only useful for debugging
        .withConfigureConnectionConsumer(
            (FTPClient ftpClient) -> {
              ftpClient.addProtocolCommandListener(
                  new PrintCommandListener(new PrintWriter(System.out), true));
            });

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 via a proxy, please provide an instance of java.net.Proxy by using the withProxy method.

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

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

val sshClient: SSHClient = new SSHClient(new DefaultConfig)
val configuredClient: SftpApi = Sftp(sshClient)

Java

copysource
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);
  }
}

Improving SFTP throughput

For SFTP connections allowing more than one unconfirmed read request to be sent by the client you can use withMaxUnconfirmedReads on SftpSettings
The command-line tool sftp uses a value of 64 by default. This can significantly improve throughput by reducing the impact of latency.

Scala

copysourceimport akka.stream.IOResult
import akka.stream.alpakka.ftp.scaladsl.Sftp
import akka.stream.scaladsl.Source
import akka.util.ByteString

import scala.concurrent.Future

def retrieveFromPath(path: String, settings: SftpSettings): Source[ByteString, Future[IOResult]] =
  Sftp.fromPath(path, settings.withMaxUnconfirmedReads(64))

Java

copysource
import akka.stream.IOResult;
import akka.stream.alpakka.ftp.SftpSettings;
import akka.stream.alpakka.ftp.javadsl.Sftp;
import akka.stream.javadsl.Source;
import akka.util.ByteString;

import java.util.concurrent.CompletionStage;

public class SftpRetrievingExample {

  public Source<ByteString, CompletionStage<IOResult>> retrieveFromPath(
      String path, SftpSettings settings) throws Exception {
    return Sftp.fromPath(path, settings.withMaxUnconfirmedReads(64));
  }
}

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

copysourceimport 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)

Java

copysourceimport akka.actor.ActorSystem;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.alpakka.ftp.javadsl.Ftp;

public class FtpTraversingExample {

  public void listFiles(String basePath, FtpSettings settings, ActorSystem system)
      throws Exception {
    Ftp.ls(basePath, settings)
        .runForeach(ftpFile -> System.out.println(ftpFile.toString()), system);
  }
}

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

copysourceimport 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)

Java

copysourceimport 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);
  }
}

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

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

val result: Future[IOResult] = Source
  .single(ByteString("this is the file contents"))
  .runWith(Ftp.toPath("file.txt", ftpSettings))

// Create a gzipped target file
import akka.stream.scaladsl.Compression
val result: Future[IOResult] = Source
  .single(ByteString("this is the file contents" * 50))
  .via(Compression.gzip)
  .runWith(Ftp.toPath("file.txt.gz", ftpSettings))

Java

copysourceimport akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.IOResult;
import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
import akka.stream.javadsl.Compression;
import akka.stream.testkit.javadsl.StreamTestKit;
import akka.util.ByteString;
import java.util.concurrent.CompletionStage;

CompletionStage<IOResult> result =
    Source.single(ByteString.fromString("this is the file contents"))
        .runWith(Ftp.toPath("file.txt", ftpSettings), materializer);

// Create a gzipped target file
CompletionStage<IOResult> result =
    Source.single(ByteString.fromString("this is the file contents"))
        .via(Compression.gzip())
        .runWith(Ftp.toPath("file.txt.gz", ftpSettings), materializer);

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

copysourceimport 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)

Java

copysourceimport 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);
  }
}

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

copysourceimport 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)

Java

copysourceimport 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);
  }
}

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.

Creating directory

In order to create a directory the user has to specify a parent directory (also known as base path) and directory’s name.

Alpakka provides a materialized API mkdirAsync (based on FutureCompletion Stage) and unmaterialized API mkdir (using Sources) to let the user choose when the action will be executed.

Scala

copysource
import akka.NotUsed
import akka.stream.scaladsl.Source
import akka.stream.alpakka.ftp.scaladsl.Ftp
import akka.Done

def mkdir(basePath: String, directoryName: String, settings: FtpSettings): Source[Done, NotUsed] =
  Ftp.mkdir(basePath, directoryName, settings)

Java

copysourceimport akka.Done;
import akka.NotUsed;
import akka.stream.alpakka.ftp.FtpSettings;
import akka.stream.alpakka.ftp.javadsl.Ftp;
import akka.stream.javadsl.Source;

public class FtpMkdirExample {
  public Source<Done, NotUsed> mkdir(
      String parentPath, String directoryName, FtpSettings settings) {
    return Ftp.mkdir(parentPath, directoryName, settings);
  }
}

Please note that to include a subdirectory in result of ls the emitTraversedDirectories has to be set to true.

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

Scala

copysourceimport java.nio.file.Files

import akka.NotUsed
import akka.stream.alpakka.ftp.scaladsl.Ftp
import akka.stream.scaladsl.{FileIO, RunnableGraph}

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))

Java

copysourceimport 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));
  }
}

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.

```
docker compose up -d ftp sftp
sbt
> ftp/test
```