Exporter des fichiers vers Cloudflare R2 avec Scala et le SDK AWS
Utilisez le SDK AWS pour Java v2 pour exporter des fichiers locaux finalisés depuis une application Scala vers un bucket Cloudflare R2 privé. Cet exemple propose un programme en ligne de commande avec des modes de téléversement synchrone, asynchrone, en flux FS2 et parallèle, ainsi qu’un moyen de vérifier les octets stockés.
Pourquoi choisir Cloudflare R2 ?
L’API compatible S3 de R2 permet au code Scala d’utiliser la signature des requêtes et les clients HTTP du SDK Java. L’exemple cible un bucket sans restriction de juridiction via le point de terminaison standard du compte. Les buckets propres à une juridiction nécessitent le point de terminaison correspondant.
Configurer votre projet Scala
Cette procédure cible les applications Scala 2.13 existantes qui utilisent sbt 1.x. Les versions fixées et testées sont Scala 2.13.18, sbt 1.13.0, Cats Effect 3.7.1, FS2 3.14.0 et AWS SDK 2.56.0. Scala 3 et sbt 2 constituent des choix de migration distincts ; ces versions fixes maintiennent l’exemple dans la famille de compilation Scala 2.13.
Utilisez Bash, cURL et une
installation d’OpenJDK 21 bénéficiant d’un support actif, avec
java dans votre PATH. La version corrective du JDK
testée est 21.0.12.1 ; utilisez la mise à jour de sécurité actuelle de votre distributeur de JDK pour
la version 21. Vous n’avez besoin ni d’un compilateur Scala ni d’une commande sbt installés
séparément. Exécutez ceci depuis un répertoire accessible en écriture pour créer un projet ; le
sous-shell laisse votre répertoire courant inchangé :
(
set -eu
command -v java >/dev/null
command -v curl >/dev/null
java -version
mkdir r2-scala
cd r2-scala
mkdir -p project src/main/scala
curl -fsSLo sbt-launch.jar https://repo.maven.apache.org/maven2/org/scala-sbt/sbt-launch/1.13.0/sbt-launch-1.13.0.jar
)
Si la création échoue parce que r2-scala existe, choisissez un autre répertoire
parent. Si le téléchargement échoue, conservez le projet et réessayez uniquement la commande cURL
depuis celui-ci. Enregistrez les fichiers suivants dans r2-scala. Utilisez
project/build.properties pour fixer la version du lanceur :
sbt.version=1.13.0
Enregistrez build.sbt :
scalaVersion := "2.13.18"
libraryDependencies ++= Seq(
"software.amazon.awssdk" % "s3" % "2.56.0",
"software.amazon.awssdk" % "netty-nio-client" % "2.56.0",
"software.amazon.awssdk" % "apache-client" % "2.56.0",
"org.typelevel" %% "cats-effect" % "3.7.1",
"co.fs2" %% "fs2-io" % "3.14.0",
"co.fs2" %% "fs2-reactive-streams" % "3.14.0"
)
Compile / run / fork := true
Configurer le client du SDK AWS
Créez un bucket et des informations d’accès S3 pour R2 dont la portée
est limitée à ce bucket, avec l’autorisation de lire et d’écrire des objets. Fournissez
R2_ACCOUNT_ID, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY et
R2_BUCKET via l’environnement de votre processus avant d’exécuter le programme.
Utilisez un gestionnaire de secrets ou la configuration de l’environnement de votre terminal ;
ne placez pas d’informations d’accès dans le fichier Scala ni dans les arguments de commande.
Le bucket reste privé et doit déjà exister.
Les clients utilisent la région auto, l’adressage de type chemin et un
encodage par blocs désactivé, conformément à la
configuration Java de Cloudflare. WHEN_REQUIRED
désactive les sommes de contrôle facultatives des requêtes du SDK.
Le programme sélectionne Apache pour les échanges HTTP synchrones et Netty pour les échanges HTTP
asynchrones, et ferme chaque client avec Resource de Cats Effect.
Effectuer un téléversement synchrone simple
Choisissez sync pour effectuer un PUT bloquant simple. Il lit le fichier
via RequestBody.fromFile sur le pool de threads de Cats Effect réservé aux opérations
bloquantes. L’annulation attend le retour de l’appel bloquant ou l’expiration de son délai ;
choisissez async ou fs2 si vous avez besoin
d’annuler le future sous-jacent.
Effectuer un téléversement asynchrone avec Cats Effect
Enregistrez ce programme complet sous src/main/scala/R2Upload.scala.
async utilise le corps de requête du SDK fondé sur un fichier ;
fs2 fournit un éditeur Reactive Streams. Les deux attendent via la
passerelle CompletableFuture de Cats Effect, qui demande
l’annulation du future lorsque sa fibre est annulée.
import cats.effect.{ExitCode, IO, IOApp, Resource}
import cats.effect.implicits._
import cats.syntax.all._
import fs2.interop.reactivestreams._
import fs2.io.file.{Files, Flags, Path}
import java.io.InputStream
import java.net.URI
import java.nio.ByteBuffer
import java.security.MessageDigest
import java.time.Duration
import java.util.Optional
import org.reactivestreams.Subscriber
import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
import software.amazon.awssdk.core.async.AsyncRequestBody
import software.amazon.awssdk.core.checksums.RequestChecksumCalculation
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration
import software.amazon.awssdk.core.sync.RequestBody
import software.amazon.awssdk.http.apache.ApacheHttpClient
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.retries.StandardRetryStrategy
import software.amazon.awssdk.services.s3.{S3AsyncClient, S3Client, S3Configuration}
import software.amazon.awssdk.services.s3.model.{GetObjectRequest, PutObjectRequest, PutObjectResponse, S3Exception}
object R2Upload extends IOApp {
final case class Config(endpoint: URI, accessKey: String, secretKey: String, bucket: String)
final case class Task(path: Path, key: String)
private val service = S3Configuration.builder()
.pathStyleAccessEnabled(true)
.chunkedEncodingEnabled(false)
.build()
private def overrides: ClientOverrideConfiguration =
ClientOverrideConfiguration.builder()
.apiCallTimeout(Duration.ofMinutes(2))
.apiCallAttemptTimeout(Duration.ofSeconds(45))
.retryStrategy(StandardRetryStrategy.builder().maxAttempts(3).build())
.build()
private def credentials(config: Config): StaticCredentialsProvider =
StaticCredentialsProvider.create(
AwsBasicCredentials.create(config.accessKey, config.secretKey)
)
def asyncClient(config: Config): Resource[IO, S3AsyncClient] =
Resource.fromAutoCloseable(IO.blocking {
S3AsyncClient.builder()
.httpClientBuilder(NettyNioAsyncHttpClient.builder()
.connectionTimeout(Duration.ofSeconds(10))
.readTimeout(Duration.ofSeconds(30))
.writeTimeout(Duration.ofSeconds(30)))
.endpointOverride(config.endpoint)
.credentialsProvider(credentials(config))
.region(Region.of("auto"))
.serviceConfiguration(service)
.requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED)
.overrideConfiguration(overrides)
.build()
})
def syncClient(config: Config): Resource[IO, S3Client] =
Resource.fromAutoCloseable(IO.blocking {
S3Client.builder()
.httpClientBuilder(ApacheHttpClient.builder()
.connectionTimeout(Duration.ofSeconds(10))
.socketTimeout(Duration.ofSeconds(30)))
.endpointOverride(config.endpoint)
.credentialsProvider(credentials(config))
.region(Region.of("auto"))
.serviceConfiguration(service)
.requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED)
.overrideConfiguration(overrides)
.build()
})
private def request(bucket: String, task: Task, size: Long): PutObjectRequest = {
require(size >= 0 && size <= 5L * 1024 * 1024 * 1024,
"Use multipart upload for files larger than 5 GiB")
PutObjectRequest.builder()
.bucket(bucket).key(task.key)
.contentType("application/octet-stream")
.contentLength(size)
.build()
}
def uploadSync(client: S3Client, bucket: String, task: Task): IO[PutObjectResponse] =
Files[IO].size(task.path).flatMap { size =>
IO.blocking(client.putObject(
request(bucket, task, size), RequestBody.fromFile(task.path.toNioPath)
))
}
def uploadAsync(client: S3AsyncClient, bucket: String, task: Task): IO[PutObjectResponse] =
Files[IO].size(task.path).flatMap { size =>
IO.fromCompletableFuture(IO {
client.putObject(request(bucket, task, size),
AsyncRequestBody.fromFile(task.path.toNioPath))
})
}
def uploadStream(client: S3AsyncClient, bucket: String, task: Task): IO[PutObjectResponse] =
Files[IO].size(task.path).flatMap { size =>
val buffers = Files[IO].readAll(task.path, 64 * 1024, Flags.Read)
.chunks.map(chunk => ByteBuffer.wrap(chunk.toArray))
buffers.toUnicastPublisher.use { publisher =>
val body = new AsyncRequestBody {
override def contentLength(): Optional[java.lang.Long] =
Optional.of(java.lang.Long.valueOf(size))
override def subscribe(subscriber: Subscriber[_ >: ByteBuffer]): Unit =
publisher.subscribe(subscriber)
}
IO.fromCompletableFuture(IO {
client.putObject(request(bucket, task, size), body)
})
}
}
def uploadMany(client: S3AsyncClient, bucket: String,
tasks: List[Task]): IO[List[PutObjectResponse]] =
tasks.parTraverseN(4)(task => uploadAsync(client, bucket, task))
private def digest(input: InputStream): Array[Byte] = {
val hash = MessageDigest.getInstance("SHA-256")
val buffer = new Array[Byte](64 * 1024)
var count = input.read(buffer)
while (count != -1) {
hash.update(buffer, 0, count)
count = input.read(buffer)
}
hash.digest()
}
def verify(client: S3Client, bucket: String, task: Task): IO[Unit] = {
val local = Resource.fromAutoCloseable(IO.blocking(
java.nio.file.Files.newInputStream(task.path.toNioPath)))
val remote = Resource.fromAutoCloseable(IO.blocking(client.getObject(
GetObjectRequest.builder().bucket(bucket).key(task.key).build())))
(local, remote).tupled.use { case (source, stored) =>
IO.blocking {
require(MessageDigest.isEqual(digest(source), digest(stored)), "Stored bytes differ")
}
}
}
override def run(args: List[String]): IO[ExitCode] = {
val operation = IO {
require(args.length >= 3 && args.length % 2 == 1,
"Usage: R2Upload sync|async|fs2|parallel|verify file key [file key ...]")
val mode = args.head
require(Set("sync", "async", "fs2", "parallel", "verify").contains(mode), "Unknown mode")
require(mode == "parallel" || args.length == 3, "Only parallel accepts multiple files")
val account = sys.env("R2_ACCOUNT_ID")
require(account.matches("[a-fA-F0-9]{32}"), "Invalid R2 account ID")
val config = Config(
URI.create(s"https://$account.r2.cloudflarestorage.com"),
sys.env("R2_ACCESS_KEY_ID"), sys.env("R2_SECRET_ACCESS_KEY"), sys.env("R2_BUCKET")
)
val tasks = args.tail.grouped(2).map(pair => Task(Path(pair.head), pair(1))).toList
require(tasks.forall(task =>
task.path.toString.matches("[A-Za-z0-9_./-]+") &&
task.key.matches("[A-Za-z0-9_./-]+") && !task.key.startsWith("-")),
"Use ASCII paths and keys containing only letters, digits, _, ., /, and -")
require(tasks.map(_.key).distinct.size == tasks.size, "Use distinct destination keys")
(mode, config, tasks)
}.flatMap { case (mode, config, tasks) =>
val upload = if (mode == "verify")
syncClient(config).use(client => verify(client, config.bucket, tasks.head))
else if (mode == "sync")
syncClient(config).use(client => uploadSync(client, config.bucket, tasks.head).void)
else
asyncClient(config).use { client =>
mode match {
case "fs2" => uploadStream(client, config.bucket, tasks.head).void
case "parallel" => uploadMany(client, config.bucket, tasks).void
case _ => uploadAsync(client, config.bucket, tasks.head).void
}
}
upload *> IO.println(if (mode == "verify") "Stored bytes match." else "Upload completed.")
}
operation.as(ExitCode.Success).handleErrorWith {
case error: S3Exception =>
IO.println(s"Upload failed (${error.getClass.getSimpleName}, HTTP ${error.statusCode()}).").as(ExitCode.Error)
case error =>
IO.println(s"Upload failed (${error.getClass.getSimpleName}). Check arguments, environment, file access, and network.")
.as(ExitCode.Error)
}
}
}
Ouvrez un terminal dans r2-scala. Ces commandes créent un petit fichier
d’entrée, compilent le programme et téléversent ce fichier. La première compilation télécharge les
dépendances. Les arguments de fichier et de clé de ce lanceur sont limités aux lettres ASCII, aux
chiffres, à _, à ., à
/ et à - ; utilisez
./ avant un nom local commençant par un trait d’union. Les clés de
destination ne doivent pas commencer par - et doivent être distinctes
au sein d’un même lot. Gardez les fichiers sources inchangés jusqu’à la fin de toutes les tentatives.
(
set -eu
printf 'Hello, R2!\n' > hello.txt
java -Dsbt.server.autostart=false -jar sbt-launch.jar compile
java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload async ./hello.txt uploads/hello.txt"
)
Remplacez async par sync ou
fs2 pour les autres modes à fichier unique. Un PUT réussi remplace tout
objet associé à la même clé, y compris lorsque le fichier d’entrée est vide. Utilisez une nouvelle
clé si vous devez conserver l’objet précédent.
Après la réussite, relisez l’objet privé et comparez les empreintes SHA-256 de l’intégralité du contenu local et du contenu stocké :
java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload verify ./hello.txt uploads/hello.txt"
Le vérificateur affiche Stored bytes match. uniquement lorsque les deux empreintes
concordent et renvoie un statut non nul en cas de différence ou d’échec de lecture. Un PUT terminé
confirme uniquement l’acceptation par le service ; exécutez la comparaison avant de supprimer un
original important. La vérification nécessite également que la source reste inchangée.
Gérer les réponses et les erreurs
Upload completed. n’apparaît qu’après la fin de tous les téléversements sélectionnés et
de la portée du client. Les erreurs du service renvoient un code de sortie non nul et un statut
HTTP. Les erreurs de configuration locale, de fichier et de réseau renvoient également un code de
sortie non nul, sans afficher d’informations d’accès ni de réponses brutes du SDK.
Un téléversement annulé n’est pas considéré comme terminé avec succès. Une annulation ou une réponse perdue peut toutefois laisser un objet enregistré à destination : inspectez-le avant de réessayer. Ce programme n’annule pas les écritures d’objets et ne supprime pas de données distantes en cas d’échec.
Gérer les nouvelles tentatives avec attente exponentielle
La stratégie standard de nouvelle tentative du SDK gère les échecs pour lesquels une nouvelle tentative est possible, avec au maximum trois tentatives par PUT. Le délai maximal d’un appel API est de deux minutes, nouvelles tentatives comprises ; celui d’une tentative est de 45 secondes. L’établissement de la connexion est limité à 10 secondes. Ces limites s’appliquent aux requêtes et ne constituent pas une échéance de deux minutes pour l’ensemble d’un lot parallèle. Les délais maximaux de lecture des transports, fixés à 30 secondes, sont inférieurs au délai maximal d’une tentative ; Netty impose également un délai maximal d’écriture de 30 secondes.
Un transfert lent peut dépasser le délai maximal même si le fichier est inférieur à la limite de taille. Vérifiez la destination, puis adaptez les limites à votre connexion avant de réessayer avec le fichier inchangé. N’ajoutez pas de boucle de nouvelle tentative sans limite. L’éditeur FS2 peut réexécuter son flux de lecture du fichier lors d’un nouvel abonnement ; il nécessite donc lui aussi une source stable et pouvant être relue.
Téléverser plusieurs fichiers en parallèle
Le mode parallel exécute au maximum quatre effets de téléversement à la fois.
Il annule les autres effets si l’un d’eux échoue et renvoie un statut d’échec. Les objets dont le
téléversement est terminé restent en place ; un lot n’est pas une transaction. Après avoir créé un
deuxième fichier local nommé report.pdf, exécutez :
java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload parallel ./hello.txt uploads/hello.txt ./report.pdf uploads/report.pdf"
La limite porte sur les téléversements actifs, pas sur la liste d’arguments en mémoire ni sur tous les tampons internes de Netty et du SDK. Elle ne divise pas un fichier en parties parallèles.
Transférer de gros fichiers en flux avec fs2
fs2 lit des blocs de 64 KiB et les expose sous forme d’éléments
ByteBuffer en fonction de la demande en aval. La ressource de l’éditeur reste
ouverte jusqu’à la fin du traitement, qu’il se termine par une réussite, un échec ou une annulation.
Cela évite de charger le fichier entier dans un tableau d’octets ; cela ne fixe pas de limite à la
mémoire totale utilisée par la JVM ou le client HTTP.
Tous les modes de téléversement utilisent des PUT uniques et refusent les fichiers de plus de 5 GiB avant l’envoi d’une requête. Le transfert en flux n’ajoute ni téléversement multipartie ni reprise. Pour les fichiers plus volumineux ou les parties dont le transfert peut reprendre, utilisez les téléversements multiparties R2.
Bonnes pratiques de sécurité
Utilisez des informations d’accès limitées au bucket et choisissez les clés de destination avec soin. Ce programme Scala exécuté côté serveur ne nécessite ni configuration CORS pour navigateur ni URL publique du bucket. Pour les fichiers fournis par les utilisateurs, terminez la validation avant de les exporter vers un emplacement dont votre application diffuse le contenu.
Erreurs courantes et solutions
| Résultat | Vérification suivante |
|---|---|
| Échec HTTP | Vérifiez le point de terminaison du compte, le nom du bucket et les autorisations du jeton sur les objets. Un statut seul ne permet pas d’identifier le paramètre incorrect. |
| Échec local | Vérifiez les quatre variables d’environnement, les restrictions sur les arguments, la lisibilité des fichiers et la limite de 5 GiB. |
| Délai dépassé ou échec de connexion | Vérifiez la connectivité et les limites des requêtes, puis inspectez la destination avant de réessayer. |
| Lot parallèle partiellement terminé | Inspectez chaque clé demandée ; les objets téléversés avec succès restent en place même si la commande globale a échoué. |
Pour conclure
Utilisez le corps de requête du SDK fondé sur un fichier pour les exports ordinaires, et FS2 lorsque votre pipeline de flux existant doit fournir les octets. Si l’export suit un Job de traitement Transloadit, le Robot 🤖 /cloudflare/store peut écrire les résultats dans R2 à la place de ce programme.