-
Notifications
You must be signed in to change notification settings - Fork 647
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
FTP Connector #3
Conversation
57d8658
to
76649d3
Compare
83f30f0
to
b803a05
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think this is looking good. Let us know if there is something specific you would like us to take an extra look at. Then adding javadsl
and Paradox.
import scala.util.Try | ||
|
||
/** | ||
* @author Juan José Vázquez Delgado |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
please remove, explained in https://github.com/akka/alpakka/blob/master/CONTRIBUTING.md#pull-request-requirements
def connect( | ||
connectionSettings: FtpConnectionSettings | ||
)(implicit ftpClient: FTPClient): Try[Handler] = Try { | ||
if (ftpClient != null) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why those null checks? isn't that a program error?
/* | ||
* Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com> | ||
*/ | ||
package akka.stream.alpakka.ftp |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we have refactored the package conventions after you started this. We now use scaladsl
and javadsl
. See the other connectors in master.
/** | ||
* @author Juan José Vázquez Delgado | ||
*/ | ||
trait SFtp { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
think about what classes should be public and what is internal. Keep the public API as small as possible. Mark internal things with private[alpakka]
, perhaps move to an impl
package also.
@patriknw Still working on this. I plan to cut a first version this week with at least the basic operations for listing and retrieving files as streams. After that, it could be a good idea going to merge and let others to contribute for improving it. FTP is an old pal and has a lot of nuances and different use cases. Thanks for your comments!. |
566b742
to
7c0ddb0
Compare
connectionSettings: RemoteFileSettings | ||
): Source[FtpFile, CompletionStage[java.lang.Long]] = { | ||
import scala.compat.java8.FutureConverters._ | ||
implicit val ec = scala.compat.java8.FutureConverters.globalExecutionContext // TODO |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@patriknw Any idea about how could I avoid this execution context import?. I need to map the Scala Future[Long]
to the Java CompletionStage<java.lang.Long>
and the longs conversion is asking me for an ExecutionContext
. Should I use a wrapper as in IOResult
?.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
GraphStage has materializer.executionContext
e948be9
to
cb22209
Compare
cb22209
to
e99d974
Compare
@@ -44,6 +44,15 @@ lazy val s3 = project | |||
Dependencies.S3 | |||
) | |||
|
|||
lazy val ftp = project | |||
.in(file("ftp")) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is not necessary, when name of the module is the same as the directory.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Just a couple of nitpicks in the docs. Otherwise LGTM.
sbt | ||
: @@@vars | ||
```scala | ||
libraryDependencies += "com.typesafe.akka" %% "akka-stream-alpakka-ftp" % "$version$" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
s/typesafe/lightbend
: @@@vars | ||
```xml | ||
<dependency> | ||
<groupId>com.typesafe.akka</groupId> |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
s/typesafe/lightbend
: @@@vars | ||
```gradle | ||
dependencies { | ||
compile group: "com.typesafe.akka", name: "akka-stream-alpakka-ftp_$scala.binaryVersion$", version: "$version$" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
s/typesafe/lightbend
|
||
### Configuring the connection settings | ||
|
||
In order to establish a connection with the remote server, you need to provide an specialized version of a @scaladoc[RemoteFileSettings](akka.stream.alpakka.ftp.RemoteFileSettings) instance. It's specialized as it depends on the kind of server you're connecting to: FTP, FTPs or SFTP. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
... you need to provide an specialized ...
Java | ||
: @@snip (../../../../ftp/src/test/java/akka/stream/alpakka/ftp/FtpSourceTest.java) { #create-settings } | ||
|
||
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: @scaladoc[FtpsSettings](akka.stream.alpakka.ftp.RemoteFileSettings.FtpsSettings) or @scaladoc[SFtpSettings](akka.stream.alpakka.ftp.RemoteFileSettings.SFtpSettings) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This does not properly link. Use akka.stream.alpakka.ftp.RemoteFileSettings$$FtpsSettings
and akka.stream.alpakka.ftp.RemoteFileSettings$$SFtpSettings
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: @scaladoc[FtpsSettings](akka.stream.alpakka.ftp.RemoteFileSettings.FtpsSettings) or @scaladoc[SFtpSettings](akka.stream.alpakka.ftp.RemoteFileSettings.SFtpSettings) | ||
respectively. | ||
|
||
For non-anonymous connection, please provide an instance of [NonAnonFtpCredentials](akka.stream.alpakka.ftp.FtpCredentials.NonAnonFtpCredentials) instead. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Missing @scaladoc
before [NonAnonFtpC...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looks promising, I had a few questions/opinions I have added.
import scala.concurrent.{ Future, Promise } | ||
|
||
private[ftp] trait FtpBrowserGraphStage[FtpClient] | ||
extends GraphStageWithMaterializedValue[SourceShape[FtpFile], Future[Done]] { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
if the stage just fails for errors, and completes the same thing can be achieved by using ftpSource.watchTermination()
, is there some special circumstance that the Future[Done]
materialized value is needed for?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Not really. The only purpose of Future[Done]
is to check termination. Should I change it and not materialize?.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Future[Done]
is great for sinks, but not really needed for sources
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Perfect. I'll change this.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done.
val ftpLike: FtpLike[FtpClient] | ||
|
||
val shape = SourceShape(Outlet[FtpFile](s"$name.out")) | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
as the blocking logic is done already in this trait I think the IODispatcher
should be specified using initialAttributes
here rather than when creating the flows in FtpApi
(and possibly miss out on it)
|
||
val logic = new FtpGraphStageLogic[FtpFile, FtpClient](shape, ftpLike, connectionSettings) { | ||
|
||
private var currentFiles: Stream[FtpFile] = Stream.empty[FtpFile] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I seem to recall that Stream
leaks memory because of memoization, are you sure this is not happening here (keeping each file in memory even after emitting it downstream)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please, take into account that the propagated element here is an FtpFile
that is a mere descriptor for the physical file. The physical file is got through other Source
, the FtpSourceIOGraphStage
in the form of stream of ByteString
. So, IMHO, the scala.collection.immutable.Stream
should not be a problem here, right?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If you list a directory with 50 000 files, it would (if I'm right about memoization) keep all of them in memory until the stream completes which isn't very nice.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Wow!. I'll check this right away. Thanks.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Finally changed the Stream
for a strict sequence as I'm quite sure that we had memoization as you mentioned. Thanks!.
protected val ftpIOSourceName: String = FtpsIOSourceName | ||
} | ||
|
||
private[ftp] trait SFtpSource { _: FtpSourceFactory[_] => |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
if FTP
is cased as Ftp
shouldn't SFTP
be Sftp
rather than SFtp
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It makes sense, but I find ugly Sftp
personally. I suppose this is a matter of taste. Other opinions?.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't care super much about which one of them, it was more about consistency, both should follow the same pattern, no?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Accepted. I changed the naming. Thanks!
1f664ca
to
29143af
Compare
|
||
override def initialAttributes: Attributes = | ||
super.initialAttributes and Attributes.name(name) and IODispatcher | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@johanandren Please, can you confirm that this is what you mean regarding initial attributes?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Exactly!
29143af
to
0895b04
Compare
0895b04
to
443f00d
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM
Thanks @juanjovazquez ! |
Please, check if this implementation might be enough for a first version of the connector. My team is already using it on a real project so this will be tested from the beginning.
closes #4
/cc @patriknw @ktoso @2m