You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2023/07/14 13:01:20 UTC
[incubator-pekko-connectors] 15/21: Scala3 ftp support (#170)
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch scala3
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
commit bc6092d04f0e86cb8b9152d53af7780ad0b19396
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Sun Jun 11 14:24:27 2023 +0100
Scala3 ftp support (#170)
* support scala3 in ftp connector
* Update BaseSpec.scala
---
.../stream/connectors/ftp/impl/FtpBrowserGraphStage.scala | 6 +++---
.../ftp/impl/FtpDirectoryOperationsGraphStage.scala | 2 +-
.../stream/connectors/ftp/impl/FtpGraphStageLogic.scala | 10 +++++-----
.../pekko/stream/connectors/ftp/impl/FtpIOGraphStage.scala | 14 +++++++-------
.../apache/pekko/stream/connectors/ftp/impl/FtpLike.scala | 13 ++++++++-----
.../pekko/stream/connectors/ftp/impl/FtpOperations.scala | 2 +-
.../pekko/stream/connectors/ftp/impl/FtpsOperations.scala | 2 +-
.../pekko/stream/connectors/ftp/impl/SftpOperations.scala | 2 +-
.../pekko/stream/connectors/ftp/javadsl/FtpApi.scala | 4 ++--
.../pekko/stream/connectors/ftp/scaladsl/FtpApi.scala | 4 ++--
.../org/apache/pekko/stream/connectors/ftp/BaseSpec.scala | 2 +-
project/Dependencies.scala | 1 -
12 files changed, 32 insertions(+), 30 deletions(-)
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpBrowserGraphStage.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpBrowserGraphStage.scala
index 769156909..589624aa4 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpBrowserGraphStage.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpBrowserGraphStage.scala
@@ -31,7 +31,7 @@ private[ftp] trait FtpBrowserGraphStage[FtpClient, S <: RemoteFileSettings]
def emitTraversedDirectories: Boolean = false
- def createLogic(inheritedAttributes: Attributes) = {
+ def createLogic(inheritedAttributes: Attributes): FtpGraphStageLogic[FtpFile, FtpClient, S] = {
val logic = new FtpGraphStageLogic[FtpFile, FtpClient, S](shape, ftpLike, connectionSettings, ftpClient) {
private[this] var buffer: Seq[FtpFile] = Seq.empty[FtpFile]
@@ -84,9 +84,9 @@ private[ftp] trait FtpBrowserGraphStage[FtpClient, S <: RemoteFileSettings]
private[this] def getFilesFromPath(basePath: String) =
if (basePath.isEmpty)
- ftpLike.listFiles(handler.get)
+ graphStageFtpLike.listFiles(handler.get)
else
- ftpLike.listFiles(basePath, handler.get)
+ graphStageFtpLike.listFiles(basePath, handler.get)
} // end of stage logic
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpDirectoryOperationsGraphStage.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpDirectoryOperationsGraphStage.scala
index c24b0d066..2e80bd58a 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpDirectoryOperationsGraphStage.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpDirectoryOperationsGraphStage.scala
@@ -31,7 +31,7 @@ private[ftp] trait FtpDirectoryOperationsGraphStage[FtpClient, S <: RemoteFileSe
out,
new OutHandler {
override def onPull(): Unit = {
- push(out, ftpLike.mkdir(basePath, directoryName, handler.get))
+ push(out, graphStageFtpLike.mkdir(basePath, directoryName, handler.get))
complete(out)
}
})
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpGraphStageLogic.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpGraphStageLogic.scala
index 0090c3504..d248dc0bf 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpGraphStageLogic.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpGraphStageLogic.scala
@@ -29,18 +29,18 @@ import scala.util.control.NonFatal
@InternalApi
private[ftp] abstract class FtpGraphStageLogic[T, FtpClient, S <: RemoteFileSettings](
val shape: Shape,
- val ftpLike: FtpLike[FtpClient, S],
+ val graphStageFtpLike: FtpLike[FtpClient, S],
val connectionSettings: S,
val ftpClient: () => FtpClient) extends GraphStageLogic(shape) {
- protected[this] implicit val client = ftpClient()
- protected[this] var handler: Option[ftpLike.Handler] = Option.empty[ftpLike.Handler]
+ protected[this] implicit val client: FtpClient = ftpClient()
+ protected[this] var handler: Option[graphStageFtpLike.Handler] = Option.empty[graphStageFtpLike.Handler]
protected[this] var failed = false
override def preStart(): Unit = {
super.preStart()
try {
- val tryConnect = ftpLike.connect(connectionSettings)
+ val tryConnect = graphStageFtpLike.connect(connectionSettings)
if (tryConnect.isSuccess) {
handler = tryConnect.toOption
} else
@@ -76,7 +76,7 @@ private[ftp] abstract class FtpGraphStageLogic[T, FtpClient, S <: RemoteFileSett
protected[this] def doPreStart(): Unit
protected[this] def disconnect(): Unit =
- handler.foreach(ftpLike.disconnect)
+ handler.foreach(graphStageFtpLike.disconnect)
protected[this] def matSuccess(): Boolean
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpIOGraphStage.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpIOGraphStage.scala
index 75414ade0..00cbc5e21 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpIOGraphStage.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpIOGraphStage.scala
@@ -106,7 +106,7 @@ private[ftp] trait FtpIOSourceStage[FtpClient, S <: RemoteFileSettings]
isOpt.foreach { os =>
try {
os.close()
- ftpLike match {
+ graphStageFtpLike match {
case cfo: CommonFtpOperations =>
if (!cfo.completePendingCommand(handler.get.asInstanceOf[cfo.Handler]))
throw new IOException("File transfer failed.")
@@ -128,13 +128,13 @@ private[ftp] trait FtpIOSourceStage[FtpClient, S <: RemoteFileSettings]
}
protected[this] def doPreStart(): Unit =
- isOpt = ftpLike match {
+ isOpt = graphStageFtpLike match {
case ur: UnconfirmedReads =>
withUnconfirmedReads(ur)
case ro: RetrieveOffset =>
Some(ro.retrieveFileInputStream(path, handler.get.asInstanceOf[ro.Handler], offset).get)
case _ =>
- Some(ftpLike.retrieveFileInputStream(path, handler.get).get)
+ Some(graphStageFtpLike.retrieveFileInputStream(path, handler.get).get)
}
private def withUnconfirmedReads(
@@ -229,7 +229,7 @@ private[ftp] trait FtpIOSinkStage[FtpClient, S <: RemoteFileSettings]
osOpt.foreach { os =>
try {
os.close()
- ftpLike match {
+ graphStageFtpLike match {
case cfo: CommonFtpOperations =>
if (!cfo.completePendingCommand(handler.get.asInstanceOf[cfo.Handler]))
throw new IOException("File transfer failed.")
@@ -251,7 +251,7 @@ private[ftp] trait FtpIOSinkStage[FtpClient, S <: RemoteFileSettings]
}
protected[this] def doPreStart(): Unit = {
- osOpt = Some(ftpLike.storeFileOutputStream(path, handler.get, append).get)
+ osOpt = Some(graphStageFtpLike.storeFileOutputStream(path, handler.get, append).get)
pull(in)
}
@@ -301,7 +301,7 @@ private[ftp] trait FtpMoveSink[FtpClient, S <: RemoteFileSettings]
override def onPush(): Unit = {
try {
val sourcePath = grab(in)
- ftpLike.move(sourcePath.path, destinationPath(sourcePath), handler.get)
+ graphStageFtpLike.move(sourcePath.path, destinationPath(sourcePath), handler.get)
numberOfMovedFiles = numberOfMovedFiles + 1
pull(in)
} catch {
@@ -356,7 +356,7 @@ private[ftp] trait FtpRemoveSink[FtpClient, S <: RemoteFileSettings]
new InHandler {
override def onPush(): Unit = {
try {
- ftpLike.remove(grab(in).path, handler.get)
+ graphStageFtpLike.remove(grab(in).path, handler.get)
numberOfRemovedFiles = numberOfRemovedFiles + 1
pull(in)
} catch {
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpLike.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpLike.scala
index d64b488fe..216d7a597 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpLike.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpLike.scala
@@ -54,7 +54,7 @@ protected[ftp] trait FtpLike[FtpClient, S <: RemoteFileSettings] {
* INTERNAL API
*/
@InternalApi
-protected[ftp] trait RetrieveOffset { _: FtpLike[_, _] =>
+protected[ftp] trait RetrieveOffset { self: FtpLike[_, _] =>
def retrieveFileInputStream(name: String, handler: Handler, offset: Long): Try[InputStream]
@@ -64,7 +64,7 @@ protected[ftp] trait RetrieveOffset { _: FtpLike[_, _] =>
* INTERNAL API
*/
@InternalApi
-protected[ftp] trait UnconfirmedReads { _: FtpLike[_, _] =>
+protected[ftp] trait UnconfirmedReads { self: FtpLike[_, _] =>
def retrieveFileInputStream(name: String, handler: Handler, offset: Long, maxUnconfirmedReads: Int): Try[InputStream]
@@ -76,8 +76,11 @@ protected[ftp] trait UnconfirmedReads { _: FtpLike[_, _] =>
@InternalApi
object FtpLike {
// type class instances
- implicit val ftpLikeInstance = new FtpLike[FTPClient, FtpSettings] with RetrieveOffset with FtpOperations
- implicit val ftpsLikeInstance = new FtpLike[FTPSClient, FtpsSettings] with RetrieveOffset with FtpsOperations
- implicit val sFtpLikeInstance =
+ implicit val ftpLikeInstance: FtpLike[FTPClient, FtpSettings] with RetrieveOffset with FtpOperations =
+ new FtpLike[FTPClient, FtpSettings] with RetrieveOffset with FtpOperations
+ implicit val ftpsLikeInstance: FtpLike[FTPSClient, FtpsSettings] with RetrieveOffset with FtpsOperations =
+ new FtpLike[FTPSClient, FtpsSettings] with RetrieveOffset with FtpsOperations
+ implicit val sFtpLikeInstance
+ : FtpLike[SSHClient, SftpSettings] with RetrieveOffset with SftpOperations with UnconfirmedReads =
new FtpLike[SSHClient, SftpSettings] with RetrieveOffset with SftpOperations with UnconfirmedReads
}
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpOperations.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpOperations.scala
index 36ac044fb..6b7cc7b6a 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpOperations.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpOperations.scala
@@ -23,7 +23,7 @@ import scala.util.Try
* INTERNAL API
*/
@InternalApi
-private[ftp] trait FtpOperations extends CommonFtpOperations { _: FtpLike[FTPClient, FtpSettings] =>
+private[ftp] trait FtpOperations extends CommonFtpOperations { self: FtpLike[FTPClient, FtpSettings] =>
def connect(connectionSettings: FtpSettings)(implicit ftpClient: FTPClient): Try[Handler] = Try {
connectionSettings.proxy.foreach(ftpClient.setProxy)
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpsOperations.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpsOperations.scala
index b1bd6ffc1..3ffe0ce2e 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpsOperations.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/FtpsOperations.scala
@@ -25,7 +25,7 @@ import scala.util.Try
*/
@InternalApi
private[ftp] trait FtpsOperations extends CommonFtpOperations {
- _: FtpLike[FTPSClient, FtpsSettings] =>
+ self: FtpLike[FTPSClient, FtpsSettings] =>
def connect(connectionSettings: FtpsSettings)(implicit ftpClient: FTPSClient): Try[Handler] =
Try {
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
index 59045f385..04cbd77bd 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/impl/SftpOperations.scala
@@ -37,7 +37,7 @@ import scala.util.{ Failure, Try }
* INTERNAL API
*/
@InternalApi
-private[ftp] trait SftpOperations { _: FtpLike[SSHClient, SftpSettings] =>
+private[ftp] trait SftpOperations { self: FtpLike[SSHClient, SftpSettings] =>
type Handler = SFTPClient
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/javadsl/FtpApi.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/javadsl/FtpApi.scala
index 9d6c299b1..3644a1f4d 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/javadsl/FtpApi.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/javadsl/FtpApi.scala
@@ -30,7 +30,7 @@ import net.schmizz.sshj.SSHClient
import org.apache.commons.net.ftp.{ FTPClient, FTPSClient }
@DoNotInherit
-sealed trait FtpApi[FtpClient, S <: RemoteFileSettings] { _: FtpSourceFactory[FtpClient, S] =>
+sealed trait FtpApi[FtpClient, S <: RemoteFileSettings] { self: FtpSourceFactory[FtpClient, S] =>
/**
* Java API: creates a [[pekko.stream.javadsl.Source Source]] of [[FtpFile]]s from the remote user `root` directory.
@@ -573,6 +573,6 @@ object Sftp extends SftpApi {
*/
def create(customSshClient: SSHClient): SftpApi =
new SftpApi {
- override val sshClient: SSHClient = customSshClient
+ override def sshClient(): SSHClient = customSshClient
}
}
diff --git a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/scaladsl/FtpApi.scala b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/scaladsl/FtpApi.scala
index 4bc7da6e4..0e574bffe 100644
--- a/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/scaladsl/FtpApi.scala
+++ b/ftp/src/main/scala/org/apache/pekko/stream/connectors/ftp/scaladsl/FtpApi.scala
@@ -28,7 +28,7 @@ import org.apache.commons.net.ftp.{ FTPClient, FTPSClient }
import scala.concurrent.Future
@DoNotInherit
-sealed trait FtpApi[FtpClient, S <: RemoteFileSettings] { _: FtpSourceFactory[FtpClient, S] =>
+sealed trait FtpApi[FtpClient, S <: RemoteFileSettings] { self: FtpSourceFactory[FtpClient, S] =>
/**
* Scala API: creates a [[pekko.stream.scaladsl.Source Source]] of [[FtpFile]]s from the remote user `root` directory.
@@ -398,6 +398,6 @@ object Sftp extends SftpApi {
*/
def apply(customSshClient: SSHClient): SftpApi =
new SftpApi {
- override val sshClient: SSHClient = customSshClient
+ override def sshClient(): SSHClient = customSshClient
}
}
diff --git a/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/BaseSpec.scala b/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/BaseSpec.scala
index 7dad2da60..d8765f891 100644
--- a/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/BaseSpec.scala
+++ b/ftp/src/test/scala/org/apache/pekko/stream/connectors/ftp/BaseSpec.scala
@@ -77,7 +77,7 @@ trait BaseSpec
// Allows to run tests n times in a row with a command line argument, useful for debugging sporadic failures
// e.g. ftp/testOnly *.FtpsStageSpec -- -Dtimes=20
// https://gist.github.com/dwickern/6ba9c5c505d2325d3737ace059302922
- override abstract protected def runTest(testName: String, args: Args): Status = {
+ override protected def runTest(testName: String, args: Args): Status = {
def run0(times: Int): Status = {
val status = super.runTest(testName, args)
if (times <= 1) status else status.thenRun(run0(times - 1))
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 82f5160c8..dd4282336 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -179,7 +179,6 @@ object Dependencies {
))
val Ftp = Seq(
- crossScalaVersions -= Scala3,
libraryDependencies ++= Seq(
"commons-net" % "commons-net" % "3.8.0", // ApacheV2
"com.hierynomus" % "sshj" % "0.33.0" // ApacheV2
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org