You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by di...@apache.org on 2019/08/22 16:42:06 UTC
[airavata-mft] branch master updated (ce2e3ad -> c2f432e)
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git.
from ce2e3ad Adding S3 stream sender
new c7b1eb4 Adding NIO based file transferring initial impl
new eda7872 Remove senstive data
new c16da98 Remove Channel from interface Connector
new 10fc560 remove sensitive data
new 7a951d2 Improve code
new 5c0a238 Adding local transport
new df53974 Change permission
new 3b15709 Add intial implementation of TCP File Server
new 4d773e4 TCP transport implementation
new 1d373eb TCP transport implementation
new 1635fe5 TCP transport implementation
new e331ff7 TCP transport implementation
new c2f432e Merge pull request #1 from isururanawaka/nioImpl
The 22 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
LICENSE | 0
README | 0
core/pom.xml | 12 +++
.../airavata/mft/core/api/CompletionCallback.java | 41 ++++++++
.../apache/airavata/mft/core/api/Connector.java | 43 +++++++++
.../airavata/mft/core/api/ConnectorChannel.java | 64 +++++++++++++
.../org/apache/airavata/mft/core/api/Mediator.java | 39 ++++++++
.../airavata/mft/core/api/SinkConnector.java | 37 +++++++
.../airavata/mft/core/api/SourceConnector.java | 31 ++++++
.../airavata/mft/core/api/StreamedReceiver.java | 0
.../airavata/mft/core/api/StreamedSender.java | 0
.../airavata/mft/core/api/TransportOperator.java | 0
.../mft/core/bufferedImpl/ConnectorException.java | 37 +++++++
.../airavata/mft/core/bufferedImpl/Constants.java | 29 ++++++
.../core/bufferedImpl/channel/AbstractChannel.java | 54 +++++++++++
.../bufferedImpl/channel/AbstractConnector.java | 50 ++++++++++
.../core/bufferedImpl/channel/ChannelUtils.java | 92 ++++++++++++++++++
.../mft/core/bufferedImpl/channel/InChannel.java | 69 ++++++++++++++
.../mft/core/bufferedImpl/channel/OutChannel.java | 67 +++++++++++++
.../mediation/PassthroughMediator.java | 84 ++++++++++++++++
.../core/streaming/DoubleByteArrayInputStream.java | 0
.../streaming/DoubleByteArrayOutputStream.java | 0
.../mft/core/streaming/TransportMetadata.java | 0
.../mft/core/streaming/TransportStream.java | 0
pom.xml | 0
transport/local-transport/pom.xml | 41 ++++++++
.../mft/transport/local/FileConstants.java | 29 ++++++
.../airavata/mft/transport/local/FileUtils.java | 52 ++++++++++
.../mft/transport/local/ResourceIdentifier.java | 37 +++++++
.../mft/transport/local/SinkFileConnector.java | 60 ++++++++++++
.../mft/transport/local/SourceFileConnector.java | 50 ++++++++++
transport/pom.xml | 2 +
transport/s3-transport/pom.xml | 12 +++
.../org/apache/airavata/mft/transport/s3/Main.java | 86 +++++++++++++++++
.../airavata/mft/transport/s3/S3Constants.java | 30 ++++++
.../airavata/mft/transport/s3/S3Receiver.java | 0
.../mft/transport/s3/S3ResourceIdentifier.java | 0
.../apache/airavata/mft/transport/s3/S3Sender.java | 0
.../airavata/mft/transport/s3/S3SinkConnector.java | 87 +++++++++++++++++
.../mft/transport/s3/S3SourceConnector.java | 61 ++++++++++++
.../mft/transport/s3/S3TransportOperator.java | 0
.../airavata/mft/transport/s3/S3TransportUtil.java | 0
transport/scp-transport/pom.xml | 0
.../apache/airavata/mft/transport/scp/Main.java | 0
.../airavata/mft/transport/scp/SCPReceiver.java | 0
.../airavata/mft/transport/scp/SCPSender.java | 0
.../mft/transport/scp/SCPTransportOperator.java | 0
.../mft/transport/scp/SCPTransportUtil.java | 0
.../mft/transport/scp/SSHResourceIdentifier.java | 0
transport/tcp-transport/pom.xml | 41 ++++++++
.../airavata/mft/transport/tcp/Constants.java | 30 ++++++
.../transport/tcp/RemoteResourceIdentifier.java | 62 ++++++++++++
.../tcp/client/RemoteFileServerConnector.java | 54 +++++++++++
.../mft/transport/tcp/server/FileServer.java | 106 +++++++++++++++++++++
.../transport/tcp/server/FileServerConnector.java | 51 ++++++++++
55 files changed, 1640 insertions(+)
mode change 100644 => 100755 LICENSE
mode change 100644 => 100755 README
mode change 100644 => 100755 core/pom.xml
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java
mode change 100644 => 100755 core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
mode change 100644 => 100755 core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
mode change 100644 => 100755 core/src/main/java/org/apache/airavata/mft/core/api/TransportOperator.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/ChannelUtils.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
create mode 100755 core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
mode change 100644 => 100755 core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
mode change 100644 => 100755 core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
mode change 100644 => 100755 core/src/main/java/org/apache/airavata/mft/core/streaming/TransportMetadata.java
mode change 100644 => 100755 core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
mode change 100644 => 100755 pom.xml
create mode 100755 transport/local-transport/pom.xml
create mode 100755 transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
create mode 100755 transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileUtils.java
create mode 100755 transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/ResourceIdentifier.java
create mode 100755 transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SinkFileConnector.java
create mode 100755 transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SourceFileConnector.java
mode change 100644 => 100755 transport/pom.xml
mode change 100644 => 100755 transport/s3-transport/pom.xml
create mode 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
create mode 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
mode change 100644 => 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java
mode change 100644 => 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3ResourceIdentifier.java
mode change 100644 => 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
create mode 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
create mode 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
mode change 100644 => 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportOperator.java
mode change 100644 => 100755 transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java
mode change 100644 => 100755 transport/scp-transport/pom.xml
mode change 100644 => 100755 transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
mode change 100644 => 100755 transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
mode change 100644 => 100755 transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
mode change 100644 => 100755 transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java
mode change 100644 => 100755 transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
mode change 100644 => 100755 transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java
create mode 100644 transport/tcp-transport/pom.xml
create mode 100644 transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
create mode 100644 transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
create mode 100644 transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
create mode 100644 transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
create mode 100644 transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
[airavata-mft] 12/22: Remove Channel from interface Connector
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit c16da986af231c23cb008b06f11c4cfcd35a5808
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Fri Aug 16 09:30:55 2019 +0530
Remove Channel from interface Connector
---
.../apache/airavata/mft/core/api/Connector.java | 3 +-
.../airavata/mft/core/api/SinkConnector.java | 3 +-
.../mft/core/bufferedImpl/AbstractConnector.java | 46 ++--------------------
.../mft/core/bufferedImpl/PassthroughMediator.java | 6 +--
.../org/apache/airavata/mft/transport/s3/Main.java | 8 ++--
.../airavata/mft/transport/s3/S3SinkConnector.java | 6 +--
.../mft/transport/s3/S3SourceConnector.java | 1 -
7 files changed, 15 insertions(+), 58 deletions(-)
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
index 69be278..8627403 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
@@ -21,7 +21,6 @@ package org.apache.airavata.mft.core.api;
import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
-import java.nio.channels.Channel;
import java.util.Properties;
/**
@@ -47,5 +46,5 @@ public interface Connector {
* @param channel
* @throws Exception
*/
- void closeChannel(Channel channel) throws Exception;
+ void closeChannel(ConnectorChannel channel) throws Exception;
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
index 4d2eb65..dd02e10 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
@@ -19,7 +19,6 @@
package org.apache.airavata.mft.core.api;
-import java.nio.channels.Channel;
import java.util.Properties;
/**
@@ -39,5 +38,5 @@ public interface SinkConnector extends Connector{
* @param channel
* @return true if succes else false
*/
- boolean verifyUpload(Channel channel);
+ boolean verifyUpload(ConnectorChannel channel);
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
index 20e5989..8849aac 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
@@ -20,42 +20,24 @@
package org.apache.airavata.mft.core.bufferedImpl;
import org.apache.airavata.mft.core.api.Connector;
+import org.apache.airavata.mft.core.api.ConnectorChannel;
import org.apache.airavata.mft.core.api.SinkConnector;
import org.apache.airavata.mft.core.api.SourceConnector;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.channels.Channel;
-import java.util.concurrent.ConcurrentHashMap;
+
/**
* Common class for {@link SinkConnector} and {@link SourceConnector} implementations
* This overrides closeChannel method and release all resources allocated to releasing channel
*/
public abstract class AbstractConnector implements Connector {
- ConcurrentHashMap<Channel, Object> channelMap = new ConcurrentHashMap<>();
@Override
- public void closeChannel(Channel channel) throws ConnectorException {
- OutChannel outChannel;
- InChannel inChannel;
+ public void closeChannel(ConnectorChannel channel) throws ConnectorException {
try {
- if (channelMap.get(channel) instanceof OutputStream) {
- outChannel = (OutChannel) channelMap.get(channel);
- if (outChannel != null) {
- outChannel.closeChannel();
- channelMap.remove(channel);
- }
-
- } else if (channelMap.get(channel) instanceof InputStream) {
- inChannel = (InChannel) channelMap.get(channel);
- if (inChannel != null) {
- inChannel.closeChannel();
- channelMap.remove(channel);
- }
- }
+ channel.closeChannel();
} catch (IOException e) {
throw new ConnectorException("Error occurred while closing stream", e);
@@ -63,24 +45,4 @@ public abstract class AbstractConnector implements Connector {
}
- /**
- * Cache the operating channel
- * @param channel
- * @param obj
- */
- public void cacheChannel(Channel channel, Object obj) {
- channelMap.put(channel, obj);
- }
-
-
- /**
- * get the channel
- * @param channel
- * @return
- */
- public Object getConnectorChannel(Channel channel) {
- return channelMap.get(channel);
- }
-
-
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
index f25348c..af9b663 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
@@ -51,7 +51,7 @@ public class PassthroughMediator implements Mediator {
Object obj = dst.getChannelAttribute(Constants.CONNECTOR);
if (obj != null && obj instanceof SinkConnector) {
SinkConnector connector = (SinkConnector) obj;
- boolean success = connector.verifyUpload(dChannel);
+ boolean success = connector.verifyUpload(dst);
if (success) {
callback.onComplete("Successfully uploaded", null);
} else {
@@ -69,8 +69,8 @@ public class PassthroughMediator implements Mediator {
Connector sourceConnector = (Connector) src.getChannelAttribute(Constants.CONNECTOR);
Connector sinkConnector = (Connector) dst.getChannelAttribute(Constants.CONNECTOR);
try {
- sourceConnector.closeChannel(rChannel);
- sinkConnector.closeChannel(dChannel);
+ sourceConnector.closeChannel(src);
+ sinkConnector.closeChannel(dst);
} catch (Exception ex) {
String msg = "Error occurred while closing channels";
ConnectorException connectorException = new ConnectorException(msg, ex);
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
index 8f3b959..b2f647c 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
@@ -42,17 +42,17 @@ public class Main {
public void execute() {
Properties connectorConfProp = new Properties();
- connectorConfProp.put(S3Constants.ACCESS_KEY, "XXXX");
- connectorConfProp.put(S3Constants.SECRET_KEY, "YYY");
+ connectorConfProp.put(S3Constants.ACCESS_KEY, "AKIA2SKSIFUH7QDAROSR");
+ connectorConfProp.put(S3Constants.SECRET_KEY, "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");
connectorConfProp.put(S3Constants.REGION, "us-east-2");
Properties srcProp = new Properties();
- srcProp.put(S3Constants.BUCKET, "test");
+ srcProp.put(S3Constants.BUCKET, "blimpit-test");
srcProp.put(S3Constants.REMOTE_FILE, "test.pdf");
Properties dstProp = new Properties();
dstProp.put(S3Constants.BUCKET, "blimpit-test");
- dstProp.put(S3Constants.REMOTE_FILE, "meemure.pdf");
+ dstProp.put(S3Constants.REMOTE_FILE, "teemure.pdf");
try {
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
index c2cb503..3ee2bae 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
@@ -33,7 +33,6 @@ import java.io.IOException;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
-import java.nio.channels.Channel;
import java.util.Properties;
/**
@@ -70,13 +69,12 @@ public class S3SinkConnector extends AbstractConnector implements SinkConnector
OutChannel outChannel = new OutChannel(outputStream);
outChannel.addChannelAttribute(S3Constants.HTTP_CONNECTION, connection);
outChannel.addChannelAttribute(Constants.CONNECTOR, this);
- cacheChannel(outChannel.getChannel(), outChannel);
return outChannel;
}
@Override
- public boolean verifyUpload(Channel channel) {
- OutChannel outChannel = (OutChannel) getConnectorChannel(channel);
+ public boolean verifyUpload(ConnectorChannel channel) {
+ OutChannel outChannel = (OutChannel) channel;
HttpURLConnection connection = (HttpURLConnection) outChannel.getChannelAttribute(S3Constants.HTTP_CONNECTION);
try {
if (connection.getResponseCode() == S3Constants.HTTP_SUCCESS_RESPONSE_CODE) {
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
index ce66c0c..6539d04 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
@@ -55,7 +55,6 @@ public class S3SourceConnector extends AbstractConnector implements SourceConnec
inputStream = s3object.getObjectContent();
InChannel inChannel = new InChannel(inputStream);
inChannel.addChannelAttribute(Constants.CONNECTOR, this);
- cacheChannel(inChannel.getChannel(), inChannel);
return inChannel;
}
return null;
[airavata-mft] 07/22: License headers
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 1f02b024c6eb5d13adab24f7c2c0643437dfbb0a
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Tue Aug 6 13:26:11 2019 -0400
License headers
---
.../mft/transport/scp/SCPTransportOperator.java | 19 +++++++++++++++++++
.../airavata/mft/transport/scp/SCPTransportUtil.java | 19 +++++++++++++++++++
2 files changed, 38 insertions(+)
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java
index 8a06a9f..eccef2d 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.transport.scp;
import com.jcraft.jsch.Channel;
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
index fe70ec1..8f9e193 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.transport.scp;
import com.jcraft.jsch.JSch;
[airavata-mft] 13/22: remove sensitive data
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 10fc560ac3381aac6d4016f751184bd6d09d1b36
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Fri Aug 16 09:33:42 2019 +0530
remove sensitive data
---
.../src/main/java/org/apache/airavata/mft/transport/s3/Main.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
index b2f647c..af29276 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
@@ -42,16 +42,16 @@ public class Main {
public void execute() {
Properties connectorConfProp = new Properties();
- connectorConfProp.put(S3Constants.ACCESS_KEY, "AKIA2SKSIFUH7QDAROSR");
- connectorConfProp.put(S3Constants.SECRET_KEY, "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");
+ connectorConfProp.put(S3Constants.ACCESS_KEY, "XXXX");
+ connectorConfProp.put(S3Constants.SECRET_KEY, "YYYY");
connectorConfProp.put(S3Constants.REGION, "us-east-2");
Properties srcProp = new Properties();
- srcProp.put(S3Constants.BUCKET, "blimpit-test");
+ srcProp.put(S3Constants.BUCKET, "test");
srcProp.put(S3Constants.REMOTE_FILE, "test.pdf");
Properties dstProp = new Properties();
- dstProp.put(S3Constants.BUCKET, "blimpit-test");
+ dstProp.put(S3Constants.BUCKET, "test");
dstProp.put(S3Constants.REMOTE_FILE, "teemure.pdf");
try {
[airavata-mft] 02/22: Adding license file
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 8de619253d384a265862359182739d9c5c22b6db
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Sun Aug 4 21:15:49 2019 -0400
Adding license file
---
LICENSE | 201 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 201 insertions(+)
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..f49a4e1
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,201 @@
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
\ No newline at end of file
[airavata-mft] 06/22: Adding class and method level comments
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 65e333188fbedd74e72c05f0e56203c864a5e4a9
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Tue Aug 6 13:24:44 2019 -0400
Adding class and method level comments
---
.../mft/core/streaming/DoubleByteArrayInputStream.java | 8 ++++++++
.../mft/core/streaming/DoubleByteArrayOutputStream.java | 16 ++++++++++++++++
2 files changed, 24 insertions(+)
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
index 6228d93..9845e84 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
@@ -22,6 +22,10 @@ package org.apache.airavata.mft.core.streaming;
import java.io.IOException;
import java.io.InputStream;
+/**
+ * This accepts {@link DoubleByteArrayOutputStream} and monitors for the availability of readable data. If there is any,
+ * this is exposing them as an {@link InputStream}
+ */
public class DoubleByteArrayInputStream extends InputStream {
private DoubleByteArrayOutputStream outputStream;
@@ -44,6 +48,10 @@ public class DoubleByteArrayInputStream extends InputStream {
return this.currentInputStream.available();
}
+ /**
+ * This will check if there is new data in the {@link DoubleByteArrayOutputStream}
+ * @throws IOException
+ */
private void refresh() throws IOException {
if (this.currentInputStream.available() == 0) {
InputStream tempInputStream = this.outputStream.asInputStream();
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
index 1f1b56b..56728c3 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
@@ -21,6 +21,13 @@ package org.apache.airavata.mft.core.streaming;
import java.io.*;
+/**
+ * This class wraps two {@link ByteArrayOutputStream} objects and implements the API of {@link OutputStream}. Main
+ * objective of this implementation is to use one stream for writing data and other stream to read the already written data.
+ * Behaviour of this Class is similar to what {@link PipedOutputStream} is doing but with more relaxed modes in thread
+ * contention and high performance. Limitation is that an instance of this class only be used between two threads. One
+ * thread can write to the output stream and other thread can digest data from asInputStream method.
+ */
public class DoubleByteArrayOutputStream extends OutputStream {
private ByteArrayOutputStream stream1 = new ByteArrayOutputStream();
@@ -49,6 +56,10 @@ public class DoubleByteArrayOutputStream extends OutputStream {
processedBytes += len;
}
+ /**
+ * This will swap currently writing {@link BufferedOutputStream} into the non active one. In order to perform
+ * this swap, non active stream should be empty.
+ */
private void swapBuffers() {
while (!clearedNonActiveStream) {
try {
@@ -90,6 +101,11 @@ public class DoubleByteArrayOutputStream extends OutputStream {
isClosed = true;
}
+ /**
+ * This will read currently non updating output stream and convert its data as an {@link InputStream}. Multiple invoking
+ * this will return null if there is no new data.
+ * @return {@link InputStream} of currently un-read data. null otherwise
+ */
public InputStream asInputStream() {
if (clearedNonActiveStream) {
return null;
[airavata-mft] 22/22: Merge pull request #1 from
isururanawaka/nioImpl
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit c2f432e9d299fb95c1715d5666d34968deb69b0e
Merge: ce2e3ad e331ff7
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Thu Aug 22 12:42:01 2019 -0400
Merge pull request #1 from isururanawaka/nioImpl
NIO based file transport implementation
LICENSE | 0
README | 0
core/pom.xml | 12 +++
.../airavata/mft/core/api/CompletionCallback.java | 41 ++++++++
.../apache/airavata/mft/core/api/Connector.java | 43 +++++++++
.../airavata/mft/core/api/ConnectorChannel.java | 64 +++++++++++++
.../org/apache/airavata/mft/core/api/Mediator.java | 39 ++++++++
.../airavata/mft/core/api/SinkConnector.java | 37 +++++++
.../airavata/mft/core/api/SourceConnector.java | 31 ++++++
.../airavata/mft/core/api/StreamedReceiver.java | 0
.../airavata/mft/core/api/StreamedSender.java | 0
.../airavata/mft/core/api/TransportOperator.java | 0
.../mft/core/bufferedImpl/ConnectorException.java | 37 +++++++
.../airavata/mft/core/bufferedImpl/Constants.java | 29 ++++++
.../core/bufferedImpl/channel/AbstractChannel.java | 54 +++++++++++
.../bufferedImpl/channel/AbstractConnector.java | 50 ++++++++++
.../core/bufferedImpl/channel/ChannelUtils.java | 92 ++++++++++++++++++
.../mft/core/bufferedImpl/channel/InChannel.java | 69 ++++++++++++++
.../mft/core/bufferedImpl/channel/OutChannel.java | 67 +++++++++++++
.../mediation/PassthroughMediator.java | 84 ++++++++++++++++
.../core/streaming/DoubleByteArrayInputStream.java | 0
.../streaming/DoubleByteArrayOutputStream.java | 0
.../mft/core/streaming/TransportMetadata.java | 0
.../mft/core/streaming/TransportStream.java | 0
pom.xml | 0
transport/local-transport/pom.xml | 41 ++++++++
.../mft/transport/local/FileConstants.java | 29 ++++++
.../airavata/mft/transport/local/FileUtils.java | 52 ++++++++++
.../mft/transport/local/ResourceIdentifier.java | 37 +++++++
.../mft/transport/local/SinkFileConnector.java | 60 ++++++++++++
.../mft/transport/local/SourceFileConnector.java | 50 ++++++++++
transport/pom.xml | 2 +
transport/s3-transport/pom.xml | 12 +++
.../org/apache/airavata/mft/transport/s3/Main.java | 86 +++++++++++++++++
.../airavata/mft/transport/s3/S3Constants.java | 30 ++++++
.../airavata/mft/transport/s3/S3Receiver.java | 0
.../mft/transport/s3/S3ResourceIdentifier.java | 0
.../apache/airavata/mft/transport/s3/S3Sender.java | 0
.../airavata/mft/transport/s3/S3SinkConnector.java | 87 +++++++++++++++++
.../mft/transport/s3/S3SourceConnector.java | 61 ++++++++++++
.../mft/transport/s3/S3TransportOperator.java | 0
.../airavata/mft/transport/s3/S3TransportUtil.java | 0
transport/scp-transport/pom.xml | 0
.../apache/airavata/mft/transport/scp/Main.java | 0
.../airavata/mft/transport/scp/SCPReceiver.java | 0
.../airavata/mft/transport/scp/SCPSender.java | 0
.../mft/transport/scp/SCPTransportOperator.java | 0
.../mft/transport/scp/SCPTransportUtil.java | 0
.../mft/transport/scp/SSHResourceIdentifier.java | 0
transport/tcp-transport/pom.xml | 41 ++++++++
.../airavata/mft/transport/tcp/Constants.java | 30 ++++++
.../transport/tcp/RemoteResourceIdentifier.java | 62 ++++++++++++
.../tcp/client/RemoteFileServerConnector.java | 54 +++++++++++
.../mft/transport/tcp/server/FileServer.java | 106 +++++++++++++++++++++
.../transport/tcp/server/FileServerConnector.java | 51 ++++++++++
55 files changed, 1640 insertions(+)
[airavata-mft] 05/22: Adding SCP sender and framework improvements
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit cc8d12ae31327f8531e0806ce3c07cdcb4cdeb96
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Tue Aug 6 13:09:08 2019 -0400
Adding SCP sender and framework improvements
---
.gitignore | 32 +++++
.../airavata/mft/core/api/StreamedReceiver.java | 2 +-
.../airavata/mft/core/api/StreamedSender.java | 2 +-
.../airavata/mft/core/api/TransportOperator.java | 5 +
.../streaming/DoubleByteArrayOutputStream.java | 50 ++++----
.../mft/core/streaming/TransportMetadata.java | 13 +++
.../mft/core/streaming/TransportStream.java | 46 ++++++--
.../apache/airavata/mft/transport/scp/Main.java | 38 ++++--
.../airavata/mft/transport/scp/SCPReceiver.java | 48 +-------
.../airavata/mft/transport/scp/SCPSender.java | 129 ++++++++++++++++++++-
.../mft/transport/scp/SCPTransportOperator.java | 25 ++++
.../mft/transport/scp/SCPTransportUtil.java | 60 ++++++++++
12 files changed, 362 insertions(+), 88 deletions(-)
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..57eb21d
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,32 @@
+# From https://github.com/github/gitignore/blob/master/Global/JetBrains.gitignore
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm
+
+## Directory-based project format
+.idea/
+# if you remove the above rule, at least ignore user-specific stuff:
+# .idea/workspace.xml
+# .idea/tasks.xml
+# and these sensitive or high-churn files:
+# .idea/dataSources.ids
+# .idea/dataSources.xml
+# .idea/sqlDataSources.xml
+# .idea/dynamic.xml
+
+## File-based project format
+*.ipr
+*.iws
+*.iml
+*target
+## Additional for IntelliJ
+out/
+
+# generated by mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# generated by JIRA plugin
+atlassian-ide-plugin.xml
+
+# generated by Crashlytics plugin (for Android Studio and Intellij)
+com_crashlytics_export_strings.xml
+
+.DS_Store
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
index 41ec680..56a070c 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
@@ -22,5 +22,5 @@ package org.apache.airavata.mft.core.api;
import org.apache.airavata.mft.core.streaming.TransportStream;
public interface StreamedReceiver {
- public void receive(String resourceIdentifier, TransportStream stream) throws Exception;
+ public void receive(TransportStream stream) throws Exception;
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
index a5430fb..1638a08 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
@@ -22,5 +22,5 @@ package org.apache.airavata.mft.core.api;
import org.apache.airavata.mft.core.streaming.TransportStream;
public interface StreamedSender {
- public void send(String resourceIdentifier, TransportStream stream) throws Exception;
+ public void send(TransportStream stream) throws Exception;
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/TransportOperator.java b/core/src/main/java/org/apache/airavata/mft/core/api/TransportOperator.java
new file mode 100644
index 0000000..cda2d5e
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/TransportOperator.java
@@ -0,0 +1,5 @@
+package org.apache.airavata.mft.core.api;
+
+public interface TransportOperator {
+ public long getResourceSize(String resourceId) throws Exception;
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
index f6919cf..1f1b56b 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
@@ -33,6 +33,8 @@ public class DoubleByteArrayOutputStream extends OutputStream {
private long processedBytes = 0;
private boolean clearedNonActiveStream = false;
+ private boolean isClosed = false;
+
@Override
public void write(int b) throws IOException {
this.currentStream.write(b);
@@ -41,29 +43,33 @@ public class DoubleByteArrayOutputStream extends OutputStream {
@Override
public void write(byte[] b, int off, int len) throws IOException {
if (processedBytes > maxBytesPerStream) {
- while (!clearedNonActiveStream) {
- try {
- Thread.sleep(100);
- System.out.println("Waiting until non active buffer gets emptied");
- } catch (InterruptedException e) {
- // ignore
- }
- }
- if (activeStream == 1) {
- activeStream = 2;
- currentStream = stream2;
- } else {
- activeStream = 1;
- currentStream = stream1;
- }
- currentStream.reset();
- processedBytes = 0;
- clearedNonActiveStream = false;
+ swapBuffers();
}
this.currentStream.write(b, off, len);
processedBytes += len;
}
+ private void swapBuffers() {
+ while (!clearedNonActiveStream) {
+ try {
+ Thread.sleep(100);
+ System.out.println("Receiver waiting until non active buffer gets emptied");
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ }
+ if (activeStream == 1) {
+ activeStream = 2;
+ currentStream = stream2;
+ } else {
+ activeStream = 1;
+ currentStream = stream1;
+ }
+ currentStream.reset();
+ processedBytes = 0;
+ clearedNonActiveStream = false;
+ }
+
@Override
public void write(byte[] b) throws IOException {
this.currentStream.write(b);
@@ -76,8 +82,12 @@ public class DoubleByteArrayOutputStream extends OutputStream {
@Override
public void close() throws IOException {
- this.stream1.close();
- this.stream2.close();
+ if (!isClosed) {
+ swapBuffers();
+ //this.stream1.close();
+ //this.stream2.close();
+ }
+ isClosed = true;
}
public InputStream asInputStream() {
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportMetadata.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportMetadata.java
new file mode 100644
index 0000000..1442052
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportMetadata.java
@@ -0,0 +1,13 @@
+package org.apache.airavata.mft.core.streaming;
+
+public class TransportMetadata {
+ private long length;
+
+ public long getLength() {
+ return length;
+ }
+
+ public void setLength(long length) {
+ this.length = length;
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
index 35956de..1f5bfd3 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
@@ -24,12 +24,19 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class TransportStream {
+ private TransportMetadata metadata;
+ private AtomicBoolean streamCompleted = new AtomicBoolean(false);
+
private OutputStream outputStream = new DoubleByteArrayOutputStream();
private InputStream inputStream = new DoubleByteArrayInputStream((DoubleByteArrayOutputStream) outputStream);
- private long length = -1;
- private AtomicBoolean streamCompleted = new AtomicBoolean(false);
- public TransportStream() throws IOException {
+ private String sourceId;
+ private String destId;
+
+ public TransportStream(String sourceId, String destId, TransportMetadata metadata) throws Exception {
+ this.sourceId = sourceId;
+ this.destId = destId;
+ this.metadata = metadata;
}
public InputStream getInputStream() {
@@ -48,19 +55,36 @@ public class TransportStream {
this.outputStream = outputStream;
}
- public long getLength() {
- return length;
+ public TransportMetadata getMetadata() {
+ return metadata;
+ }
+
+ public void setMetadata(TransportMetadata metadata) {
+ this.metadata = metadata;
+ }
+
+ public Boolean isStreamCompleted() {
+ return streamCompleted.get();
+ }
+
+ public void setStreamCompleted(boolean completed) throws IOException {
+ this.outputStream.close();
+ this.streamCompleted.set(completed);
+ }
+
+ public String getSourceId() {
+ return sourceId;
}
- public void setLength(long length) {
- this.length = length;
+ public void setSourceId(String sourceId) {
+ this.sourceId = sourceId;
}
- public AtomicBoolean getStreamCompleted() {
- return streamCompleted;
+ public String getDestId() {
+ return destId;
}
- public void setStreamCompleted(AtomicBoolean streamCompleted) {
- this.streamCompleted = streamCompleted;
+ public void setDestId(String destId) {
+ this.destId = destId;
}
}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
index 2047cd2..3dbeae6 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
@@ -19,15 +19,18 @@
*/
package org.apache.airavata.mft.transport.scp;
-import com.jcraft.jsch.JSchException;
+import org.apache.airavata.mft.core.streaming.TransportMetadata;
import org.apache.airavata.mft.core.streaming.TransportStream;
import java.io.IOException;
public class Main {
- public static void main(final String[] arg) throws IOException, JSchException {
+ public static void main(final String[] arg) throws Exception {
- final TransportStream stream = new TransportStream();
+ TransportMetadata metadata = new TransportMetadata();
+ SCPTransportOperator operator = new SCPTransportOperator();
+ metadata.setLength(operator.getResourceSize("1"));
+ final TransportStream stream = new TransportStream("1", "2", metadata);
Runnable r1 = new Runnable() {
@Override
@@ -42,7 +45,7 @@ public class Main {
//char c = (char)asInt;
//System.out.print(c);
} else {
- if (stream.getStreamCompleted().get()) {
+ if (stream.isStreamCompleted()) {
break;
} else {
try {
@@ -60,23 +63,38 @@ public class Main {
}
};
- Runnable r2 = new Runnable() {
+ Runnable receiverRun = new Runnable() {
@Override
public void run() {
SCPReceiver receiver = new SCPReceiver();
try {
- receiver.receive("1", stream);
+ receiver.receive(stream);
} catch (Exception e) {
e.printStackTrace();
}
}
};
- Thread t1 = new Thread(r1);
- t1.start();
+ Runnable senderRun = new Runnable() {
+ @Override
+ public void run() {
+ SCPSender sender = new SCPSender();
+ try {
+ sender.send(stream);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ };
+
+ //Thread t1 = new Thread(r1);
+ //t1.start();
+
+ Thread senderThread = new Thread(senderRun);
+ senderThread.start();
- Thread t2 = new Thread(r2);
- t2.start();
+ Thread receiverThread = new Thread(receiverRun);
+ receiverThread.start();
System.out.println("Done");
}
}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
index 61307e2..4093fa7 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
@@ -24,32 +24,19 @@ import org.apache.airavata.mft.core.api.StreamedReceiver;
import org.apache.airavata.mft.core.streaming.TransportStream;
import java.io.*;
-import java.util.Properties;
public class SCPReceiver implements StreamedReceiver {
@Override
- public void receive(String resourceIdentifier, TransportStream stream) throws Exception {
- SSHResourceIdentifier sshResourceIdentifier = getSSHResourceIdentifier(resourceIdentifier);
- Session session = createSession(sshResourceIdentifier.getUser(), sshResourceIdentifier.getHost(),
+ public void receive(TransportStream stream) throws Exception {
+ SSHResourceIdentifier sshResourceIdentifier = SCPTransportUtil.getSSHResourceIdentifier(stream.getSourceId());
+ Session session = SCPTransportUtil.createSession(sshResourceIdentifier.getUser(), sshResourceIdentifier.getHost(),
sshResourceIdentifier.getPort(),
sshResourceIdentifier.getKeyFile(),
sshResourceIdentifier.getKeyPassphrase());
transferRemoteToStream(session, sshResourceIdentifier.getRemotePath(), stream);
}
- // TODO replace with an API call to the registry
- private SSHResourceIdentifier getSSHResourceIdentifier(String resourceId) {
- SSHResourceIdentifier identifier = new SSHResourceIdentifier();
- identifier.setHost("pgadev.scigap.org");
- identifier.setUser("pga");
- identifier.setPort(22);
- identifier.setKeyFile("/Users/user/.ssh/id_rsa");
- identifier.setKeyPassphrase(null);
- identifier.setRemotePath("/var/www/portals/gateway-user-data/dev-seagrid/eromads6/DefaultProject/Gaussian_C11470169729/file.txt");
- return identifier;
- }
-
private void transferRemoteToStream(Session session, String from, TransportStream stream) throws JSchException, IOException {
// exec 'scp -f rfile' remotely
@@ -99,7 +86,6 @@ public class SCPReceiver implements StreamedReceiver {
}
System.out.println("file-size=" + filesize + ", file=" + file);
- stream.setLength(filesize);
// send '\0'
buf[0] = 0;
out.write(buf, 0, 1);
@@ -132,7 +118,7 @@ public class SCPReceiver implements StreamedReceiver {
out.flush();
}
- stream.getStreamCompleted().set(true);
+ stream.setStreamCompleted(true);
channel.disconnect();
session.disconnect();
}
@@ -163,30 +149,4 @@ public class SCPReceiver implements StreamedReceiver {
}
return b;
}
-
- private static Session createSession(String user, String host, int port, String keyFilePath, String keyPassword) {
- try {
- JSch jsch = new JSch();
-
- if (keyFilePath != null) {
- if (keyPassword != null) {
- jsch.addIdentity(keyFilePath, keyPassword);
- } else {
- jsch.addIdentity(keyFilePath);
- }
- }
-
- Properties config = new java.util.Properties();
- config.put("StrictHostKeyChecking", "no");
-
- Session session = jsch.getSession(user, host, port);
- session.setConfig(config);
- session.connect();
-
- return session;
- } catch (JSchException e) {
- System.out.println(e);
- return null;
- }
- }
}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
index d0a99f5..6b4dfe7 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
@@ -19,12 +19,139 @@
*/
package org.apache.airavata.mft.transport.scp;
+import com.jcraft.jsch.Channel;
+import com.jcraft.jsch.ChannelExec;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
import org.apache.airavata.mft.core.api.StreamedSender;
import org.apache.airavata.mft.core.streaming.TransportStream;
+import java.io.*;
+
public class SCPSender implements StreamedSender {
+
@Override
- public void send(String resourceIdentifier, TransportStream stream) throws Exception {
+ public void send(TransportStream stream) throws Exception {
+ SSHResourceIdentifier sshResourceIdentifier = SCPTransportUtil.getSSHResourceIdentifier(stream.getDestId());
+ Session session = SCPTransportUtil.createSession(sshResourceIdentifier.getUser(), sshResourceIdentifier.getHost(),
+ sshResourceIdentifier.getPort(),
+ sshResourceIdentifier.getKeyFile(),
+ sshResourceIdentifier.getKeyPassphrase());
+ copyLocalToRemote(session, sshResourceIdentifier.getRemotePath(), stream);
+ }
+
+ private static void copyLocalToRemote(Session session, String to, TransportStream stream) throws JSchException, IOException {
+ boolean ptimestamp = true;
+
+ // exec 'scp -t rfile' remotely
+ String command = "scp " + (ptimestamp ? "-p" : "") + " -t " + to;
+ Channel channel = session.openChannel("exec");
+ ((ChannelExec) channel).setCommand(command);
+
+ // get I/O streams for remote scp
+ OutputStream out = channel.getOutputStream();
+ InputStream in = channel.getInputStream();
+
+ channel.connect();
+
+ if (checkAck(in) != 0) {
+ System.exit(0);
+ }
+
+ if (ptimestamp) {
+ command = "T" + (System.currentTimeMillis() / 1000) + " 0";
+ // The access time should be sent here,
+ // but it is not accessible with JavaAPI ;-<
+ command += (" " + (System.currentTimeMillis() / 1000) + " 0\n");
+ out.write(command.getBytes());
+ out.flush();
+ if (checkAck(in) != 0) {
+ System.exit(0);
+ }
+ }
+
+ // send "C0644 filesize filename", where filename should not include '/'
+ long filesize = stream.getMetadata().getLength();
+ command = "C0644 " + filesize + " ";
+ if (to.lastIndexOf('/') > 0) {
+ command += to.substring(to.lastIndexOf('/') + 1);
+ } else {
+ command += to;
+ }
+
+ command += "\n";
+ out.write(command.getBytes());
+ out.flush();
+
+ if (checkAck(in) != 0) {
+ System.exit(0);
+ }
+
+ // send a content of lfile
+ byte[] buf = new byte[1024];
+ long totalWritten = 0;
+ while (true) {
+ int len = stream.getInputStream().read(buf, 0, buf.length);
+ if (len <= 0) {
+ if (totalWritten == stream.getMetadata().getLength()) {
+ break;
+ }
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ //ignore
+ }
+ System.out.println("SCP Sender waiting for data. Wrote " + totalWritten + " Total " + stream.getMetadata().getLength());
+ } else {
+ out.write(buf, 0, len); //out.flush();
+ totalWritten += len;
+ }
+ }
+
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
+
+ if (checkAck(in) != 0) {
+ System.exit(0);
+ }
+ out.close();
+
+ try {
+ if (stream.getInputStream() != null) stream.getInputStream().close();
+ } catch (Exception ex) {
+ System.out.println(ex);
+ }
+
+ channel.disconnect();
+ session.disconnect();
+ }
+
+ public static int checkAck(InputStream in) throws IOException {
+ int b = in.read();
+ // b may be 0 for success,
+ // 1 for error,
+ // 2 for fatal error,
+ // -1
+ if (b == 0) return b;
+ if (b == -1) return b;
+ if (b == 1 || b == 2) {
+ StringBuffer sb = new StringBuffer();
+ int c;
+ do {
+ c = in.read();
+ sb.append((char) c);
+ }
+ while (c != '\n');
+ if (b == 1) { // error
+ System.out.print(sb.toString());
+ }
+ if (b == 2) { // fatal error
+ System.out.print(sb.toString());
+ }
+ }
+ return b;
}
}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java
new file mode 100644
index 0000000..8a06a9f
--- /dev/null
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java
@@ -0,0 +1,25 @@
+package org.apache.airavata.mft.transport.scp;
+
+import com.jcraft.jsch.Channel;
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.Session;
+import com.jcraft.jsch.SftpATTRS;
+import org.apache.airavata.mft.core.api.TransportOperator;
+
+public class SCPTransportOperator implements TransportOperator {
+ @Override
+ public long getResourceSize(String resourceId) throws Exception {
+ SSHResourceIdentifier sshResourceIdentifier = SCPTransportUtil.getSSHResourceIdentifier(resourceId);
+ Session session = SCPTransportUtil.createSession(sshResourceIdentifier.getUser(), sshResourceIdentifier.getHost(),
+ sshResourceIdentifier.getPort(),
+ sshResourceIdentifier.getKeyFile(),
+ sshResourceIdentifier.getKeyPassphrase());
+ Channel channel=session.openChannel("sftp");
+ channel.connect();
+ ChannelSftp sftp =(ChannelSftp)channel;
+ SftpATTRS stat = sftp.stat(sshResourceIdentifier.getRemotePath());
+ channel.disconnect();
+ session.disconnect();
+ return stat.getSize();
+ }
+}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
new file mode 100644
index 0000000..fe70ec1
--- /dev/null
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
@@ -0,0 +1,60 @@
+package org.apache.airavata.mft.transport.scp;
+
+import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
+
+import java.util.Properties;
+
+public class SCPTransportUtil {
+ // TODO replace with an API call to the registry
+ public static SSHResourceIdentifier getSSHResourceIdentifier(String resourceId) {
+
+ SSHResourceIdentifier identifier = new SSHResourceIdentifier();
+ switch (resourceId){
+ case "1":
+ identifier.setHost("pgadev.scigap.org");
+ identifier.setUser("pga");
+ identifier.setPort(22);
+ identifier.setKeyFile("/Users/dwannipu/.ssh/id_rsa");
+ identifier.setKeyPassphrase(null);
+ identifier.setRemotePath("/var/www/portals/gateway-user-data/dev-seagrid/eromads6/DefaultProject/Gaussian_C11470169729/file.txt");
+ return identifier;
+ case "2":
+ identifier.setHost("pgadev.scigap.org");
+ identifier.setUser("pga");
+ identifier.setPort(22);
+ identifier.setKeyFile("/Users/dwannipu/.ssh/id_rsa");
+ identifier.setKeyPassphrase(null);
+ identifier.setRemotePath("/var/www/portals/gateway-user-data/dev-seagrid/eromads6/DefaultProject/Gaussian_C11470169729/new-file.txt");
+ return identifier;
+ }
+ return null;
+ }
+
+ public static Session createSession(String user, String host, int port, String keyFilePath, String keyPassword) {
+ try {
+ JSch jsch = new JSch();
+
+ if (keyFilePath != null) {
+ if (keyPassword != null) {
+ jsch.addIdentity(keyFilePath, keyPassword);
+ } else {
+ jsch.addIdentity(keyFilePath);
+ }
+ }
+
+ Properties config = new java.util.Properties();
+ config.put("StrictHostKeyChecking", "no");
+
+ Session session = jsch.getSession(user, host, port);
+ session.setConfig(config);
+ session.connect();
+
+ return session;
+ } catch (JSchException e) {
+ System.out.println(e);
+ return null;
+ }
+ }
+}
[airavata-mft] 19/22: TCP transport implementation
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 1d373ebaa6c9d5d1de1c3e995b825506b052278b
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Thu Aug 22 11:12:43 2019 +0530
TCP transport implementation
---
.../main/java/org/apache/airavata/mft/transport/tcp/Constants.java | 1 +
.../mft/transport/tcp/client/RemoteFileServerConnector.java | 2 +-
.../org/apache/airavata/mft/transport/tcp/server/FileServer.java | 6 ++----
.../airavata/mft/transport/tcp/server/FileServerConnector.java | 3 ---
4 files changed, 4 insertions(+), 8 deletions(-)
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
index ba07517..274f8af 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
@@ -26,4 +26,5 @@ public final class Constants {
public static final String UNKNOWN_HOST = "UNKNOWN_HOST";
public static int MAX_CACHING_CONNECTIONS = 100;
+ public static int MAX_NUM_OF_HOSTS = 100;
}
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
index b978e54..31fe519 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
@@ -32,7 +32,7 @@ import java.nio.channels.SocketChannel;
*/
public class RemoteFileServerConnector extends AbstractConnector implements SinkConnector {
- RemoteResourceIdentifier identifier;
+ private RemoteResourceIdentifier identifier;
public RemoteFileServerConnector(RemoteResourceIdentifier remoteResourceIdentifier) {
this.identifier = remoteResourceIdentifier;
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
index 38da89b..f41f673 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
@@ -33,15 +33,13 @@ import java.util.concurrent.ConcurrentHashMap;
*/
public class FileServer {
-
- //If too many host address
- private ConcurrentHashMap<String, FileServerConnector> remoteHostToConnectorMap = new ConcurrentHashMap<>();
+ private ConcurrentHashMap<String, FileServerConnector> remoteHostToConnectorMap =
+ new ConcurrentHashMap<>(Constants.MAX_NUM_OF_HOSTS);
private volatile boolean runServer = true;
/**
* This starts the server and accepts incoming connections
- *
* @return
*/
public void start(int port) {
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
index 0183632..aa617d5 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
@@ -36,7 +36,6 @@ public class FileServerConnector extends AbstractConnector implements SourceConn
private LinkedBlockingQueue<SocketChannel> acceptedChannelCache = new
LinkedBlockingQueue<>(Constants.MAX_CACHING_CONNECTIONS);
-
@Override
public ConnectorChannel openChannel() throws Exception {
SocketChannel socketChannel = this.acceptedChannelCache.take();
@@ -45,10 +44,8 @@ public class FileServerConnector extends AbstractConnector implements SourceConn
}
-
public void addChannel(SocketChannel channel) throws InterruptedException {
this.acceptedChannelCache.put(channel);
}
-
}
[airavata-mft] 03/22: Initial stream based data movement framework
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 036213f18f6b8a473a620481b3826535ee522e46
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Tue Aug 6 02:45:28 2019 -0400
Initial stream based data movement framework
---
core/pom.xml | 15 ++
.../airavata/mft/core/api/StreamedReceiver.java | 7 +
.../airavata/mft/core/api/StreamedSender.java | 7 +
.../core/streaming/DoubleByteArrayInputStream.java | 36 +++++
.../streaming/DoubleByteArrayOutputStream.java | 72 +++++++++
.../mft/core/streaming/TransportStream.java | 47 ++++++
pom.xml | 82 ++++++++++
transport/pom.xml | 19 +++
transport/scp-transport/pom.xml | 27 ++++
.../apache/airavata/mft/transport/scp/Main.java | 63 ++++++++
.../airavata/mft/transport/scp/SCPReceiver.java | 173 +++++++++++++++++++++
.../airavata/mft/transport/scp/SCPSender.java | 11 ++
.../mft/transport/scp/SSHResourceIdentifier.java | 67 ++++++++
13 files changed, 626 insertions(+)
diff --git a/core/pom.xml b/core/pom.xml
new file mode 100644
index 0000000..c87446e
--- /dev/null
+++ b/core/pom.xml
@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>airavata-mft</artifactId>
+ <groupId>org.apache.airavata</groupId>
+ <version>0.18-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>mft-core</artifactId>
+
+
+</project>
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
new file mode 100644
index 0000000..47db51e
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
@@ -0,0 +1,7 @@
+package org.apache.airavata.mft.core.api;
+
+import org.apache.airavata.mft.core.streaming.TransportStream;
+
+public interface StreamedReceiver {
+ public void receive(String resourceIdentifier, TransportStream stream) throws Exception;
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
new file mode 100644
index 0000000..9d3986c
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
@@ -0,0 +1,7 @@
+package org.apache.airavata.mft.core.api;
+
+import org.apache.airavata.mft.core.streaming.TransportStream;
+
+public interface StreamedSender {
+ public void send(String resourceIdentifier, TransportStream stream) throws Exception;
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
new file mode 100644
index 0000000..56f5dc5
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
@@ -0,0 +1,36 @@
+package org.apache.airavata.mft.core.streaming;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+public class DoubleByteArrayInputStream extends InputStream {
+
+ private DoubleByteArrayOutputStream outputStream;
+ private InputStream currentInputStream;
+
+ public DoubleByteArrayInputStream(DoubleByteArrayOutputStream outputStream) {
+ this.outputStream = outputStream;
+ this.currentInputStream = outputStream.asInputStream();
+ }
+
+ @Override
+ public int read() throws IOException {
+ refresh();
+ return this.currentInputStream.read();
+ }
+
+ @Override
+ public int available() throws IOException {
+ refresh();
+ return this.currentInputStream.available();
+ }
+
+ private void refresh() throws IOException {
+ if (this.currentInputStream.available() == 0) {
+ InputStream tempInputStream = this.outputStream.asInputStream();
+ if (tempInputStream != null) {
+ this.currentInputStream = tempInputStream;
+ }
+ }
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
new file mode 100644
index 0000000..416a8eb
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
@@ -0,0 +1,72 @@
+package org.apache.airavata.mft.core.streaming;
+
+import java.io.*;
+
+public class DoubleByteArrayOutputStream extends OutputStream {
+
+ private ByteArrayOutputStream stream1 = new ByteArrayOutputStream();
+ private ByteArrayOutputStream stream2 = new ByteArrayOutputStream();
+
+ private ByteArrayOutputStream currentStream = stream1;
+ private int activeStream = 1;
+
+ private long maxBytesPerStream = 1 * 1000 * 1000;
+ private long processedBytes = 0;
+ private boolean clearedNonActiveStream = false;
+
+ @Override
+ public void write(int b) throws IOException {
+ this.currentStream.write(b);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ if (processedBytes > maxBytesPerStream) {
+ while (!clearedNonActiveStream) {
+ try {
+ Thread.sleep(100);
+ System.out.println("Waiting until non active buffer gets emptied");
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ }
+ if (activeStream == 1) {
+ activeStream = 2;
+ currentStream = stream2;
+ } else {
+ activeStream = 1;
+ currentStream = stream1;
+ }
+ currentStream.reset();
+ processedBytes = 0;
+ clearedNonActiveStream = false;
+ }
+ this.currentStream.write(b, off, len);
+ processedBytes += len;
+ }
+
+ @Override
+ public void write(byte[] b) throws IOException {
+ this.currentStream.write(b);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ this.currentStream.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.stream1.close();
+ this.stream2.close();
+ }
+
+ public InputStream asInputStream() {
+ if (clearedNonActiveStream) {
+ return null;
+ }
+ ByteArrayInputStream inputStream = new ByteArrayInputStream(activeStream == 1 ? stream2.toByteArray(): stream1.toByteArray());
+ this.clearedNonActiveStream = true;
+ return inputStream;
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
new file mode 100644
index 0000000..d8a95b6
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
@@ -0,0 +1,47 @@
+package org.apache.airavata.mft.core.streaming;
+
+import java.io.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class TransportStream {
+
+ private OutputStream outputStream = new DoubleByteArrayOutputStream();
+ private InputStream inputStream = new DoubleByteArrayInputStream((DoubleByteArrayOutputStream) outputStream);
+ private long length = -1;
+ private AtomicBoolean streamCompleted = new AtomicBoolean(false);
+
+ public TransportStream() throws IOException {
+ }
+
+ public InputStream getInputStream() {
+ return inputStream;
+ }
+
+ public void setInputStream(InputStream inputStream) {
+ this.inputStream = inputStream;
+ }
+
+ public OutputStream getOutputStream() {
+ return outputStream;
+ }
+
+ public void setOutputStream(OutputStream outputStream) {
+ this.outputStream = outputStream;
+ }
+
+ public long getLength() {
+ return length;
+ }
+
+ public void setLength(long length) {
+ this.length = length;
+ }
+
+ public AtomicBoolean getStreamCompleted() {
+ return streamCompleted;
+ }
+
+ public void setStreamCompleted(AtomicBoolean streamCompleted) {
+ this.streamCompleted = streamCompleted;
+ }
+}
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..983b1cc
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <prerequisites>
+ <maven>3.0</maven>
+ </prerequisites>
+ <modules>
+ <module>core</module>
+ <module>transport</module>
+ </modules>
+
+ <parent>
+ <groupId>org.apache</groupId>
+ <artifactId>apache</artifactId>
+ <version>21</version>
+ </parent>
+
+ <groupId>org.apache.airavata</groupId>
+ <artifactId>airavata-mft</artifactId>
+ <packaging>pom</packaging>
+ <name>Airavata MFT</name>
+ <version>0.18-SNAPSHOT</version>
+
+ <url>http://airavata.apache.org/</url>
+ <inceptionYear>2011</inceptionYear>
+
+ <scm>
+ <connection>scm:git:https://github.com/apache/airavata.git</connection>
+ <developerConnection>scm:git:https://github.com/apache/airavata.git</developerConnection>
+ <url>https://github.com/apache/airavata</url>
+ <tag>HEAD</tag>
+ </scm>
+
+ <mailingLists>
+
+ <mailingList>
+ <name>Airavata Developer List</name>
+ <subscribe>dev-subscribe@airavata.apache.org</subscribe>
+ <unsubscribe>dev-unsubscribe@airavata.apache.org</unsubscribe>
+ <post>mailto:dev@airavata.apache.org</post>
+ <archive>http://mail-archives.apache.org/mod_mbox/airavata-dev/</archive>
+ </mailingList>
+
+ <mailingList>
+ <name>Airavata Users List</name>
+ <subscribe>users-subscribe@airavata.apache.org</subscribe>
+ <unsubscribe>users-unsubscribe@airavata.apache.org</unsubscribe>
+ <post>mailto:users@airavata.apache.org</post>
+ <archive>http://mail-archives.apache.org/mod_mbox/airavata-users/</archive>
+ </mailingList>
+
+ </mailingLists>
+
+ <issueManagement>
+ <url>https://issues.apache.org/jira/browse/AIRAVATA</url>
+ </issueManagement>
+
+</project>
\ No newline at end of file
diff --git a/transport/pom.xml b/transport/pom.xml
new file mode 100644
index 0000000..5e26abd
--- /dev/null
+++ b/transport/pom.xml
@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>airavata-mft</artifactId>
+ <groupId>org.apache.airavata</groupId>
+ <version>0.18-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>mft-transport</artifactId>
+ <packaging>pom</packaging>
+ <modules>
+ <module>scp-transport</module>
+ </modules>
+
+
+</project>
\ No newline at end of file
diff --git a/transport/scp-transport/pom.xml b/transport/scp-transport/pom.xml
new file mode 100644
index 0000000..f48bd0d
--- /dev/null
+++ b/transport/scp-transport/pom.xml
@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>mft-transport</artifactId>
+ <groupId>org.apache.airavata</groupId>
+ <version>0.18-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>mft-scp-transport</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>com.jcraft</groupId>
+ <artifactId>jsch</artifactId>
+ <version>0.1.55</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.airavata</groupId>
+ <artifactId>mft-core</artifactId>
+ <version>0.18-SNAPSHOT</version>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
new file mode 100644
index 0000000..33fdb9f
--- /dev/null
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
@@ -0,0 +1,63 @@
+package org.apache.airavata.mft.transport.scp;
+
+import com.jcraft.jsch.JSchException;
+import org.apache.airavata.mft.core.streaming.TransportStream;
+
+import java.io.IOException;
+
+public class Main {
+ public static void main(final String[] arg) throws IOException, JSchException {
+
+ final TransportStream stream = new TransportStream();
+
+ Runnable r1 = new Runnable() {
+ @Override
+ public void run() {
+ int asInt;
+ try {
+ long read = 0;
+ while (true) {
+ if (stream.getInputStream().available() > 0) {
+ asInt = stream.getInputStream().read();
+ read++;
+ //char c = (char)asInt;
+ //System.out.print(c);
+ } else {
+ if (stream.getStreamCompleted().get()) {
+ break;
+ } else {
+ try {
+ Thread.sleep(100);
+ System.out.println("Waiting " + read);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ };
+
+ Runnable r2 = new Runnable() {
+ @Override
+ public void run() {
+ SCPReceiver receiver = new SCPReceiver();
+ try {
+ receiver.receive("1", stream);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ };
+
+ Thread t1 = new Thread(r1);
+ t1.start();
+
+ Thread t2 = new Thread(r2);
+ t2.start();
+ System.out.println("Done");
+ }
+}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
new file mode 100644
index 0000000..bba84b8
--- /dev/null
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
@@ -0,0 +1,173 @@
+package org.apache.airavata.mft.transport.scp;
+
+import com.jcraft.jsch.*;
+import org.apache.airavata.mft.core.api.StreamedReceiver;
+import org.apache.airavata.mft.core.streaming.TransportStream;
+
+import java.io.*;
+import java.util.Properties;
+
+public class SCPReceiver implements StreamedReceiver {
+
+ @Override
+ public void receive(String resourceIdentifier, TransportStream stream) throws Exception {
+ SSHResourceIdentifier sshResourceIdentifier = getSSHResourceIdentifier(resourceIdentifier);
+ Session session = createSession(sshResourceIdentifier.getUser(), sshResourceIdentifier.getHost(),
+ sshResourceIdentifier.getPort(),
+ sshResourceIdentifier.getKeyFile(),
+ sshResourceIdentifier.getKeyPassphrase());
+ transferRemoteToStream(session, sshResourceIdentifier.getRemotePath(), stream);
+ }
+
+ // TODO replace with an API call to the registry
+ private SSHResourceIdentifier getSSHResourceIdentifier(String resourceId) {
+ SSHResourceIdentifier identifier = new SSHResourceIdentifier();
+ identifier.setHost("pgadev.scigap.org");
+ identifier.setUser("pga");
+ identifier.setPort(22);
+ identifier.setKeyFile("/Users/user/.ssh/id_rsa");
+ identifier.setKeyPassphrase(null);
+ identifier.setRemotePath("/var/www/portals/gateway-user-data/dev-seagrid/eromads6/DefaultProject/Gaussian_C11470169729/file.txt");
+ return identifier;
+ }
+
+ private void transferRemoteToStream(Session session, String from, TransportStream stream) throws JSchException, IOException {
+
+ // exec 'scp -f rfile' remotely
+ String command = "scp -f " + from;
+ Channel channel = session.openChannel("exec");
+ ((ChannelExec) channel).setCommand(command);
+
+ // get I/O streams for remote scp
+ OutputStream out = channel.getOutputStream();
+ InputStream in = channel.getInputStream();
+
+ channel.connect();
+
+ byte[] buf = new byte[1024];
+
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
+
+ while (true) {
+ int c = checkAck(in);
+ if (c != 'C') {
+ break;
+ }
+
+ // read '0644 '
+ in.read(buf, 0, 5);
+
+ long filesize = 0L;
+ while (true) {
+ if (in.read(buf, 0, 1) < 0) {
+ // error
+ break;
+ }
+ if (buf[0] == ' ') break;
+ filesize = filesize * 10L + (long) (buf[0] - '0');
+ }
+
+ String file = null;
+ for (int i = 0; ; i++) {
+ in.read(buf, i, 1);
+ if (buf[i] == (byte) 0x0a) {
+ file = new String(buf, 0, i);
+ break;
+ }
+ }
+
+ System.out.println("file-size=" + filesize + ", file=" + file);
+ stream.setLength(filesize);
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
+
+ // read a content of lfile
+ int bufSize;
+ while (true) {
+ if (buf.length < filesize) bufSize = buf.length;
+ else bufSize = (int) filesize;
+ bufSize = in.read(buf, 0, bufSize);
+ if (bufSize < 0) {
+ // error
+ break;
+ }
+ stream.getOutputStream().write(buf, 0, bufSize);
+ stream.getOutputStream().flush();
+
+ filesize -= bufSize;
+ if (filesize == 0L) break;
+ }
+
+ if (checkAck(in) != 0) {
+ System.exit(0);
+ }
+
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
+ }
+
+ stream.getStreamCompleted().set(true);
+ channel.disconnect();
+ session.disconnect();
+ }
+
+ public int checkAck(InputStream in) throws IOException {
+ int b = in.read();
+ // b may be 0 for success,
+ // 1 for error,
+ // 2 for fatal error,
+ // -1
+ if (b == 0) return b;
+ if (b == -1) return b;
+
+ if (b == 1 || b == 2) {
+ StringBuffer sb = new StringBuffer();
+ int c;
+ do {
+ c = in.read();
+ sb.append((char) c);
+ }
+ while (c != '\n');
+ if (b == 1) { // error
+ System.out.print(sb.toString());
+ }
+ if (b == 2) { // fatal error
+ System.out.print(sb.toString());
+ }
+ }
+ return b;
+ }
+
+ private static Session createSession(String user, String host, int port, String keyFilePath, String keyPassword) {
+ try {
+ JSch jsch = new JSch();
+
+ if (keyFilePath != null) {
+ if (keyPassword != null) {
+ jsch.addIdentity(keyFilePath, keyPassword);
+ } else {
+ jsch.addIdentity(keyFilePath);
+ }
+ }
+
+ Properties config = new java.util.Properties();
+ config.put("StrictHostKeyChecking", "no");
+
+ Session session = jsch.getSession(user, host, port);
+ session.setConfig(config);
+ session.connect();
+
+ return session;
+ } catch (JSchException e) {
+ System.out.println(e);
+ return null;
+ }
+ }
+}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
new file mode 100644
index 0000000..64e2210
--- /dev/null
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
@@ -0,0 +1,11 @@
+package org.apache.airavata.mft.transport.scp;
+
+import org.apache.airavata.mft.core.api.StreamedSender;
+import org.apache.airavata.mft.core.streaming.TransportStream;
+
+public class SCPSender implements StreamedSender {
+ @Override
+ public void send(String resourceIdentifier, TransportStream stream) throws Exception {
+
+ }
+}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java
new file mode 100644
index 0000000..6335b7c
--- /dev/null
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java
@@ -0,0 +1,67 @@
+package org.apache.airavata.mft.transport.scp;
+
+public class SSHResourceIdentifier {
+ private String id;
+ private String remotePath;
+ private String host;
+ private String user;
+ private int port;
+ private String keyFile;
+ private String keyPassphrase;
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public String getRemotePath() {
+ return remotePath;
+ }
+
+ public void setRemotePath(String remotePath) {
+ this.remotePath = remotePath;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public String getUser() {
+ return user;
+ }
+
+ public void setUser(String user) {
+ this.user = user;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ public String getKeyFile() {
+ return keyFile;
+ }
+
+ public void setKeyFile(String keyFile) {
+ this.keyFile = keyFile;
+ }
+
+ public String getKeyPassphrase() {
+ return keyPassphrase;
+ }
+
+ public void setKeyPassphrase(String keyPassphrase) {
+ this.keyPassphrase = keyPassphrase;
+ }
+}
[airavata-mft] 01/22: Bootstrapping the repo
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 2a1a4416ed9b4cae51895573d4f176102b93bc73
Author: Suresh Marru <sm...@apache.org>
AuthorDate: Fri Jun 21 15:44:21 2019 -0400
Bootstrapping the repo
---
README | 1 +
1 file changed, 1 insertion(+)
diff --git a/README b/README
new file mode 100644
index 0000000..45d049e
--- /dev/null
+++ b/README
@@ -0,0 +1 @@
+Airavata Managed File Transfer Service and Clients
\ No newline at end of file
[airavata-mft] 08/22: Adding S3 stream receiver
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 360e5ad33c33eb9d400e37321a0708c7abb1b774
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Tue Aug 6 15:47:02 2019 -0400
Adding S3 stream receiver
---
.../streaming/DoubleByteArrayOutputStream.java | 4 ++
transport/pom.xml | 1 +
transport/{scp-transport => s3-transport}/pom.xml | 8 +--
.../airavata/mft/transport/s3/S3Receiver.java | 45 ++++++++++++++
.../mft/transport/s3/S3ResourceIdentifier.java | 68 ++++++++++++++++++++++
.../apache/airavata/mft/transport/s3/S3Sender.java | 30 ++++++++++
.../mft/transport/s3/S3TransportOperator.java | 36 ++++++++++++
.../airavata/mft/transport/s3/S3TransportUtil.java | 55 +++++++++++++++++
transport/scp-transport/pom.xml | 5 ++
.../apache/airavata/mft/transport/scp/Main.java | 12 ++--
10 files changed, 256 insertions(+), 8 deletions(-)
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
index 56728c3..799ba42 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
@@ -44,7 +44,11 @@ public class DoubleByteArrayOutputStream extends OutputStream {
@Override
public void write(int b) throws IOException {
+ if (processedBytes > maxBytesPerStream) {
+ swapBuffers();
+ }
this.currentStream.write(b);
+ processedBytes += 1;
}
@Override
diff --git a/transport/pom.xml b/transport/pom.xml
index 3bcbefa..0d4952a 100644
--- a/transport/pom.xml
+++ b/transport/pom.xml
@@ -34,6 +34,7 @@
<packaging>pom</packaging>
<modules>
<module>scp-transport</module>
+ <module>s3-transport</module>
</modules>
diff --git a/transport/scp-transport/pom.xml b/transport/s3-transport/pom.xml
similarity index 89%
copy from transport/scp-transport/pom.xml
copy to transport/s3-transport/pom.xml
index eccfdca..9395db8 100644
--- a/transport/scp-transport/pom.xml
+++ b/transport/s3-transport/pom.xml
@@ -30,13 +30,13 @@
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>mft-scp-transport</artifactId>
+ <artifactId>mft-s3-transport</artifactId>
<dependencies>
<dependency>
- <groupId>com.jcraft</groupId>
- <artifactId>jsch</artifactId>
- <version>0.1.55</version>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk</artifactId>
+ <version>1.11.163</version>
</dependency>
<dependency>
<groupId>org.apache.airavata</groupId>
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java
new file mode 100644
index 0000000..d7c516a
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java
@@ -0,0 +1,45 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.airavata.mft.transport.s3;
+
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.S3Object;
+import com.amazonaws.services.s3.model.S3ObjectInputStream;
+import org.apache.airavata.mft.core.api.StreamedReceiver;
+import org.apache.airavata.mft.core.streaming.TransportStream;
+
+public class S3Receiver implements StreamedReceiver {
+
+ @Override
+ public void receive(TransportStream stream) throws Exception {
+
+ S3ResourceIdentifier resourceIdentifier = S3TransportUtil.getS3ResourceIdentifier(stream.getSourceId());
+ AmazonS3 s3client = S3TransportUtil.getS3Client(resourceIdentifier.getAccessKey(),
+ resourceIdentifier.getSecretKey(), resourceIdentifier.getRegion());
+ S3Object s3object = s3client.getObject(resourceIdentifier.getBucket(), resourceIdentifier.getRemoteFile());
+ S3ObjectInputStream inputStream = s3object.getObjectContent();
+
+ int read;
+ while ((read = inputStream.read()) != -1) {
+ stream.getOutputStream().write(read);
+ }
+ stream.setStreamCompleted(true);
+ }
+}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3ResourceIdentifier.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3ResourceIdentifier.java
new file mode 100644
index 0000000..085589d
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3ResourceIdentifier.java
@@ -0,0 +1,68 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.airavata.mft.transport.s3;
+
+public class S3ResourceIdentifier {
+ private String bucket;
+ private String remoteFile;
+ private String region;
+ private String accessKey;
+ private String secretKey;
+
+ public String getBucket() {
+ return bucket;
+ }
+
+ public void setBucket(String bucket) {
+ this.bucket = bucket;
+ }
+
+ public String getRemoteFile() {
+ return remoteFile;
+ }
+
+ public void setRemoteFile(String remoteFile) {
+ this.remoteFile = remoteFile;
+ }
+
+ public String getRegion() {
+ return region;
+ }
+
+ public void setRegion(String region) {
+ this.region = region;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
new file mode 100644
index 0000000..cdeb1e5
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
@@ -0,0 +1,30 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.airavata.mft.transport.s3;
+
+import org.apache.airavata.mft.core.api.StreamedSender;
+import org.apache.airavata.mft.core.streaming.TransportStream;
+
+public class S3Sender implements StreamedSender {
+ @Override
+ public void send(TransportStream stream) throws Exception {
+
+ }
+}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportOperator.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportOperator.java
new file mode 100644
index 0000000..79db705
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportOperator.java
@@ -0,0 +1,36 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.airavata.mft.transport.s3;
+
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.S3Object;
+import org.apache.airavata.mft.core.api.TransportOperator;
+
+public class S3TransportOperator implements TransportOperator {
+
+ @Override
+ public long getResourceSize(String resourceId) throws Exception {
+ S3ResourceIdentifier resourceIdentifier = S3TransportUtil.getS3ResourceIdentifier(resourceId);
+ AmazonS3 s3client = S3TransportUtil.getS3Client(resourceIdentifier.getAccessKey(),
+ resourceIdentifier.getSecretKey(), resourceIdentifier.getRegion());
+ S3Object s3object = s3client.getObject(resourceIdentifier.getBucket(), resourceIdentifier.getRemoteFile());
+ return s3object.getObjectMetadata().getContentLength();
+ }
+}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java
new file mode 100644
index 0000000..ca694bb
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java
@@ -0,0 +1,55 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.airavata.mft.transport.s3;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+
+public class S3TransportUtil {
+ public static S3ResourceIdentifier getS3ResourceIdentifier(String resourceId) {
+ S3ResourceIdentifier identifier = new S3ResourceIdentifier();
+
+ switch (resourceId){
+ case "3":
+ identifier.setAccessKey("key");
+ identifier.setSecretKey("secret");
+ identifier.setBucket("airavata-s3");
+ identifier.setRegion("us-east-2");
+ identifier.setRemoteFile("file.txt");
+ return identifier;
+ default:
+ return null;
+ }
+ }
+
+ public static AmazonS3 getS3Client(String accessKey, String secretKey, String region) {
+ AWSCredentials credentials = new BasicAWSCredentials(accessKey, secretKey);
+ AmazonS3 s3client = AmazonS3ClientBuilder
+ .standard()
+ .withCredentials(new AWSStaticCredentialsProvider(credentials))
+ .withRegion(Regions.fromName(region))
+ .build();
+ return s3client;
+ }
+}
diff --git a/transport/scp-transport/pom.xml b/transport/scp-transport/pom.xml
index eccfdca..d3cbc4e 100644
--- a/transport/scp-transport/pom.xml
+++ b/transport/scp-transport/pom.xml
@@ -43,6 +43,11 @@
<artifactId>mft-core</artifactId>
<version>0.18-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.airavata</groupId>
+ <artifactId>mft-s3-transport</artifactId>
+ <version>0.18-SNAPSHOT</version>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
index 3dbeae6..00fe32e 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
@@ -21,6 +21,8 @@ package org.apache.airavata.mft.transport.scp;
import org.apache.airavata.mft.core.streaming.TransportMetadata;
import org.apache.airavata.mft.core.streaming.TransportStream;
+import org.apache.airavata.mft.transport.s3.S3Receiver;
+import org.apache.airavata.mft.transport.s3.S3TransportOperator;
import java.io.IOException;
@@ -28,9 +30,10 @@ public class Main {
public static void main(final String[] arg) throws Exception {
TransportMetadata metadata = new TransportMetadata();
- SCPTransportOperator operator = new SCPTransportOperator();
- metadata.setLength(operator.getResourceSize("1"));
- final TransportStream stream = new TransportStream("1", "2", metadata);
+ //SCPTransportOperator operator = new SCPTransportOperator();
+ S3TransportOperator operator = new S3TransportOperator();
+ metadata.setLength(operator.getResourceSize("3"));
+ final TransportStream stream = new TransportStream("3", "2", metadata);
Runnable r1 = new Runnable() {
@Override
@@ -66,7 +69,8 @@ public class Main {
Runnable receiverRun = new Runnable() {
@Override
public void run() {
- SCPReceiver receiver = new SCPReceiver();
+ //SCPReceiver receiver = new SCPReceiver();
+ S3Receiver receiver = new S3Receiver();
try {
receiver.receive(stream);
} catch (Exception e) {
[airavata-mft] 11/22: Remove senstive data
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit eda78727453b8deadaa5ae35329a224e2c75861c
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Fri Aug 16 06:53:43 2019 +0530
Remove senstive data
---
.../airavata/mft/core/api/ConnectorChannel.java | 4 ++--
.../org/apache/airavata/mft/core/api/Mediator.java | 3 +--
.../mft/core/bufferedImpl/AbstractConnector.java | 27 +++++++++++++++-------
.../mft/core/bufferedImpl/PassthroughMediator.java | 1 -
.../org/apache/airavata/mft/transport/s3/Main.java | 6 ++---
5 files changed, 25 insertions(+), 16 deletions(-)
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java b/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
index d6065b9..5e037b1 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
@@ -23,11 +23,11 @@ import java.io.IOException;
import java.nio.channels.Channel;
/**
- * An interface represents the underlying channel of a {@Link Connector}
+ * An interface represents a channel of a {@Link Connector}
*/
public interface ConnectorChannel {
/**
- * get the channel of given connector
+ * get the NIO channel of ChannelConnector
*
* @return
*/
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java b/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
index e3ea081..76e1f69 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
@@ -22,14 +22,13 @@ package org.apache.airavata.mft.core.api;
/**
* This represent the interface for all mediators, which mediates
- * data from source connectors and send them to sink connectors.
+ * data from source connector hannel and send them to sink connector channel.
*/
public interface Mediator {
/**
* Mediates the content from source connector and writes the mediated content
* to destination connector
- * WritableByteChannel
* @param src
* @param dst
* @param callback
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
index 2045356..20e5989 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
@@ -34,7 +34,7 @@ import java.util.concurrent.ConcurrentHashMap;
* This overrides closeChannel method and release all resources allocated to releasing channel
*/
public abstract class AbstractConnector implements Connector {
- ConcurrentHashMap<Channel, Object> channelToStreamMap = new ConcurrentHashMap<>();
+ ConcurrentHashMap<Channel, Object> channelMap = new ConcurrentHashMap<>();
@Override
public void closeChannel(Channel channel) throws ConnectorException {
@@ -42,17 +42,18 @@ public abstract class AbstractConnector implements Connector {
InChannel inChannel;
try {
- if (channelToStreamMap.get(channel) instanceof OutputStream) {
- outChannel = (OutChannel) channelToStreamMap.get(channel);
+ if (channelMap.get(channel) instanceof OutputStream) {
+ outChannel = (OutChannel) channelMap.get(channel);
if (outChannel != null) {
outChannel.closeChannel();
- System.out.println("Closing output stream");
+ channelMap.remove(channel);
}
- } else if (channelToStreamMap.get(channel) instanceof InputStream) {
- inChannel = (InChannel) channelToStreamMap.get(channel);
+ } else if (channelMap.get(channel) instanceof InputStream) {
+ inChannel = (InChannel) channelMap.get(channel);
if (inChannel != null) {
inChannel.closeChannel();
+ channelMap.remove(channel);
}
}
@@ -62,13 +63,23 @@ public abstract class AbstractConnector implements Connector {
}
+ /**
+ * Cache the operating channel
+ * @param channel
+ * @param obj
+ */
public void cacheChannel(Channel channel, Object obj) {
- channelToStreamMap.put(channel, obj);
+ channelMap.put(channel, obj);
}
+ /**
+ * get the channel
+ * @param channel
+ * @return
+ */
public Object getConnectorChannel(Channel channel) {
- return channelToStreamMap.get(channel);
+ return channelMap.get(channel);
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
index 36b6801..f25348c 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
@@ -47,7 +47,6 @@ public class PassthroughMediator implements Mediator {
ChannelUtils.copyData(rChannel, dChannel);
}
- // dst.closeChannel();
Object obj = dst.getChannelAttribute(Constants.CONNECTOR);
if (obj != null && obj instanceof SinkConnector) {
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
index 42cde42..8f3b959 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
@@ -42,12 +42,12 @@ public class Main {
public void execute() {
Properties connectorConfProp = new Properties();
- connectorConfProp.put(S3Constants.ACCESS_KEY, "AKIA2SKSIFUH7QDAROSR");
- connectorConfProp.put(S3Constants.SECRET_KEY, "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");
+ connectorConfProp.put(S3Constants.ACCESS_KEY, "XXXX");
+ connectorConfProp.put(S3Constants.SECRET_KEY, "YYY");
connectorConfProp.put(S3Constants.REGION, "us-east-2");
Properties srcProp = new Properties();
- srcProp.put(S3Constants.BUCKET, "blimpit-test");
+ srcProp.put(S3Constants.BUCKET, "test");
srcProp.put(S3Constants.REMOTE_FILE, "test.pdf");
Properties dstProp = new Properties();
[airavata-mft] 10/22: Adding NIO based file transferring initial
impl
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit c7b1eb4e1f325faa401e074927c8529277ef216a
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Thu Aug 15 13:24:40 2019 +0530
Adding NIO based file transferring initial impl
---
core/pom.xml | 12 +++
.../airavata/mft/core/api/CompletionCallback.java | 40 ++++++++++
.../apache/airavata/mft/core/api/Connector.java | 51 ++++++++++++
.../airavata/mft/core/api/ConnectorChannel.java | 58 ++++++++++++++
.../org/apache/airavata/mft/core/api/Mediator.java | 39 +++++++++
.../airavata/mft/core/api/SinkConnector.java | 43 ++++++++++
.../airavata/mft/core/api/SourceConnector.java | 37 +++++++++
.../mft/core/bufferedImpl/AbstractConnector.java | 75 +++++++++++++++++
.../mft/core/bufferedImpl/ChannelUtils.java | 87 ++++++++++++++++++++
.../mft/core/bufferedImpl/ConnectorConfig.java | 39 +++++++++
.../mft/core/bufferedImpl/ConnectorException.java | 37 +++++++++
.../airavata/mft/core/bufferedImpl/Constants.java | 30 +++++++
.../airavata/mft/core/bufferedImpl/InChannel.java | 72 +++++++++++++++++
.../airavata/mft/core/bufferedImpl/OutChannel.java | 69 ++++++++++++++++
.../mft/core/bufferedImpl/PassthroughMediator.java | 83 +++++++++++++++++++
.../org/apache/airavata/mft/transport/s3/Main.java | 90 +++++++++++++++++++++
.../airavata/mft/transport/s3/S3Constants.java | 35 ++++++++
.../airavata/mft/transport/s3/S3SinkConnector.java | 93 ++++++++++++++++++++++
.../mft/transport/s3/S3SourceConnector.java | 65 +++++++++++++++
19 files changed, 1055 insertions(+)
diff --git a/core/pom.xml b/core/pom.xml
index 1a4e07e..ef33495 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -31,6 +31,18 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>mft-core</artifactId>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>8</source>
+ <target>8</target>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
</project>
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java b/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java
new file mode 100644
index 0000000..657b017
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.api;
+
+
+import com.sun.istack.internal.Nullable;
+
+/**
+ * This represents the callback of {@link Mediator}
+ * which indicates the completion state of mediation
+ */
+@FunctionalInterface
+public interface CompletionCallback {
+
+ /**
+ * Implementation of this should contain the logic
+ * for execution after mediation returns
+ * @param message
+ * @param error
+ */
+ void onComplete(String message, @Nullable Exception error);
+
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
new file mode 100644
index 0000000..69be278
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.api;
+
+import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
+
+import java.nio.channels.Channel;
+import java.util.Properties;
+
+/**
+ * This represents a connection between external source or sink
+ */
+public interface Connector {
+
+ /**
+ * Initiates the connector object
+ * @param connectorConfig
+ * @return initation state whether success or not
+ */
+ boolean initiate(ConnectorConfig connectorConfig);
+
+ /**
+ * This returns a {@link ConnectorChannel}
+ * @return Channel
+ */
+ ConnectorChannel openChannel(Properties properties) throws Exception;
+
+ /**
+ * This is used to close the channel and release resources related to channel
+ * @param channel
+ * @throws Exception
+ */
+ void closeChannel(Channel channel) throws Exception;
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java b/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
new file mode 100644
index 0000000..d6065b9
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.api;
+
+import java.io.IOException;
+import java.nio.channels.Channel;
+
+/**
+ * An interface represents the underlying channel of a {@Link Connector}
+ */
+public interface ConnectorChannel {
+ /**
+ * get the channel of given connector
+ *
+ * @return
+ */
+ Channel getChannel();
+
+ /**
+ * close the channel if open else return
+ *
+ * @throws IOException
+ */
+ void closeChannel() throws IOException;
+
+ /**
+ * Save channel attribute to ConnectorChannel context
+ *
+ * @param key
+ * @param value
+ */
+ void addChannelAttribute(String key, Object value);
+
+ /**
+ * Get channel attribute from ConnectorChannel context
+ *
+ * @param key
+ * @return
+ */
+ Object getChannelAttribute(String key);
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java b/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
new file mode 100644
index 0000000..e3ea081
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.api;
+
+
+/**
+ * This represent the interface for all mediators, which mediates
+ * data from source connectors and send them to sink connectors.
+ */
+public interface Mediator {
+
+ /**
+ * Mediates the content from source connector and writes the mediated content
+ * to destination connector
+ * WritableByteChannel
+ * @param src
+ * @param dst
+ * @param callback
+ */
+ void mediate(ConnectorChannel src, ConnectorChannel dst, CompletionCallback callback);
+
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
new file mode 100644
index 0000000..4d2eb65
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.api;
+
+import java.nio.channels.Channel;
+import java.util.Properties;
+
+/**
+ * This represents the output connector, where to write
+ * data from the application.
+ */
+public interface SinkConnector extends Connector{
+
+ /**
+ * provides the channel to write data to external location
+ * @return ConnectorChannel
+ */
+ ConnectorChannel openChannel(Properties properties) throws Exception;
+
+ /**
+ * Verify content upload of the given channel is completed
+ * @param channel
+ * @return true if succes else false
+ */
+ boolean verifyUpload(Channel channel);
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java
new file mode 100644
index 0000000..a6b1ac5
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.api;
+
+import java.util.Properties;
+
+
+/**
+ * This represents the input connector, where to read data
+ * from the application
+ */
+public interface SourceConnector extends Connector{
+
+ /**
+ * provides the channel to read data from external location
+ * @return ConnectorChannel
+ */
+ ConnectorChannel openChannel(Properties properties);
+
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
new file mode 100644
index 0000000..2045356
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.bufferedImpl;
+
+import org.apache.airavata.mft.core.api.Connector;
+import org.apache.airavata.mft.core.api.SinkConnector;
+import org.apache.airavata.mft.core.api.SourceConnector;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.Channel;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Common class for {@link SinkConnector} and {@link SourceConnector} implementations
+ * This overrides closeChannel method and release all resources allocated to releasing channel
+ */
+public abstract class AbstractConnector implements Connector {
+ ConcurrentHashMap<Channel, Object> channelToStreamMap = new ConcurrentHashMap<>();
+
+ @Override
+ public void closeChannel(Channel channel) throws ConnectorException {
+ OutChannel outChannel;
+ InChannel inChannel;
+
+ try {
+ if (channelToStreamMap.get(channel) instanceof OutputStream) {
+ outChannel = (OutChannel) channelToStreamMap.get(channel);
+ if (outChannel != null) {
+ outChannel.closeChannel();
+ System.out.println("Closing output stream");
+ }
+
+ } else if (channelToStreamMap.get(channel) instanceof InputStream) {
+ inChannel = (InChannel) channelToStreamMap.get(channel);
+ if (inChannel != null) {
+ inChannel.closeChannel();
+ }
+ }
+
+ } catch (IOException e) {
+ throw new ConnectorException("Error occurred while closing stream", e);
+ }
+
+ }
+
+ public void cacheChannel(Channel channel, Object obj) {
+ channelToStreamMap.put(channel, obj);
+ }
+
+
+ public Object getConnectorChannel(Channel channel) {
+ return channelToStreamMap.get(channel);
+ }
+
+
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ChannelUtils.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ChannelUtils.java
new file mode 100644
index 0000000..1172142
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ChannelUtils.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.bufferedImpl;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+
+/**
+ * Contains the utility methods for Channel Operations
+ */
+public class ChannelUtils {
+
+ /**
+ * Tranfer data from readable byte channel to FileChannel using zero-copy
+ * @param byteChannel
+ * @param to
+ * @throws IOException
+ */
+ public static void transferFrom(ReadableByteChannel byteChannel, FileChannel to) throws IOException {
+ long count;
+ long total = 0;
+ while ((count = to.transferFrom(byteChannel, total, Constants.TRANSFER_MAX_SIZE)) > 0) {
+ total = +count;
+ }
+ }
+
+ /**
+ * Transfer data from FileChannel to writable byte channel using zero-copy
+ * @param to
+ * @param from
+ * @throws IOException
+ */
+ public static void transferTo(WritableByteChannel to, FileChannel from) throws IOException {
+ long count;
+ long total = 0;
+ while ((count = from.transferTo(total, Constants.TRANSFER_MAX_SIZE, to)) > 0) {
+ total = +count;
+ }
+ }
+
+ /**
+ * Copy data from readable byte channel to writable byte channel
+ * @param src
+ * @param dest
+ * @throws IOException
+ */
+ public static void copyData(ReadableByteChannel src, WritableByteChannel dest) throws IOException {
+ final ByteBuffer buffer = ByteBuffer.allocateDirect(Constants.BUFFER_SIZE);
+ int count = 0;
+ while ((count =src.read(buffer)) != -1) {
+ // prepare the buffer to be drained
+ buffer.flip();
+ // write to the channel, may block
+ dest.write(buffer);
+ // If partial transfer, shift remainder down
+ // If buffer is empty, same as doing clear()
+ buffer.compact();
+ }
+ // EOF will leave buffer in fill state
+ buffer.flip();
+ // make sure the buffer is fully drained.
+ while (buffer.hasRemaining()) {
+ dest.write(buffer);
+ }
+
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorConfig.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorConfig.java
new file mode 100644
index 0000000..240cd2a
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorConfig.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.bufferedImpl;
+
+
+import java.util.Properties;
+
+/**
+ * This class wraps all the content related to connector
+ */
+public class ConnectorConfig {
+
+ private Properties properties;
+
+ public ConnectorConfig(Properties properties) {
+ this.properties = properties;
+ }
+
+ public String getValue(String key) {
+ return properties.getProperty(key);
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java
new file mode 100644
index 0000000..6fa1371
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.bufferedImpl;
+
+/**
+ * This class encapulate all the errors releated to Connector
+ * Configurations
+ */
+public class ConnectorException extends Exception {
+ private String msg;
+
+ public ConnectorException(String message, Throwable cause) {
+ super(message, cause);
+ this.msg = msg;
+ }
+
+ public String getMsg() {
+ return msg;
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
new file mode 100644
index 0000000..1f5b46c
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.bufferedImpl;
+
+/**
+ * Contains all the constants used in core classes.
+ */
+public interface Constants {
+
+ long TRANSFER_MAX_SIZE = (1024 * 1024);
+ int BUFFER_SIZE = 8 * 1024;
+ String CONNECTOR = "CONNECTOR";
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/InChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/InChannel.java
new file mode 100644
index 0000000..542f283
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/InChannel.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.bufferedImpl;
+
+import org.apache.airavata.mft.core.api.ConnectorChannel;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.channels.Channel;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.util.HashMap;
+
+/**
+ * A class which represents the channel of a {@Link SinkConnector}
+ */
+public class InChannel implements ConnectorChannel {
+
+ private InputStream inputStream;
+
+ private ReadableByteChannel readableByteChannel;
+
+ private HashMap<String, Object> contextAttributeMap = new HashMap<>();
+
+ public InChannel(InputStream inputStream) {
+ this.inputStream = inputStream;
+ readableByteChannel = Channels.newChannel(inputStream);
+
+ }
+
+ public InputStream getInputStream() {
+ return inputStream;
+ }
+
+
+ @Override
+ public Channel getChannel() {
+ return readableByteChannel;
+ }
+
+ @Override
+ public void closeChannel() throws IOException {
+ inputStream.close();
+ }
+
+ @Override
+ public void addChannelAttribute(String key, Object value) {
+ contextAttributeMap.put(key,value );
+ }
+
+ @Override
+ public Object getChannelAttribute(String key) {
+ return contextAttributeMap.get(key);
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/OutChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/OutChannel.java
new file mode 100644
index 0000000..5d4e3c8
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/OutChannel.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.bufferedImpl;
+
+import org.apache.airavata.mft.core.api.ConnectorChannel;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.channels.Channel;
+import java.nio.channels.Channels;
+import java.nio.channels.WritableByteChannel;
+import java.util.HashMap;
+
+/**
+ * A class which represents the channel in {@Link SinkConnector}
+ */
+public class OutChannel implements ConnectorChannel {
+
+ private OutputStream outputStream;
+
+ private WritableByteChannel writableByteChannel;
+
+ private HashMap<String, Object> contextAttributeMap = new HashMap();
+
+ public OutChannel(OutputStream outputStream) {
+ this.outputStream = outputStream;
+ writableByteChannel = Channels.newChannel(outputStream);
+
+ }
+
+ @Override
+ public Channel getChannel() {
+ return writableByteChannel;
+ }
+
+ @Override
+ public void closeChannel() throws IOException {
+ outputStream.close();
+ }
+
+ @Override
+ public void addChannelAttribute(String key, Object value) {
+ contextAttributeMap.put(key, value);
+ }
+
+ @Override
+ public Object getChannelAttribute(String key){
+ return contextAttributeMap.get(key);
+ }
+
+
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
new file mode 100644
index 0000000..36b6801
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.core.bufferedImpl;
+
+import org.apache.airavata.mft.core.api.*;
+
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+
+/**
+ * This class wire the input stream and output stream based on the Channel type.
+ * If either source or destination channels are FileChannel then wire using
+ * zero copy technique of FileChannel class.
+ * Other wise use DirectByteBuffers to copy data from source to destination.
+ */
+public class PassthroughMediator implements Mediator {
+
+ @Override
+ public void mediate(ConnectorChannel src, ConnectorChannel dst, CompletionCallback callback) {
+ ReadableByteChannel rChannel = (ReadableByteChannel) src.getChannel();
+ WritableByteChannel dChannel = (WritableByteChannel) dst.getChannel();
+ try {
+ if (rChannel instanceof FileChannel) {
+ ChannelUtils.transferTo(dChannel, (FileChannel) src);
+ } else if (dChannel instanceof FileChannel) {
+ ChannelUtils.transferFrom(rChannel, (FileChannel) dst);
+ } else {
+ ChannelUtils.copyData(rChannel, dChannel);
+ }
+
+ // dst.closeChannel();
+
+ Object obj = dst.getChannelAttribute(Constants.CONNECTOR);
+ if (obj != null && obj instanceof SinkConnector) {
+ SinkConnector connector = (SinkConnector) obj;
+ boolean success = connector.verifyUpload(dChannel);
+ if (success) {
+ callback.onComplete("Successfully uploaded", null);
+ } else {
+ String msg = "Upload Failed";
+ ConnectorException connectorException = new ConnectorException(msg, null);
+ callback.onComplete("Upload failed ", connectorException);
+ }
+ }
+
+ } catch (IOException e) {
+ String msg = "Upload Failed";
+ ConnectorException connectorException = new ConnectorException(msg, e);
+ callback.onComplete("Upload failed ", connectorException);
+ } finally {
+ Connector sourceConnector = (Connector) src.getChannelAttribute(Constants.CONNECTOR);
+ Connector sinkConnector = (Connector) dst.getChannelAttribute(Constants.CONNECTOR);
+ try {
+ sourceConnector.closeChannel(rChannel);
+ sinkConnector.closeChannel(dChannel);
+ } catch (Exception ex) {
+ String msg = "Error occurred while closing channels";
+ ConnectorException connectorException = new ConnectorException(msg, ex);
+ callback.onComplete(msg, connectorException);
+ }
+ }
+
+ }
+}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
new file mode 100644
index 0000000..42cde42
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.transport.s3;
+
+import org.apache.airavata.mft.core.api.ConnectorChannel;
+import org.apache.airavata.mft.core.bufferedImpl.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
+import org.apache.airavata.mft.core.bufferedImpl.PassthroughMediator;
+
+import java.util.Properties;
+
+/**
+ * For Testing
+ */
+public class Main {
+
+
+
+ public static void main(String[] args) {
+
+ Main main = new Main();
+ main.execute();
+
+ }
+
+ public void execute() {
+ Properties connectorConfProp = new Properties();
+ connectorConfProp.put(S3Constants.ACCESS_KEY, "AKIA2SKSIFUH7QDAROSR");
+ connectorConfProp.put(S3Constants.SECRET_KEY, "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");
+ connectorConfProp.put(S3Constants.REGION, "us-east-2");
+
+ Properties srcProp = new Properties();
+ srcProp.put(S3Constants.BUCKET, "blimpit-test");
+ srcProp.put(S3Constants.REMOTE_FILE, "test.pdf");
+
+ Properties dstProp = new Properties();
+ dstProp.put(S3Constants.BUCKET, "blimpit-test");
+ dstProp.put(S3Constants.REMOTE_FILE, "meemure.pdf");
+
+ try {
+
+ ConnectorConfig connectorConfig = new ConnectorConfig(connectorConfProp);
+
+ AbstractConnector s3SourceConnector = new S3SourceConnector();
+ s3SourceConnector.initiate(connectorConfig);
+ ConnectorChannel srcChannel = s3SourceConnector.openChannel(srcProp);
+
+ AbstractConnector s3SinkConnector = new S3SinkConnector();
+ s3SinkConnector.initiate(connectorConfig);
+ ConnectorChannel dstChannel = s3SinkConnector.openChannel(dstProp);
+
+ PassthroughMediator passthroughMediator = new PassthroughMediator();
+
+
+ // Runnable r = () -> {
+ passthroughMediator.mediate(srcChannel, dstChannel, (message, error) -> {
+ try {
+ System.out.println(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ });
+ // };
+
+
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+
+}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
new file mode 100644
index 0000000..8058ca0
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.transport.s3;
+
+/**
+ * Includes constants related to S3 SDK
+ */
+public interface S3Constants {
+
+ String ACCESS_KEY="ACCESS_KEY";
+ String SECRET_KEY="SECRET_KEY";
+ String BUCKET="BUCKET";
+ String REGION="REGION";
+ String REMOTE_FILE="REMOTE_FILE";
+ int CONNECTION_EXPIRE_TIME = 1000 * 60 * 60;
+ String HTTP_CONNECTION = "HTTP_CONNECTION";
+ int HTTP_SUCCESS_RESPONSE_CODE = 200;
+}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
new file mode 100644
index 0000000..c2cb503
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.transport.s3;
+
+import com.amazonaws.HttpMethod;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest;
+import org.apache.airavata.mft.core.api.ConnectorChannel;
+import org.apache.airavata.mft.core.api.SinkConnector;
+import org.apache.airavata.mft.core.bufferedImpl.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
+import org.apache.airavata.mft.core.bufferedImpl.Constants;
+import org.apache.airavata.mft.core.bufferedImpl.OutChannel;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.channels.Channel;
+import java.util.Properties;
+
+/**
+ * Connector class which connects to a given S3 destination and provides
+ * S3 output Stream.
+ */
+public class S3SinkConnector extends AbstractConnector implements SinkConnector {
+ private AmazonS3 s3Client;
+
+
+ @Override
+ public boolean initiate(ConnectorConfig connectorConfig) {
+ s3Client = S3TransportUtil.getS3Client(connectorConfig.getValue(S3Constants.ACCESS_KEY),
+ connectorConfig.getValue(S3Constants.SECRET_KEY), connectorConfig.getValue(S3Constants.REGION));
+ return true;
+ }
+
+ @Override
+ public ConnectorChannel openChannel(Properties properties) throws IOException {
+ java.util.Date expiration = new java.util.Date();
+ long expTimeMillis = expiration.getTime();
+ expTimeMillis += S3Constants.CONNECTION_EXPIRE_TIME;
+ expiration.setTime(expTimeMillis);
+
+ GeneratePresignedUrlRequest generatePresignedUrlRequest = new GeneratePresignedUrlRequest
+ (properties.getProperty(S3Constants.BUCKET), properties.getProperty(S3Constants.REMOTE_FILE))
+ .withMethod(HttpMethod.PUT)
+ .withExpiration(expiration);
+ URL url = s3Client.generatePresignedUrl(generatePresignedUrlRequest);
+ HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+ connection.setDoOutput(true);
+ connection.setRequestMethod("PUT");
+ OutputStream outputStream = connection.getOutputStream();
+ OutChannel outChannel = new OutChannel(outputStream);
+ outChannel.addChannelAttribute(S3Constants.HTTP_CONNECTION, connection);
+ outChannel.addChannelAttribute(Constants.CONNECTOR, this);
+ cacheChannel(outChannel.getChannel(), outChannel);
+ return outChannel;
+ }
+
+ @Override
+ public boolean verifyUpload(Channel channel) {
+ OutChannel outChannel = (OutChannel) getConnectorChannel(channel);
+ HttpURLConnection connection = (HttpURLConnection) outChannel.getChannelAttribute(S3Constants.HTTP_CONNECTION);
+ try {
+ if (connection.getResponseCode() == S3Constants.HTTP_SUCCESS_RESPONSE_CODE) {
+ return true;
+ } else {
+ return false;
+ }
+ } catch (Exception error) {
+ return false;
+ }
+ }
+
+
+}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
new file mode 100644
index 0000000..ce66c0c
--- /dev/null
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.transport.s3;
+
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.S3Object;
+import org.apache.airavata.mft.core.api.ConnectorChannel;
+import org.apache.airavata.mft.core.api.SourceConnector;
+import org.apache.airavata.mft.core.bufferedImpl.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
+import org.apache.airavata.mft.core.bufferedImpl.Constants;
+import org.apache.airavata.mft.core.bufferedImpl.InChannel;
+
+import java.io.InputStream;
+import java.util.Properties;
+
+/**
+ * Connector class which connects to a given S3 source and provides
+ * S3 Input Stream.
+ */
+public class S3SourceConnector extends AbstractConnector implements SourceConnector {
+
+ private AmazonS3 s3Client;
+
+ @Override
+ public boolean initiate(ConnectorConfig connectorConfig) {
+ s3Client = S3TransportUtil.getS3Client(connectorConfig.getValue(S3Constants.ACCESS_KEY),
+ connectorConfig.getValue(S3Constants.SECRET_KEY), connectorConfig.getValue(S3Constants.REGION));
+ return true;
+ }
+
+ @Override
+ public ConnectorChannel openChannel(Properties properties) {
+ S3Object s3object = s3Client.getObject(properties.getProperty(S3Constants.BUCKET),
+ properties.getProperty(S3Constants.REMOTE_FILE));
+ InputStream inputStream;
+ if (s3object != null && s3object.getObjectContent() != null) {
+ inputStream = s3object.getObjectContent();
+ InChannel inChannel = new InChannel(inputStream);
+ inChannel.addChannelAttribute(Constants.CONNECTOR, this);
+ cacheChannel(inChannel.getChannel(), inChannel);
+ return inChannel;
+ }
+ return null;
+ }
+
+
+}
[airavata-mft] 09/22: Adding S3 stream sender
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit ce2e3ad27c985acb5bf5fabf5984ef4d9d4aff3f
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Thu Aug 8 11:10:50 2019 -0400
Adding S3 stream sender
---
.../core/streaming/DoubleByteArrayInputStream.java | 26 ++++++++++++++++++++--
.../streaming/DoubleByteArrayOutputStream.java | 11 ++++++++-
.../mft/core/streaming/TransportStream.java | 10 +--------
.../apache/airavata/mft/transport/s3/S3Sender.java | 10 +++++++++
.../airavata/mft/transport/s3/S3TransportUtil.java | 7 ++++++
.../apache/airavata/mft/transport/scp/Main.java | 16 +++++++------
.../airavata/mft/transport/scp/SCPReceiver.java | 5 ++++-
7 files changed, 65 insertions(+), 20 deletions(-)
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
index 9845e84..12323e4 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
@@ -30,6 +30,7 @@ public class DoubleByteArrayInputStream extends InputStream {
private DoubleByteArrayOutputStream outputStream;
private InputStream currentInputStream;
+ private int readCount;
public DoubleByteArrayInputStream(DoubleByteArrayOutputStream outputStream) {
this.outputStream = outputStream;
@@ -38,8 +39,29 @@ public class DoubleByteArrayInputStream extends InputStream {
@Override
public int read() throws IOException {
- refresh();
- return this.currentInputStream.read();
+ int data = this.currentInputStream.read();
+ if (data == -1) {
+ refresh();
+ while (data == -1) {
+ try {
+ Thread.sleep(1);
+ } catch (InterruptedException e) {
+ // ignore
+ }
+ refresh();
+ if (outputStream.isClosed()) {
+ // If the stream is closed, do another refresh
+ refresh();
+ data = this.currentInputStream.read();
+ break;
+ }
+ data = this.currentInputStream.read();
+ }
+ return data;
+ } else {
+ readCount ++;
+ }
+ return data;
}
@Override
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
index 799ba42..4aefc2e 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
@@ -38,6 +38,7 @@ public class DoubleByteArrayOutputStream extends OutputStream {
private long maxBytesPerStream = 1 * 1000 * 1000;
private long processedBytes = 0;
+ private long totalBytes = 0;
private boolean clearedNonActiveStream = false;
private boolean isClosed = false;
@@ -49,6 +50,7 @@ public class DoubleByteArrayOutputStream extends OutputStream {
}
this.currentStream.write(b);
processedBytes += 1;
+ totalBytes += 1;
}
@Override
@@ -58,6 +60,7 @@ public class DoubleByteArrayOutputStream extends OutputStream {
}
this.currentStream.write(b, off, len);
processedBytes += len;
+ totalBytes += len;
}
/**
@@ -68,7 +71,7 @@ public class DoubleByteArrayOutputStream extends OutputStream {
while (!clearedNonActiveStream) {
try {
Thread.sleep(100);
- System.out.println("Receiver waiting until non active buffer gets emptied");
+ System.out.println("Receiver waiting until non active buffer gets emptied " + processedBytes);
} catch (InterruptedException e) {
// ignore
}
@@ -98,10 +101,12 @@ public class DoubleByteArrayOutputStream extends OutputStream {
@Override
public void close() throws IOException {
if (!isClosed) {
+ System.out.println("Swapping buffers...");
swapBuffers();
//this.stream1.close();
//this.stream2.close();
}
+ System.out.println("Output stream received total bytes " + totalBytes);
isClosed = true;
}
@@ -118,4 +123,8 @@ public class DoubleByteArrayOutputStream extends OutputStream {
this.clearedNonActiveStream = true;
return inputStream;
}
+
+ public boolean isClosed() {
+ return isClosed;
+ }
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
index 1f5bfd3..5956ffe 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
@@ -28,7 +28,7 @@ public class TransportStream {
private AtomicBoolean streamCompleted = new AtomicBoolean(false);
private OutputStream outputStream = new DoubleByteArrayOutputStream();
- private InputStream inputStream = new DoubleByteArrayInputStream((DoubleByteArrayOutputStream) outputStream);
+ private DoubleByteArrayInputStream inputStream = new DoubleByteArrayInputStream((DoubleByteArrayOutputStream) outputStream);
private String sourceId;
private String destId;
@@ -43,18 +43,10 @@ public class TransportStream {
return inputStream;
}
- public void setInputStream(InputStream inputStream) {
- this.inputStream = inputStream;
- }
-
public OutputStream getOutputStream() {
return outputStream;
}
- public void setOutputStream(OutputStream outputStream) {
- this.outputStream = outputStream;
- }
-
public TransportMetadata getMetadata() {
return metadata;
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
index cdeb1e5..d90a34b 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
@@ -19,12 +19,22 @@
*/
package org.apache.airavata.mft.transport.s3;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.ObjectMetadata;
import org.apache.airavata.mft.core.api.StreamedSender;
import org.apache.airavata.mft.core.streaming.TransportStream;
public class S3Sender implements StreamedSender {
@Override
public void send(TransportStream stream) throws Exception {
+ S3ResourceIdentifier resourceIdentifier = S3TransportUtil.getS3ResourceIdentifier(stream.getDestId());
+ AmazonS3 s3client = S3TransportUtil.getS3Client(resourceIdentifier.getAccessKey(),
+ resourceIdentifier.getSecretKey(), resourceIdentifier.getRegion());
+ ObjectMetadata metadata = new ObjectMetadata();
+ metadata.setContentLength(stream.getMetadata().getLength());
+
+ s3client.putObject(resourceIdentifier.getBucket(), resourceIdentifier.getRemoteFile(),
+ stream.getInputStream(), metadata);
}
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java
index ca694bb..c7d56f0 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java
@@ -38,6 +38,13 @@ public class S3TransportUtil {
identifier.setRegion("us-east-2");
identifier.setRemoteFile("file.txt");
return identifier;
+ case "4":
+ identifier.setAccessKey("key");
+ identifier.setSecretKey("secret");
+ identifier.setBucket("airavata-s3");
+ identifier.setRegion("us-east-2");
+ identifier.setRemoteFile("file2.txt");
+ return identifier;
default:
return null;
}
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
index 00fe32e..f0d63eb 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
@@ -22,6 +22,7 @@ package org.apache.airavata.mft.transport.scp;
import org.apache.airavata.mft.core.streaming.TransportMetadata;
import org.apache.airavata.mft.core.streaming.TransportStream;
import org.apache.airavata.mft.transport.s3.S3Receiver;
+import org.apache.airavata.mft.transport.s3.S3Sender;
import org.apache.airavata.mft.transport.s3.S3TransportOperator;
import java.io.IOException;
@@ -30,10 +31,10 @@ public class Main {
public static void main(final String[] arg) throws Exception {
TransportMetadata metadata = new TransportMetadata();
- //SCPTransportOperator operator = new SCPTransportOperator();
- S3TransportOperator operator = new S3TransportOperator();
- metadata.setLength(operator.getResourceSize("3"));
- final TransportStream stream = new TransportStream("3", "2", metadata);
+ SCPTransportOperator operator = new SCPTransportOperator();
+ //S3TransportOperator operator = new S3TransportOperator();
+ metadata.setLength(operator.getResourceSize("1"));
+ final TransportStream stream = new TransportStream("1", "4", metadata);
Runnable r1 = new Runnable() {
@Override
@@ -69,8 +70,8 @@ public class Main {
Runnable receiverRun = new Runnable() {
@Override
public void run() {
- //SCPReceiver receiver = new SCPReceiver();
- S3Receiver receiver = new S3Receiver();
+ SCPReceiver receiver = new SCPReceiver();
+ //S3Receiver receiver = new S3Receiver();
try {
receiver.receive(stream);
} catch (Exception e) {
@@ -82,7 +83,8 @@ public class Main {
Runnable senderRun = new Runnable() {
@Override
public void run() {
- SCPSender sender = new SCPSender();
+ //SCPSender sender = new SCPSender();
+ S3Sender sender = new S3Sender();
try {
sender.send(stream);
} catch (Exception e) {
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
index 4093fa7..d5871cb 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
@@ -38,7 +38,7 @@ public class SCPReceiver implements StreamedReceiver {
}
private void transferRemoteToStream(Session session, String from, TransportStream stream) throws JSchException, IOException {
-
+ System.out.println("Starting scp receive");
// exec 'scp -f rfile' remotely
String command = "scp -f " + from;
Channel channel = session.openChannel("exec");
@@ -121,6 +121,9 @@ public class SCPReceiver implements StreamedReceiver {
stream.setStreamCompleted(true);
channel.disconnect();
session.disconnect();
+
+ System.out.println("Completed scp receive");
+
}
public int checkAck(InputStream in) throws IOException {
[airavata-mft] 04/22: Adding license headers
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 8b52f31a7e89381d5012c9630bade125233fda1a
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Tue Aug 6 02:53:33 2019 -0400
Adding license headers
---
core/pom.xml | 21 +++++++++++++++++++++
.../airavata/mft/core/api/StreamedReceiver.java | 19 +++++++++++++++++++
.../airavata/mft/core/api/StreamedSender.java | 19 +++++++++++++++++++
.../core/streaming/DoubleByteArrayInputStream.java | 19 +++++++++++++++++++
.../core/streaming/DoubleByteArrayOutputStream.java | 19 +++++++++++++++++++
.../mft/core/streaming/TransportStream.java | 19 +++++++++++++++++++
transport/pom.xml | 21 +++++++++++++++++++++
transport/scp-transport/pom.xml | 21 +++++++++++++++++++++
.../org/apache/airavata/mft/transport/scp/Main.java | 19 +++++++++++++++++++
.../airavata/mft/transport/scp/SCPReceiver.java | 19 +++++++++++++++++++
.../airavata/mft/transport/scp/SCPSender.java | 19 +++++++++++++++++++
.../mft/transport/scp/SSHResourceIdentifier.java | 19 +++++++++++++++++++
12 files changed, 234 insertions(+)
diff --git a/core/pom.xml b/core/pom.xml
index c87446e..1a4e07e 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -1,4 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
index 47db51e..41ec680 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.core.api;
import org.apache.airavata.mft.core.streaming.TransportStream;
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
index 9d3986c..a5430fb 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.core.api;
import org.apache.airavata.mft.core.streaming.TransportStream;
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
index 56f5dc5..6228d93 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.core.streaming;
import java.io.IOException;
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
index 416a8eb..f6919cf 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.core.streaming;
import java.io.*;
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
index d8a95b6..35956de 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.core.streaming;
import java.io.*;
diff --git a/transport/pom.xml b/transport/pom.xml
index 5e26abd..3bcbefa 100644
--- a/transport/pom.xml
+++ b/transport/pom.xml
@@ -1,4 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
diff --git a/transport/scp-transport/pom.xml b/transport/scp-transport/pom.xml
index f48bd0d..eccfdca 100644
--- a/transport/scp-transport/pom.xml
+++ b/transport/scp-transport/pom.xml
@@ -1,4 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
index 33fdb9f..2047cd2 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.transport.scp;
import com.jcraft.jsch.JSchException;
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
index bba84b8..61307e2 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.transport.scp;
import com.jcraft.jsch.*;
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
index 64e2210..d0a99f5 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.transport.scp;
import org.apache.airavata.mft.core.api.StreamedSender;
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java
index 6335b7c..571b0ab 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java
@@ -1,3 +1,22 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
package org.apache.airavata.mft.transport.scp;
public class SSHResourceIdentifier {
[airavata-mft] 21/22: TCP transport implementation
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit e331ff7ec3c66eab6ac5b6c1b7a7c16458ce6e5c
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Thu Aug 22 11:19:32 2019 +0530
TCP transport implementation
---
.../main/java/org/apache/airavata/mft/transport/tcp/Constants.java | 4 ++--
.../java/org/apache/airavata/mft/transport/tcp/server/FileServer.java | 2 ++
2 files changed, 4 insertions(+), 2 deletions(-)
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
index 274f8af..715eb6c 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
@@ -25,6 +25,6 @@ package org.apache.airavata.mft.transport.tcp;
public final class Constants {
public static final String UNKNOWN_HOST = "UNKNOWN_HOST";
- public static int MAX_CACHING_CONNECTIONS = 100;
- public static int MAX_NUM_OF_HOSTS = 100;
+ public static final int MAX_CACHING_CONNECTIONS = 100;
+ public static final int MAX_NUM_OF_HOSTS = 100;
}
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
index e1f7bcb..2089e69 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
@@ -49,6 +49,8 @@ public class FileServer {
//run while loop without waiting for connections
serverSocketChannel.configureBlocking(false);
+ serverSocketChannel.bind(new InetSocketAddress(port));
+
SocketChannel channel = serverSocketChannel.accept();
if (channel != null) {
[airavata-mft] 15/22: Adding local transport
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 5c0a238a2ea3a38131eab8db06ebe110d456d3a0
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Mon Aug 19 12:59:22 2019 +0530
Adding local transport
---
.../airavata/mft/core/api/ConnectorChannel.java | 6 +++
.../airavata/mft/core/bufferedImpl/Constants.java | 7 ++-
.../{InChannel.java => AbstractChannel.java} | 36 ++++---------
.../bufferedImpl/channel/AbstractConnector.java | 1 +
.../mft/core/bufferedImpl/channel/InChannel.java | 17 ++----
.../mft/core/bufferedImpl/channel/OutChannel.java | 20 ++------
.../mediation/PassthroughMediator.java | 7 ++-
pom.xml | 1 +
transport/local-transport/pom.xml | 42 +++++++++++++++
.../mft/transport/local/FileConstants.java | 11 ++--
.../airavata/mft/transport/local/FileUtils.java | 50 ++++++++----------
.../mft/transport/local/ResourceIdentifier.java | 19 ++++---
.../mft/transport/local/SinkFileConnector.java | 60 ++++++++++++++++++++++
.../mft/transport/local/SourceFileConnector.java | 33 ++++++------
transport/pom.xml | 1 +
transport/s3-transport/pom.xml | 12 +++++
.../airavata/mft/transport/s3/S3Constants.java | 13 ++---
.../airavata/mft/transport/s3/S3SinkConnector.java | 3 +-
.../mft/transport/s3/S3SourceConnector.java | 4 +-
19 files changed, 211 insertions(+), 132 deletions(-)
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java b/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
index 5e037b1..5818ee8 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
@@ -55,4 +55,10 @@ public interface ConnectorChannel {
* @return
*/
Object getChannelAttribute(String key);
+
+ /**
+ * provides the initiated connector
+ * @return Connector
+ */
+ Connector getSourceConnector();
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
index 93ee219..ff7172d 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
@@ -22,9 +22,8 @@ package org.apache.airavata.mft.core.bufferedImpl;
/**
* Contains all the constants used in core classes.
*/
-public class Constants {
+public final class Constants {
- public static long TRANSFER_MAX_SIZE = (1024 * 1024);
- public static int BUFFER_SIZE = 8 * 1024;
- public static String CONNECTOR = "CONNECTOR";
+ public static final long TRANSFER_MAX_SIZE = (1024 * 1024);
+ public static final int BUFFER_SIZE = 8 * 1024;
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java
similarity index 65%
copy from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
copy to core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java
index 5e2ab4c..0c2c259 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java
@@ -19,41 +19,22 @@
package org.apache.airavata.mft.core.bufferedImpl.channel;
+import org.apache.airavata.mft.core.api.Connector;
import org.apache.airavata.mft.core.api.ConnectorChannel;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.channels.Channel;
-import java.nio.channels.Channels;
-import java.nio.channels.ReadableByteChannel;
import java.util.HashMap;
/**
- * A class which represents the channel of a {@Link SinkConnector}
+ * A class to implement common method of In and Out Channels
*/
-public class InChannel implements ConnectorChannel {
-
- private InputStream inputStream;
-
- private ReadableByteChannel readableByteChannel;
+public abstract class AbstractChannel implements ConnectorChannel {
private HashMap<String, Object> contextAttributeMap = new HashMap<>();
- public InChannel(InputStream inputStream) {
- this.inputStream = inputStream;
- readableByteChannel = Channels.newChannel(inputStream);
-
- }
-
-
- @Override
- public Channel getChannel() {
- return readableByteChannel;
- }
+ private Connector myConnector;
- @Override
- public void closeChannel() throws IOException {
- inputStream.close();
+ public AbstractChannel(Connector myConnector) {
+ this.myConnector = myConnector;
}
@Override
@@ -65,4 +46,9 @@ public class InChannel implements ConnectorChannel {
public Object getChannelAttribute(String key) {
return contextAttributeMap.get(key);
}
+
+ @Override
+ public Connector getSourceConnector() {
+ return myConnector;
+ }
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
index de119a5..b0916dc 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
@@ -46,4 +46,5 @@ public abstract class AbstractConnector implements Connector {
}
+
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
index 5e2ab4c..4b7803b 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
@@ -19,27 +19,26 @@
package org.apache.airavata.mft.core.bufferedImpl.channel;
-import org.apache.airavata.mft.core.api.ConnectorChannel;
+import org.apache.airavata.mft.core.api.Connector;
import java.io.IOException;
import java.io.InputStream;
import java.nio.channels.Channel;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
-import java.util.HashMap;
/**
* A class which represents the channel of a {@Link SinkConnector}
*/
-public class InChannel implements ConnectorChannel {
+public class InChannel extends AbstractChannel {
private InputStream inputStream;
private ReadableByteChannel readableByteChannel;
- private HashMap<String, Object> contextAttributeMap = new HashMap<>();
- public InChannel(InputStream inputStream) {
+ public InChannel(InputStream inputStream, Connector sourceConnector) {
+ super(sourceConnector);
this.inputStream = inputStream;
readableByteChannel = Channels.newChannel(inputStream);
@@ -56,13 +55,5 @@ public class InChannel implements ConnectorChannel {
inputStream.close();
}
- @Override
- public void addChannelAttribute(String key, Object value) {
- contextAttributeMap.put(key, value);
- }
- @Override
- public Object getChannelAttribute(String key) {
- return contextAttributeMap.get(key);
- }
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
index 1df09c3..2af77d8 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
@@ -19,30 +19,28 @@
package org.apache.airavata.mft.core.bufferedImpl.channel;
-import org.apache.airavata.mft.core.api.ConnectorChannel;
+import org.apache.airavata.mft.core.api.Connector;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.channels.Channel;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
-import java.util.HashMap;
/**
* A class which represents the channel in {@Link SinkConnector}
*/
-public class OutChannel implements ConnectorChannel {
+public class OutChannel extends AbstractChannel {
private OutputStream outputStream;
private WritableByteChannel writableByteChannel;
- private HashMap<String, Object> contextAttributeMap = new HashMap();
- public OutChannel(OutputStream outputStream) {
+ public OutChannel(OutputStream outputStream, Connector sourceConnector) {
+ super(sourceConnector);
this.outputStream = outputStream;
writableByteChannel = Channels.newChannel(outputStream);
-
}
@Override
@@ -55,15 +53,5 @@ public class OutChannel implements ConnectorChannel {
outputStream.close();
}
- @Override
- public void addChannelAttribute(String key, Object value) {
- contextAttributeMap.put(key, value);
- }
-
- @Override
- public Object getChannelAttribute(String key) {
- return contextAttributeMap.get(key);
- }
-
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
index f6e2044..b9b10de 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
@@ -21,7 +21,6 @@ package org.apache.airavata.mft.core.bufferedImpl.mediation;
import org.apache.airavata.mft.core.api.*;
import org.apache.airavata.mft.core.bufferedImpl.ConnectorException;
-import org.apache.airavata.mft.core.bufferedImpl.Constants;
import org.apache.airavata.mft.core.bufferedImpl.channel.ChannelUtils;
import java.io.IOException;
@@ -51,7 +50,7 @@ public class PassthroughMediator implements Mediator {
}
- Object obj = dst.getChannelAttribute(Constants.CONNECTOR);
+ Object obj = dst.getSourceConnector();
if (obj != null && obj instanceof SinkConnector) {
SinkConnector connector = (SinkConnector) obj;
boolean success = connector.verifyUpload(dst);
@@ -69,8 +68,8 @@ public class PassthroughMediator implements Mediator {
ConnectorException connectorException = new ConnectorException(msg, e);
callback.onComplete("Upload failed ", connectorException);
} finally {
- Connector sourceConnector = (Connector) src.getChannelAttribute(Constants.CONNECTOR);
- Connector sinkConnector = (Connector) dst.getChannelAttribute(Constants.CONNECTOR);
+ Connector sourceConnector = src.getSourceConnector();
+ Connector sinkConnector = dst.getSourceConnector();
try {
sourceConnector.closeChannel(src);
sinkConnector.closeChannel(dst);
diff --git a/pom.xml b/pom.xml
index 983b1cc..1a461a9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -31,6 +31,7 @@
<modules>
<module>core</module>
<module>transport</module>
+ <module>transport/local-transport</module>
</modules>
<parent>
diff --git a/transport/local-transport/pom.xml b/transport/local-transport/pom.xml
new file mode 100644
index 0000000..850fc0b
--- /dev/null
+++ b/transport/local-transport/pom.xml
@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>airavata-mft</artifactId>
+ <groupId>org.apache.airavata</groupId>
+ <version>0.18-SNAPSHOT</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>mft-local-transport</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.airavata</groupId>
+ <artifactId>mft-core</artifactId>
+ <version>0.18-SNAPSHOT</version>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
similarity index 74%
copy from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
copy to transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
index 93ee219..424940a 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
@@ -17,14 +17,13 @@
* under the License.
*/
-package org.apache.airavata.mft.core.bufferedImpl;
+package org.apache.airavata.mft.transport.local;
/**
- * Contains all the constants used in core classes.
+ * A class for represent local file transport constants
*/
-public class Constants {
+public final class FileConstants {
- public static long TRANSFER_MAX_SIZE = (1024 * 1024);
- public static int BUFFER_SIZE = 8 * 1024;
- public static String CONNECTOR = "CONNECTOR";
+
+ public static final String OUTPUT_FILE = "OUT_FILE";
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileUtils.java
similarity index 53%
copy from core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
copy to transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileUtils.java
index 5e037b1..99a6732 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileUtils.java
@@ -17,42 +17,36 @@
* under the License.
*/
-package org.apache.airavata.mft.core.api;
+package org.apache.airavata.mft.transport.local;
-import java.io.IOException;
-import java.nio.channels.Channel;
+import org.apache.airavata.mft.core.bufferedImpl.ConnectorException;
+
+import java.io.File;
/**
- * An interface represents a channel of a {@Link Connector}
+ * A class having utility methods of local file manipulation
*/
-public interface ConnectorChannel {
+public class FileUtils {
+
/**
- * get the NIO channel of ChannelConnector
- *
+ * Validates and provides the file
+ * @param src
* @return
+ * @throws ConnectorException
*/
- Channel getChannel();
+ public static File getFile(String src) throws ConnectorException {
+ if (src != null) {
+ final File file = new File(src);
+ if (file.isFile()) {
+ return file;
+ } else {
+ throw new ConnectorException("File " + src + " is not a valid file", null);
+ }
+ } else {
+ throw new ConnectorException("Please specify validate file name", null);
+ }
- /**
- * close the channel if open else return
- *
- * @throws IOException
- */
- void closeChannel() throws IOException;
+ }
- /**
- * Save channel attribute to ConnectorChannel context
- *
- * @param key
- * @param value
- */
- void addChannelAttribute(String key, Object value);
- /**
- * Get channel attribute from ConnectorChannel context
- *
- * @param key
- * @return
- */
- Object getChannelAttribute(String key);
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/ResourceIdentifier.java
similarity index 70%
copy from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
copy to transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/ResourceIdentifier.java
index 93ee219..cac3b94 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/ResourceIdentifier.java
@@ -17,14 +17,21 @@
* under the License.
*/
-package org.apache.airavata.mft.core.bufferedImpl;
+package org.apache.airavata.mft.transport.local;
/**
- * Contains all the constants used in core classes.
+ * A class used to identify the local file path. Path should be a
+ * same node local path
*/
-public class Constants {
+public class ResourceIdentifier {
+ private String filePath;
+
+ public void setFilePath(String filePath) {
+ this.filePath = filePath;
+ }
+
+ public String getFilePath() {
+ return filePath;
+ }
- public static long TRANSFER_MAX_SIZE = (1024 * 1024);
- public static int BUFFER_SIZE = 8 * 1024;
- public static String CONNECTOR = "CONNECTOR";
}
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SinkFileConnector.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SinkFileConnector.java
new file mode 100644
index 0000000..ab94295
--- /dev/null
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SinkFileConnector.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airavata.mft.transport.local;
+
+import org.apache.airavata.mft.core.api.ConnectorChannel;
+import org.apache.airavata.mft.core.api.SinkConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.OutChannel;
+
+import java.io.File;
+import java.io.FileOutputStream;
+
+/**
+ * A class responsible for open up the connection for write to the given
+ * local file
+ */
+public class SinkFileConnector extends AbstractConnector implements SinkConnector {
+ private ResourceIdentifier resourceIdentifier;
+
+ public SinkFileConnector(ResourceIdentifier resourceIdentifier) {
+ this.resourceIdentifier = resourceIdentifier;
+ }
+
+ @Override
+ public boolean verifyUpload(ConnectorChannel channel) {
+ Object obj = channel.getChannelAttribute(FileConstants.OUTPUT_FILE);
+ if (obj instanceof File) {
+ File file = (File) obj;
+ return file.exists();
+ }
+ return false;
+ }
+
+ @Override
+ public ConnectorChannel openChannel() throws Exception {
+ File file = FileUtils.getFile(resourceIdentifier.getFilePath());
+ FileOutputStream outputStream = new FileOutputStream(file);
+ OutChannel outChannel = new OutChannel(outputStream, this);
+ outChannel.addChannelAttribute(FileConstants.OUTPUT_FILE, file);
+ return outChannel;
+ }
+
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SourceFileConnector.java
similarity index 54%
copy from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
copy to transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SourceFileConnector.java
index de119a5..6a2e1fc 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SourceFileConnector.java
@@ -17,33 +17,34 @@
* under the License.
*/
-package org.apache.airavata.mft.core.bufferedImpl.channel;
+package org.apache.airavata.mft.transport.local;
-import org.apache.airavata.mft.core.api.Connector;
import org.apache.airavata.mft.core.api.ConnectorChannel;
-import org.apache.airavata.mft.core.api.SinkConnector;
import org.apache.airavata.mft.core.api.SourceConnector;
-import org.apache.airavata.mft.core.bufferedImpl.ConnectorException;
+import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.InChannel;
-import java.io.IOException;
+import java.io.File;
+import java.io.FileInputStream;
/**
- * Common class for {@link SinkConnector} and {@link SourceConnector} implementations
- * This overrides closeChannel method and release all resources allocated to releasing channel
+ * A class responsible for open up the connection to read from file
*/
-public abstract class AbstractConnector implements Connector {
+public class SourceFileConnector extends AbstractConnector implements SourceConnector {
- @Override
- public void closeChannel(ConnectorChannel channel) throws ConnectorException {
-
- try {
- channel.closeChannel();
+ private ResourceIdentifier resourceIdentifier;
- } catch (IOException e) {
- throw new ConnectorException("Error occurred while closing stream", e);
- }
+ public SourceFileConnector(ResourceIdentifier resourceIdentifier) {
+ this.resourceIdentifier = resourceIdentifier;
+ }
+ @Override
+ public ConnectorChannel openChannel() throws Exception {
+ File file = FileUtils.getFile(resourceIdentifier.getFilePath());
+ FileInputStream inputStream = new FileInputStream(file);
+ return new InChannel(inputStream, this);
}
+
}
diff --git a/transport/pom.xml b/transport/pom.xml
index 0d4952a..2e93966 100644
--- a/transport/pom.xml
+++ b/transport/pom.xml
@@ -35,6 +35,7 @@
<modules>
<module>scp-transport</module>
<module>s3-transport</module>
+ <module>local-transport</module>
</modules>
diff --git a/transport/s3-transport/pom.xml b/transport/s3-transport/pom.xml
index 9395db8..8f50b85 100644
--- a/transport/s3-transport/pom.xml
+++ b/transport/s3-transport/pom.xml
@@ -31,6 +31,18 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>mft-s3-transport</artifactId>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>8</source>
+ <target>8</target>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
<dependencies>
<dependency>
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
index 9103319..5e7aec0 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
@@ -22,14 +22,9 @@ package org.apache.airavata.mft.transport.s3;
/**
* Includes constants related to S3 SDK
*/
-public class S3Constants {
+public final class S3Constants {
- public static String ACCESS_KEY = "ACCESS_KEY";
- public static String SECRET_KEY = "SECRET_KEY";
- public static String BUCKET = "BUCKET";
- public static String REGION = "REGION";
- public static String REMOTE_FILE = "REMOTE_FILE";
- public static int CONNECTION_EXPIRE_TIME = 1000 * 60 * 60;
- public static String HTTP_CONNECTION = "HTTP_CONNECTION";
- public static int HTTP_SUCCESS_RESPONSE_CODE = 200;
+ public static final int CONNECTION_EXPIRE_TIME = 1000 * 60 * 60;
+ public static final String HTTP_CONNECTION = "HTTP_CONNECTION";
+ public static final int HTTP_SUCCESS_RESPONSE_CODE = 200;
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
index 2434332..b6cc3fd 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
@@ -63,9 +63,8 @@ public class S3SinkConnector extends AbstractConnector implements SinkConnector
connection.setDoOutput(true);
connection.setRequestMethod("PUT");
OutputStream outputStream = connection.getOutputStream();
- OutChannel outChannel = new OutChannel(outputStream);
+ OutChannel outChannel = new OutChannel(outputStream,this);
outChannel.addChannelAttribute(S3Constants.HTTP_CONNECTION, connection);
- outChannel.addChannelAttribute(Constants.CONNECTOR, this);
return outChannel;
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
index 8f486be..c1dc2e9 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
@@ -23,7 +23,6 @@ import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.S3Object;
import org.apache.airavata.mft.core.api.ConnectorChannel;
import org.apache.airavata.mft.core.api.SourceConnector;
-import org.apache.airavata.mft.core.bufferedImpl.Constants;
import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
import org.apache.airavata.mft.core.bufferedImpl.channel.InChannel;
@@ -52,8 +51,7 @@ public class S3SourceConnector extends AbstractConnector implements SourceConnec
InputStream inputStream;
if (s3object != null && s3object.getObjectContent() != null) {
inputStream = s3object.getObjectContent();
- InChannel inChannel = new InChannel(inputStream);
- inChannel.addChannelAttribute(Constants.CONNECTOR, this);
+ InChannel inChannel = new InChannel(inputStream,this);
return inChannel;
}
return null;
[airavata-mft] 20/22: TCP transport implementation
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 1635fe53bf1c5263c76ca2259e787d0fe394f80a
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Thu Aug 22 11:14:57 2019 +0530
TCP transport implementation
---
.../apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java | 3 ---
.../org/apache/airavata/mft/transport/tcp/server/FileServer.java | 5 -----
2 files changed, 8 deletions(-)
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
index 7877a66..7fddfaf 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
@@ -28,13 +28,11 @@ public class RemoteResourceIdentifier {
private String host;
private int port;
-
private InetSocketAddress inetSocketAddress;
public RemoteResourceIdentifier(String host, int port) {
this.host = host;
this.port = port;
-
inetSocketAddress = new InetSocketAddress(host, port);
}
@@ -56,7 +54,6 @@ public class RemoteResourceIdentifier {
/**
* provides the InetSocketAddress of given host and port
- *
* @return InetSocketAddress
*/
public InetSocketAddress getAddress() {
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
index f41f673..e1f7bcb 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
@@ -35,7 +35,6 @@ public class FileServer {
private ConcurrentHashMap<String, FileServerConnector> remoteHostToConnectorMap =
new ConcurrentHashMap<>(Constants.MAX_NUM_OF_HOSTS);
-
private volatile boolean runServer = true;
/**
@@ -65,8 +64,6 @@ public class FileServer {
}
fileServerConnector.addChannel(channel);
-
-
}
}
} catch (Exception ex) {
@@ -76,7 +73,6 @@ public class FileServer {
/**
* Server is shutdown and stops accepting incoming connections
- *
* @return
*/
public void stop() {
@@ -86,7 +82,6 @@ public class FileServer {
/**
* Get connector bound to remote client. If remote host is not known keep it null.
- *
* @param host
* @return
*/
[airavata-mft] 18/22: TCP transport implementation
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 4d773e46f81933825f7f1c0843daa49410b8c808
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Thu Aug 22 11:08:25 2019 +0530
TCP transport implementation
---
.../{server/ServerConfig.java => Constants.java} | 31 ++--------
.../transport/tcp/RemoteResourceIdentifier.java | 24 +++++++-
.../tcp/client/RemoteFileServerConnector.java | 18 +++++-
.../mft/transport/tcp/server/FileServer.java | 68 +++++++++++++++++++++-
.../transport/tcp/server/FileServerConnector.java | 22 ++++++-
5 files changed, 131 insertions(+), 32 deletions(-)
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/ServerConfig.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
similarity index 60%
rename from transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/ServerConfig.java
rename to transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
index 5264c86..ba07517 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/ServerConfig.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
@@ -17,34 +17,13 @@
* under the License.
*/
-package org.apache.airavata.mft.transport.tcp.server;
+package org.apache.airavata.mft.transport.tcp;
/**
- * A class which represents the server configurations
+ * A class which contains all Constants
*/
-public class ServerConfig {
+public final class Constants {
- private String host;
- private int port;
-
- public ServerConfig(String host, int port) {
- this.host = host;
- this.port = port;
- }
-
- public String getHost() {
- return host;
- }
-
- public void setHost(String host) {
- this.host = host;
- }
-
- public int getPort() {
- return port;
- }
-
- public void setPort(int port) {
- this.port = port;
- }
+ public static final String UNKNOWN_HOST = "UNKNOWN_HOST";
+ public static int MAX_CACHING_CONNECTIONS = 100;
}
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
index d02a29f..7877a66 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
@@ -19,13 +19,24 @@
package org.apache.airavata.mft.transport.tcp;
+import java.net.InetSocketAddress;
+
/**
- * A class used to identify remote servers
+ * A class used to identify remote clients and servers
*/
public class RemoteResourceIdentifier {
private String host;
- private int port;
+ private int port;
+
+ private InetSocketAddress inetSocketAddress;
+
+ public RemoteResourceIdentifier(String host, int port) {
+ this.host = host;
+ this.port = port;
+
+ inetSocketAddress = new InetSocketAddress(host, port);
+ }
public String getHost() {
return host;
@@ -42,4 +53,13 @@ public class RemoteResourceIdentifier {
public void setPort(int port) {
this.port = port;
}
+
+ /**
+ * provides the InetSocketAddress of given host and port
+ *
+ * @return InetSocketAddress
+ */
+ public InetSocketAddress getAddress() {
+ return inetSocketAddress;
+ }
}
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
index f2252a4..b978e54 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
@@ -22,19 +22,33 @@ package org.apache.airavata.mft.transport.tcp.client;
import org.apache.airavata.mft.core.api.ConnectorChannel;
import org.apache.airavata.mft.core.api.SinkConnector;
import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.OutChannel;
+import org.apache.airavata.mft.transport.tcp.RemoteResourceIdentifier;
+
+import java.nio.channels.SocketChannel;
/**
* A class which represents the connections between remote server and local client
*/
public class RemoteFileServerConnector extends AbstractConnector implements SinkConnector {
+ RemoteResourceIdentifier identifier;
+
+ public RemoteFileServerConnector(RemoteResourceIdentifier remoteResourceIdentifier) {
+ this.identifier = remoteResourceIdentifier;
+ }
+
+
@Override
public boolean verifyUpload(ConnectorChannel channel) {
- return false;
+ //TODO Implement this later
+ return true;
}
@Override
public ConnectorChannel openChannel() throws Exception {
- return null;
+ SocketChannel socketChannel = SocketChannel.open(this.identifier.getAddress());
+ OutChannel outChannel = new OutChannel(socketChannel, this);
+ return outChannel;
}
}
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
index ef24d6c..38da89b 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
@@ -19,26 +19,92 @@
package org.apache.airavata.mft.transport.tcp.server;
+import com.sun.istack.internal.Nullable;
+import org.apache.airavata.mft.transport.tcp.Constants;
+
+import java.net.InetSocketAddress;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.ConcurrentHashMap;
+
/**
* A class which is responsible for accepting incoming tcp connections
* and delegate accepted channels to worker objects
*/
public class FileServer {
+
+ //If too many host address
+ private ConcurrentHashMap<String, FileServerConnector> remoteHostToConnectorMap = new ConcurrentHashMap<>();
+
+ private volatile boolean runServer = true;
+
/**
* This starts the server and accepts incoming connections
+ *
* @return
*/
- public void start() {
+ public void start(int port) {
+ try {
+ remoteHostToConnectorMap.put(Constants.UNKNOWN_HOST, new FileServerConnector());
+ while (runServer) {
+ final ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
+ //run while loop without waiting for connections
+ serverSocketChannel.configureBlocking(false);
+
+ SocketChannel channel = serverSocketChannel.accept();
+
+ if (channel != null) {
+ InetSocketAddress socketAddress = (InetSocketAddress) channel.getRemoteAddress();
+
+ String hostName = socketAddress.getHostName();
+
+ FileServerConnector fileServerConnector = remoteHostToConnectorMap.get(hostName);
+ if (fileServerConnector == null) {
+ fileServerConnector = new FileServerConnector();
+ remoteHostToConnectorMap.put(hostName, fileServerConnector);
+ }
+
+ fileServerConnector.addChannel(channel);
+
+
+ }
+ }
+ } catch (Exception ex) {
+ //TODO: log exception
+ }
}
/**
* Server is shutdown and stops accepting incoming connections
+ *
* @return
*/
public void stop() {
+ this.runServer = false;
+ }
+
+
+ /**
+ * Get connector bound to remote client. If remote host is not known keep it null.
+ *
+ * @param host
+ * @return
+ */
+ public FileServerConnector getConnector(@Nullable String host) {
+ FileServerConnector fileServerConnector = null;
+ if (host != null) {
+ fileServerConnector = remoteHostToConnectorMap.get(host);
+ if (fileServerConnector == null) {
+ fileServerConnector = new FileServerConnector();
+ remoteHostToConnectorMap.put(host, fileServerConnector);
+ }
+ } else {
+ fileServerConnector = remoteHostToConnectorMap.get(Constants.UNKNOWN_HOST);
+ }
+ return fileServerConnector;
}
diff --git a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
index b655fdf..0183632 100644
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
@@ -22,13 +22,33 @@ package org.apache.airavata.mft.transport.tcp.server;
import org.apache.airavata.mft.core.api.ConnectorChannel;
import org.apache.airavata.mft.core.api.SourceConnector;
import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.InChannel;
+import org.apache.airavata.mft.transport.tcp.Constants;
+
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.LinkedBlockingQueue;
/**
* Represents a connection between remote client and FileServer
*/
public class FileServerConnector extends AbstractConnector implements SourceConnector {
+
+ private LinkedBlockingQueue<SocketChannel> acceptedChannelCache = new
+ LinkedBlockingQueue<>(Constants.MAX_CACHING_CONNECTIONS);
+
+
@Override
public ConnectorChannel openChannel() throws Exception {
- return null;
+ SocketChannel socketChannel = this.acceptedChannelCache.take();
+ InChannel inChannel = new InChannel(socketChannel, this);
+ return inChannel;
+
}
+
+
+ public void addChannel(SocketChannel channel) throws InterruptedException {
+ this.acceptedChannelCache.put(channel);
+ }
+
+
}
[airavata-mft] 14/22: Improve code
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 7a951d285465238ff1093131714e3361ef54bd55
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Mon Aug 19 11:00:52 2019 +0530
Improve code
---
.../airavata/mft/core/api/CompletionCallback.java | 1 +
.../apache/airavata/mft/core/api/Connector.java | 15 ++------
.../org/apache/airavata/mft/core/api/Mediator.java | 1 +
.../airavata/mft/core/api/SinkConnector.java | 9 +----
.../airavata/mft/core/api/SourceConnector.java | 12 ++----
.../mft/core/bufferedImpl/ConnectorConfig.java | 39 -------------------
.../airavata/mft/core/bufferedImpl/Constants.java | 8 ++--
.../{ => channel}/AbstractConnector.java | 3 +-
.../bufferedImpl/{ => channel}/ChannelUtils.java | 11 ++++--
.../core/bufferedImpl/{ => channel}/InChannel.java | 8 +---
.../bufferedImpl/{ => channel}/OutChannel.java | 6 +--
.../{ => mediation}/PassthroughMediator.java | 5 ++-
.../org/apache/airavata/mft/transport/s3/Main.java | 44 ++++++++++------------
.../airavata/mft/transport/s3/S3Constants.java | 18 ++++-----
.../airavata/mft/transport/s3/S3SinkConnector.java | 21 +++++------
.../mft/transport/s3/S3SourceConnector.java | 23 ++++++-----
16 files changed, 83 insertions(+), 141 deletions(-)
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java b/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java
index 657b017..f80c398 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java
@@ -32,6 +32,7 @@ public interface CompletionCallback {
/**
* Implementation of this should contain the logic
* for execution after mediation returns
+ *
* @param message
* @param error
*/
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
index 8627403..191ccc0 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
@@ -19,32 +19,25 @@
package org.apache.airavata.mft.core.api;
-import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
-
-import java.util.Properties;
/**
* This represents a connection between external source or sink
*/
public interface Connector {
- /**
- * Initiates the connector object
- * @param connectorConfig
- * @return initation state whether success or not
- */
- boolean initiate(ConnectorConfig connectorConfig);
/**
* This returns a {@link ConnectorChannel}
+ *
* @return Channel
*/
- ConnectorChannel openChannel(Properties properties) throws Exception;
+ ConnectorChannel openChannel() throws Exception;
/**
* This is used to close the channel and release resources related to channel
+ *
* @param channel
* @throws Exception
*/
- void closeChannel(ConnectorChannel channel) throws Exception;
+ void closeChannel(ConnectorChannel channel) throws Exception;
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java b/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
index 76e1f69..92c479a 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
@@ -29,6 +29,7 @@ public interface Mediator {
/**
* Mediates the content from source connector and writes the mediated content
* to destination connector
+ *
* @param src
* @param dst
* @param callback
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
index dd02e10..da5137f 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
@@ -19,22 +19,17 @@
package org.apache.airavata.mft.core.api;
-import java.util.Properties;
/**
* This represents the output connector, where to write
* data from the application.
*/
-public interface SinkConnector extends Connector{
+public interface SinkConnector extends Connector {
- /**
- * provides the channel to write data to external location
- * @return ConnectorChannel
- */
- ConnectorChannel openChannel(Properties properties) throws Exception;
/**
* Verify content upload of the given channel is completed
+ *
* @param channel
* @return true if succes else false
*/
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java
index a6b1ac5..db9922e 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java
@@ -19,19 +19,13 @@
package org.apache.airavata.mft.core.api;
-import java.util.Properties;
-
/**
* This represents the input connector, where to read data
- * from the application
+ * from the application. Use this connector to specify reader related
+ * methods
*/
-public interface SourceConnector extends Connector{
+public interface SourceConnector extends Connector {
- /**
- * provides the channel to read data from external location
- * @return ConnectorChannel
- */
- ConnectorChannel openChannel(Properties properties);
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorConfig.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorConfig.java
deleted file mode 100644
index 240cd2a..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorConfig.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.airavata.mft.core.bufferedImpl;
-
-
-import java.util.Properties;
-
-/**
- * This class wraps all the content related to connector
- */
-public class ConnectorConfig {
-
- private Properties properties;
-
- public ConnectorConfig(Properties properties) {
- this.properties = properties;
- }
-
- public String getValue(String key) {
- return properties.getProperty(key);
- }
-}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
index 1f5b46c..93ee219 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
@@ -22,9 +22,9 @@ package org.apache.airavata.mft.core.bufferedImpl;
/**
* Contains all the constants used in core classes.
*/
-public interface Constants {
+public class Constants {
- long TRANSFER_MAX_SIZE = (1024 * 1024);
- int BUFFER_SIZE = 8 * 1024;
- String CONNECTOR = "CONNECTOR";
+ public static long TRANSFER_MAX_SIZE = (1024 * 1024);
+ public static int BUFFER_SIZE = 8 * 1024;
+ public static String CONNECTOR = "CONNECTOR";
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
similarity index 92%
rename from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
rename to core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
index 8849aac..de119a5 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/AbstractConnector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
@@ -17,12 +17,13 @@
* under the License.
*/
-package org.apache.airavata.mft.core.bufferedImpl;
+package org.apache.airavata.mft.core.bufferedImpl.channel;
import org.apache.airavata.mft.core.api.Connector;
import org.apache.airavata.mft.core.api.ConnectorChannel;
import org.apache.airavata.mft.core.api.SinkConnector;
import org.apache.airavata.mft.core.api.SourceConnector;
+import org.apache.airavata.mft.core.bufferedImpl.ConnectorException;
import java.io.IOException;
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ChannelUtils.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/ChannelUtils.java
similarity index 90%
rename from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ChannelUtils.java
rename to core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/ChannelUtils.java
index 1172142..7a402c6 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ChannelUtils.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/ChannelUtils.java
@@ -17,7 +17,9 @@
* under the License.
*/
-package org.apache.airavata.mft.core.bufferedImpl;
+package org.apache.airavata.mft.core.bufferedImpl.channel;
+
+import org.apache.airavata.mft.core.bufferedImpl.Constants;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -32,6 +34,7 @@ public class ChannelUtils {
/**
* Tranfer data from readable byte channel to FileChannel using zero-copy
+ *
* @param byteChannel
* @param to
* @throws IOException
@@ -46,6 +49,7 @@ public class ChannelUtils {
/**
* Transfer data from FileChannel to writable byte channel using zero-copy
+ *
* @param to
* @param from
* @throws IOException
@@ -60,14 +64,15 @@ public class ChannelUtils {
/**
* Copy data from readable byte channel to writable byte channel
+ *
* @param src
* @param dest
* @throws IOException
*/
- public static void copyData(ReadableByteChannel src, WritableByteChannel dest) throws IOException {
+ public static void copyData(ReadableByteChannel src, WritableByteChannel dest) throws IOException {
final ByteBuffer buffer = ByteBuffer.allocateDirect(Constants.BUFFER_SIZE);
int count = 0;
- while ((count =src.read(buffer)) != -1) {
+ while ((count = src.read(buffer)) != -1) {
// prepare the buffer to be drained
buffer.flip();
// write to the channel, may block
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/InChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
similarity index 91%
rename from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/InChannel.java
rename to core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
index 542f283..5e2ab4c 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/InChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.airavata.mft.core.bufferedImpl;
+package org.apache.airavata.mft.core.bufferedImpl.channel;
import org.apache.airavata.mft.core.api.ConnectorChannel;
@@ -45,10 +45,6 @@ public class InChannel implements ConnectorChannel {
}
- public InputStream getInputStream() {
- return inputStream;
- }
-
@Override
public Channel getChannel() {
@@ -62,7 +58,7 @@ public class InChannel implements ConnectorChannel {
@Override
public void addChannelAttribute(String key, Object value) {
- contextAttributeMap.put(key,value );
+ contextAttributeMap.put(key, value);
}
@Override
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/OutChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
similarity index 93%
rename from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/OutChannel.java
rename to core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
index 5d4e3c8..1df09c3 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/OutChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.airavata.mft.core.bufferedImpl;
+package org.apache.airavata.mft.core.bufferedImpl.channel;
import org.apache.airavata.mft.core.api.ConnectorChannel;
@@ -52,7 +52,7 @@ public class OutChannel implements ConnectorChannel {
@Override
public void closeChannel() throws IOException {
- outputStream.close();
+ outputStream.close();
}
@Override
@@ -61,7 +61,7 @@ public class OutChannel implements ConnectorChannel {
}
@Override
- public Object getChannelAttribute(String key){
+ public Object getChannelAttribute(String key) {
return contextAttributeMap.get(key);
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
similarity index 93%
rename from core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
rename to core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
index af9b663..f6e2044 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/PassthroughMediator.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
@@ -17,9 +17,12 @@
* under the License.
*/
-package org.apache.airavata.mft.core.bufferedImpl;
+package org.apache.airavata.mft.core.bufferedImpl.mediation;
import org.apache.airavata.mft.core.api.*;
+import org.apache.airavata.mft.core.bufferedImpl.ConnectorException;
+import org.apache.airavata.mft.core.bufferedImpl.Constants;
+import org.apache.airavata.mft.core.bufferedImpl.channel.ChannelUtils;
import java.io.IOException;
import java.nio.channels.FileChannel;
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
index af29276..76c5ae5 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
@@ -20,11 +20,8 @@
package org.apache.airavata.mft.transport.s3;
import org.apache.airavata.mft.core.api.ConnectorChannel;
-import org.apache.airavata.mft.core.bufferedImpl.AbstractConnector;
-import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
-import org.apache.airavata.mft.core.bufferedImpl.PassthroughMediator;
-
-import java.util.Properties;
+import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.mediation.PassthroughMediator;
/**
* For Testing
@@ -32,7 +29,6 @@ import java.util.Properties;
public class Main {
-
public static void main(String[] args) {
Main main = new Main();
@@ -41,30 +37,31 @@ public class Main {
}
public void execute() {
- Properties connectorConfProp = new Properties();
- connectorConfProp.put(S3Constants.ACCESS_KEY, "XXXX");
- connectorConfProp.put(S3Constants.SECRET_KEY, "YYYY");
- connectorConfProp.put(S3Constants.REGION, "us-east-2");
- Properties srcProp = new Properties();
- srcProp.put(S3Constants.BUCKET, "test");
- srcProp.put(S3Constants.REMOTE_FILE, "test.pdf");
+ S3ResourceIdentifier src = new S3ResourceIdentifier();
+ S3ResourceIdentifier dst = new S3ResourceIdentifier();
+
+ src.setAccessKey("XXX");
+ src.setSecretKey("YYY");
+ src.setRegion("us-east-2");
+ src.setBucket("test");
+ src.setRemoteFile("test.pdf");
+
+ dst.setAccessKey("XXX");
+ dst.setSecretKey("YYY");
+ dst.setRegion("us-east-2");
+ dst.setBucket("test");
+ dst.setRemoteFile("testA.pdf");
- Properties dstProp = new Properties();
- dstProp.put(S3Constants.BUCKET, "test");
- dstProp.put(S3Constants.REMOTE_FILE, "teemure.pdf");
try {
- ConnectorConfig connectorConfig = new ConnectorConfig(connectorConfProp);
- AbstractConnector s3SourceConnector = new S3SourceConnector();
- s3SourceConnector.initiate(connectorConfig);
- ConnectorChannel srcChannel = s3SourceConnector.openChannel(srcProp);
+ AbstractConnector s3SourceConnector = new S3SourceConnector(src);
+ ConnectorChannel srcChannel = s3SourceConnector.openChannel();
- AbstractConnector s3SinkConnector = new S3SinkConnector();
- s3SinkConnector.initiate(connectorConfig);
- ConnectorChannel dstChannel = s3SinkConnector.openChannel(dstProp);
+ AbstractConnector s3SinkConnector = new S3SinkConnector(dst);
+ ConnectorChannel dstChannel = s3SinkConnector.openChannel();
PassthroughMediator passthroughMediator = new PassthroughMediator();
@@ -80,7 +77,6 @@ public class Main {
// };
-
} catch (Exception e) {
e.printStackTrace();
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
index 8058ca0..9103319 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
@@ -22,14 +22,14 @@ package org.apache.airavata.mft.transport.s3;
/**
* Includes constants related to S3 SDK
*/
-public interface S3Constants {
+public class S3Constants {
- String ACCESS_KEY="ACCESS_KEY";
- String SECRET_KEY="SECRET_KEY";
- String BUCKET="BUCKET";
- String REGION="REGION";
- String REMOTE_FILE="REMOTE_FILE";
- int CONNECTION_EXPIRE_TIME = 1000 * 60 * 60;
- String HTTP_CONNECTION = "HTTP_CONNECTION";
- int HTTP_SUCCESS_RESPONSE_CODE = 200;
+ public static String ACCESS_KEY = "ACCESS_KEY";
+ public static String SECRET_KEY = "SECRET_KEY";
+ public static String BUCKET = "BUCKET";
+ public static String REGION = "REGION";
+ public static String REMOTE_FILE = "REMOTE_FILE";
+ public static int CONNECTION_EXPIRE_TIME = 1000 * 60 * 60;
+ public static String HTTP_CONNECTION = "HTTP_CONNECTION";
+ public static int HTTP_SUCCESS_RESPONSE_CODE = 200;
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
index 3ee2bae..2434332 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
@@ -24,16 +24,14 @@ import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest;
import org.apache.airavata.mft.core.api.ConnectorChannel;
import org.apache.airavata.mft.core.api.SinkConnector;
-import org.apache.airavata.mft.core.bufferedImpl.AbstractConnector;
-import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
import org.apache.airavata.mft.core.bufferedImpl.Constants;
-import org.apache.airavata.mft.core.bufferedImpl.OutChannel;
+import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.OutChannel;
import java.io.IOException;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
-import java.util.Properties;
/**
* Connector class which connects to a given S3 destination and provides
@@ -41,24 +39,23 @@ import java.util.Properties;
*/
public class S3SinkConnector extends AbstractConnector implements SinkConnector {
private AmazonS3 s3Client;
+ private S3ResourceIdentifier s3ResourceIdentifier;
-
- @Override
- public boolean initiate(ConnectorConfig connectorConfig) {
- s3Client = S3TransportUtil.getS3Client(connectorConfig.getValue(S3Constants.ACCESS_KEY),
- connectorConfig.getValue(S3Constants.SECRET_KEY), connectorConfig.getValue(S3Constants.REGION));
- return true;
+ public S3SinkConnector(S3ResourceIdentifier s3ResourceIdentifier) {
+ this.s3ResourceIdentifier = s3ResourceIdentifier;
+ this.s3Client = S3TransportUtil.getS3Client(s3ResourceIdentifier.getAccessKey(),
+ s3ResourceIdentifier.getSecretKey(), s3ResourceIdentifier.getRegion());
}
@Override
- public ConnectorChannel openChannel(Properties properties) throws IOException {
+ public ConnectorChannel openChannel() throws IOException {
java.util.Date expiration = new java.util.Date();
long expTimeMillis = expiration.getTime();
expTimeMillis += S3Constants.CONNECTION_EXPIRE_TIME;
expiration.setTime(expTimeMillis);
GeneratePresignedUrlRequest generatePresignedUrlRequest = new GeneratePresignedUrlRequest
- (properties.getProperty(S3Constants.BUCKET), properties.getProperty(S3Constants.REMOTE_FILE))
+ (s3ResourceIdentifier.getBucket(), s3ResourceIdentifier.getRemoteFile())
.withMethod(HttpMethod.PUT)
.withExpiration(expiration);
URL url = s3Client.generatePresignedUrl(generatePresignedUrlRequest);
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
index 6539d04..8f486be 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
@@ -23,13 +23,11 @@ import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.S3Object;
import org.apache.airavata.mft.core.api.ConnectorChannel;
import org.apache.airavata.mft.core.api.SourceConnector;
-import org.apache.airavata.mft.core.bufferedImpl.AbstractConnector;
-import org.apache.airavata.mft.core.bufferedImpl.ConnectorConfig;
import org.apache.airavata.mft.core.bufferedImpl.Constants;
-import org.apache.airavata.mft.core.bufferedImpl.InChannel;
+import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.InChannel;
import java.io.InputStream;
-import java.util.Properties;
/**
* Connector class which connects to a given S3 source and provides
@@ -39,17 +37,18 @@ public class S3SourceConnector extends AbstractConnector implements SourceConnec
private AmazonS3 s3Client;
- @Override
- public boolean initiate(ConnectorConfig connectorConfig) {
- s3Client = S3TransportUtil.getS3Client(connectorConfig.getValue(S3Constants.ACCESS_KEY),
- connectorConfig.getValue(S3Constants.SECRET_KEY), connectorConfig.getValue(S3Constants.REGION));
- return true;
+ private S3ResourceIdentifier identifier;
+
+ public S3SourceConnector(S3ResourceIdentifier identifier) {
+ this.identifier = identifier;
+ this.s3Client = S3TransportUtil.getS3Client(identifier.getAccessKey(),
+ identifier.getSecretKey(), identifier.getAccessKey());
}
@Override
- public ConnectorChannel openChannel(Properties properties) {
- S3Object s3object = s3Client.getObject(properties.getProperty(S3Constants.BUCKET),
- properties.getProperty(S3Constants.REMOTE_FILE));
+ public ConnectorChannel openChannel() {
+ S3Object s3object = s3Client.getObject(identifier.getBucket(),
+ identifier.getRemoteFile());
InputStream inputStream;
if (s3object != null && s3object.getObjectContent() != null) {
inputStream = s3object.getObjectContent();
[airavata-mft] 16/22: Change permission
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit df53974468f02907b8d06c53c6d6db729efb287d
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Mon Aug 19 13:01:31 2019 +0530
Change permission
---
LICENSE | 0
README | 0
core/pom.xml | 0
.../main/java/org/apache/airavata/mft/core/api/CompletionCallback.java | 0
core/src/main/java/org/apache/airavata/mft/core/api/Connector.java | 0
core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java | 0
core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java | 0
core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java | 0
core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java | 0
core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java | 0
core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java | 0
.../src/main/java/org/apache/airavata/mft/core/api/TransportOperator.java | 0
.../org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java | 0
.../main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java | 0
.../apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java | 0
.../apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java | 0
.../org/apache/airavata/mft/core/bufferedImpl/channel/ChannelUtils.java | 0
.../java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java | 0
.../org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java | 0
.../airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java | 0
.../apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java | 0
.../apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java | 0
.../java/org/apache/airavata/mft/core/streaming/TransportMetadata.java | 0
.../main/java/org/apache/airavata/mft/core/streaming/TransportStream.java | 0
pom.xml | 0
transport/local-transport/pom.xml | 0
.../main/java/org/apache/airavata/mft/transport/local/FileConstants.java | 0
.../src/main/java/org/apache/airavata/mft/transport/local/FileUtils.java | 0
.../java/org/apache/airavata/mft/transport/local/ResourceIdentifier.java | 0
.../java/org/apache/airavata/mft/transport/local/SinkFileConnector.java | 0
.../java/org/apache/airavata/mft/transport/local/SourceFileConnector.java | 0
transport/pom.xml | 0
transport/s3-transport/pom.xml | 0
.../src/main/java/org/apache/airavata/mft/transport/s3/Main.java | 0
.../src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java | 0
.../src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java | 0
.../java/org/apache/airavata/mft/transport/s3/S3ResourceIdentifier.java | 0
.../src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java | 0
.../main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java | 0
.../main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java | 0
.../java/org/apache/airavata/mft/transport/s3/S3TransportOperator.java | 0
.../main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java | 0
transport/scp-transport/pom.xml | 0
.../src/main/java/org/apache/airavata/mft/transport/scp/Main.java | 0
.../src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java | 0
.../src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java | 0
.../java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java | 0
.../main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java | 0
.../java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java | 0
49 files changed, 0 insertions(+), 0 deletions(-)
diff --git a/LICENSE b/LICENSE
old mode 100644
new mode 100755
diff --git a/README b/README
old mode 100644
new mode 100755
diff --git a/core/pom.xml b/core/pom.xml
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java b/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java b/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java b/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java b/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java b/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/TransportOperator.java b/core/src/main/java/org/apache/airavata/mft/core/api/TransportOperator.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/ChannelUtils.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/ChannelUtils.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportMetadata.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportMetadata.java
old mode 100644
new mode 100755
diff --git a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java b/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
old mode 100644
new mode 100755
diff --git a/pom.xml b/pom.xml
old mode 100644
new mode 100755
diff --git a/transport/local-transport/pom.xml b/transport/local-transport/pom.xml
old mode 100644
new mode 100755
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
old mode 100644
new mode 100755
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileUtils.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileUtils.java
old mode 100644
new mode 100755
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/ResourceIdentifier.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/ResourceIdentifier.java
old mode 100644
new mode 100755
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SinkFileConnector.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SinkFileConnector.java
old mode 100644
new mode 100755
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SourceFileConnector.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SourceFileConnector.java
old mode 100644
new mode 100755
diff --git a/transport/pom.xml b/transport/pom.xml
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/pom.xml b/transport/s3-transport/pom.xml
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3ResourceIdentifier.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3ResourceIdentifier.java
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportOperator.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportOperator.java
old mode 100644
new mode 100755
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportUtil.java
old mode 100644
new mode 100755
diff --git a/transport/scp-transport/pom.xml b/transport/scp-transport/pom.xml
old mode 100644
new mode 100755
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
old mode 100644
new mode 100755
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
old mode 100644
new mode 100755
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
old mode 100644
new mode 100755
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java
old mode 100644
new mode 100755
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
old mode 100644
new mode 100755
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java
old mode 100644
new mode 100755
[airavata-mft] 17/22: Add intial implementation of TCP File Server
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
commit 3b157095c060a1c610d3532c70ce4c71572f3fdd
Author: isururanawaka <ir...@gmail.com>
AuthorDate: Wed Aug 21 12:24:11 2019 +0530
Add intial implementation of TCP File Server
---
.../mft/core/bufferedImpl/channel/InChannel.java | 12 ++++++++-
.../mft/core/bufferedImpl/channel/OutChannel.java | 12 ++++++++-
pom.xml | 1 -
transport/local-transport/pom.xml | 3 +--
.../mft/transport/local/FileConstants.java | 2 +-
transport/pom.xml | 1 +
.../{local-transport => tcp-transport}/pom.xml | 5 ++--
.../transport/tcp/RemoteResourceIdentifier.java} | 24 +++++++++++++++---
.../tcp/client/RemoteFileServerConnector.java} | 19 +++++++++++---
.../mft/transport/tcp/server/FileServer.java} | 24 +++++++++++++++---
.../transport/tcp/server/FileServerConnector.java} | 17 ++++++++-----
.../mft/transport/tcp/server/ServerConfig.java} | 29 +++++++++++++++++++---
12 files changed, 118 insertions(+), 31 deletions(-)
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
index 4b7803b..20ee77d 100755
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
@@ -44,6 +44,12 @@ public class InChannel extends AbstractChannel {
}
+ public InChannel(ReadableByteChannel channel, Connector sourceConnector) {
+ super(sourceConnector);
+ readableByteChannel = channel;
+
+ }
+
@Override
public Channel getChannel() {
@@ -52,7 +58,11 @@ public class InChannel extends AbstractChannel {
@Override
public void closeChannel() throws IOException {
- inputStream.close();
+ if (inputStream != null) {
+ inputStream.close();
+ } else if (readableByteChannel != null) {
+ readableByteChannel.close();
+ }
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
index 2af77d8..0bba57a 100755
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
@@ -43,6 +43,12 @@ public class OutChannel extends AbstractChannel {
writableByteChannel = Channels.newChannel(outputStream);
}
+ public OutChannel(WritableByteChannel byteChannel, Connector sourceConnector) {
+ super(sourceConnector);
+ this.writableByteChannel = byteChannel;
+ }
+
+
@Override
public Channel getChannel() {
return writableByteChannel;
@@ -50,7 +56,11 @@ public class OutChannel extends AbstractChannel {
@Override
public void closeChannel() throws IOException {
- outputStream.close();
+ if (outputStream != null) {
+ outputStream.close();
+ } else if (writableByteChannel != null) {
+ writableByteChannel.close();
+ }
}
diff --git a/pom.xml b/pom.xml
index 1a461a9..983b1cc 100755
--- a/pom.xml
+++ b/pom.xml
@@ -31,7 +31,6 @@
<modules>
<module>core</module>
<module>transport</module>
- <module>transport/local-transport</module>
</modules>
<parent>
diff --git a/transport/local-transport/pom.xml b/transport/local-transport/pom.xml
index 850fc0b..b0c2c7d 100755
--- a/transport/local-transport/pom.xml
+++ b/transport/local-transport/pom.xml
@@ -22,10 +22,9 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>airavata-mft</artifactId>
+ <artifactId>mft-transport</artifactId>
<groupId>org.apache.airavata</groupId>
<version>0.18-SNAPSHOT</version>
- <relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
index 424940a..1d529af 100755
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
@@ -20,7 +20,7 @@
package org.apache.airavata.mft.transport.local;
/**
- * A class for represent local file transport constants
+ * A class of represent local file transport constants
*/
public final class FileConstants {
diff --git a/transport/pom.xml b/transport/pom.xml
index 2e93966..90bc015 100755
--- a/transport/pom.xml
+++ b/transport/pom.xml
@@ -36,6 +36,7 @@
<module>scp-transport</module>
<module>s3-transport</module>
<module>local-transport</module>
+ <module>tcp-transport</module>
</modules>
diff --git a/transport/local-transport/pom.xml b/transport/tcp-transport/pom.xml
old mode 100755
new mode 100644
similarity index 90%
copy from transport/local-transport/pom.xml
copy to transport/tcp-transport/pom.xml
index 850fc0b..b17cc41
--- a/transport/local-transport/pom.xml
+++ b/transport/tcp-transport/pom.xml
@@ -22,14 +22,13 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>airavata-mft</artifactId>
+ <artifactId>mft-transport</artifactId>
<groupId>org.apache.airavata</groupId>
<version>0.18-SNAPSHOT</version>
- <relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>mft-local-transport</artifactId>
+ <artifactId>mft-tcp-transport</artifactId>
<dependencies>
<dependency>
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
old mode 100755
new mode 100644
similarity index 66%
copy from transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
copy to transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
index 424940a..d02a29f
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
@@ -17,13 +17,29 @@
* under the License.
*/
-package org.apache.airavata.mft.transport.local;
+package org.apache.airavata.mft.transport.tcp;
/**
- * A class for represent local file transport constants
+ * A class used to identify remote servers
*/
-public final class FileConstants {
+public class RemoteResourceIdentifier {
+ private String host;
+ private int port;
- public static final String OUTPUT_FILE = "OUT_FILE";
+ public String getHost() {
+ return host;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
}
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
old mode 100755
new mode 100644
similarity index 57%
copy from transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
copy to transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
index 424940a..f2252a4
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
@@ -17,13 +17,24 @@
* under the License.
*/
-package org.apache.airavata.mft.transport.local;
+package org.apache.airavata.mft.transport.tcp.client;
+
+import org.apache.airavata.mft.core.api.ConnectorChannel;
+import org.apache.airavata.mft.core.api.SinkConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
/**
- * A class for represent local file transport constants
+ * A class which represents the connections between remote server and local client
*/
-public final class FileConstants {
+public class RemoteFileServerConnector extends AbstractConnector implements SinkConnector {
+ @Override
+ public boolean verifyUpload(ConnectorChannel channel) {
+ return false;
+ }
- public static final String OUTPUT_FILE = "OUT_FILE";
+ @Override
+ public ConnectorChannel openChannel() throws Exception {
+ return null;
+ }
}
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
old mode 100755
new mode 100644
similarity index 64%
copy from transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
copy to transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
index 424940a..ef24d6c
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
@@ -17,13 +17,29 @@
* under the License.
*/
-package org.apache.airavata.mft.transport.local;
+package org.apache.airavata.mft.transport.tcp.server;
/**
- * A class for represent local file transport constants
+ * A class which is responsible for accepting incoming tcp connections
+ * and delegate accepted channels to worker objects
*/
-public final class FileConstants {
+public class FileServer {
+
+ /**
+ * This starts the server and accepts incoming connections
+ * @return
+ */
+ public void start() {
+
+ }
+
+ /**
+ * Server is shutdown and stops accepting incoming connections
+ * @return
+ */
+ public void stop() {
+
+ }
- public static final String OUTPUT_FILE = "OUT_FILE";
}
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
old mode 100755
new mode 100644
similarity index 62%
copy from transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
copy to transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
index 424940a..b655fdf
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
@@ -17,13 +17,18 @@
* under the License.
*/
-package org.apache.airavata.mft.transport.local;
+package org.apache.airavata.mft.transport.tcp.server;
+
+import org.apache.airavata.mft.core.api.ConnectorChannel;
+import org.apache.airavata.mft.core.api.SourceConnector;
+import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
/**
- * A class for represent local file transport constants
+ * Represents a connection between remote client and FileServer
*/
-public final class FileConstants {
-
-
- public static final String OUTPUT_FILE = "OUT_FILE";
+public class FileServerConnector extends AbstractConnector implements SourceConnector {
+ @Override
+ public ConnectorChannel openChannel() throws Exception {
+ return null;
+ }
}
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/ServerConfig.java
old mode 100755
new mode 100644
similarity index 60%
copy from transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
copy to transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/ServerConfig.java
index 424940a..5264c86
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
+++ b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/ServerConfig.java
@@ -17,13 +17,34 @@
* under the License.
*/
-package org.apache.airavata.mft.transport.local;
+package org.apache.airavata.mft.transport.tcp.server;
/**
- * A class for represent local file transport constants
+ * A class which represents the server configurations
*/
-public final class FileConstants {
+public class ServerConfig {
+ private String host;
+ private int port;
- public static final String OUTPUT_FILE = "OUT_FILE";
+ public ServerConfig(String host, int port) {
+ this.host = host;
+ this.port = port;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
}