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/12/09 21:32:44 UTC
[airavata-mft] branch master updated: Refactoring streaming
transport to use StreamBuffer
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
The following commit(s) were added to refs/heads/master by this push:
new af4c67c Refactoring streaming transport to use StreamBuffer
af4c67c is described below
commit af4c67c47156d55dfe81c3f00b670c0eb1202f51
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Mon Dec 9 16:32:32 2019 -0500
Refactoring streaming transport to use StreamBuffer
---
core/pom.xml | 8 ++
.../apache/airavata/mft/core/ConnectorContext.java | 25 ++++
.../apache/airavata/mft/core/ResourceMetadata.java | 41 +++++++
.../org/apache/airavata/mft/core/TransferTask.java | 23 ++++
.../airavata/mft/core/api/CompletionCallback.java | 41 -------
.../apache/airavata/mft/core/api/Connector.java | 42 +------
.../airavata/mft/core/api/ConnectorChannel.java | 64 ----------
.../org/apache/airavata/mft/core/api/Mediator.java | 39 -------
.../airavata/mft/core/api/MetadataCollector.java | 7 ++
.../airavata/mft/core/api/SinkConnector.java | 37 ------
.../airavata/mft/core/api/SourceConnector.java | 31 -----
.../airavata/mft/core/api/StreamedReceiver.java | 26 -----
.../airavata/mft/core/api/StreamedSender.java | 26 -----
.../airavata/mft/core/api/TransportOperator.java | 5 -
.../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 | 85 --------------
.../streaming/DoubleByteArrayOutputStream.java | 130 ---------------------
.../mft/core/streaming/TransportMetadata.java | 13 ---
.../mft/core/streaming/TransportStream.java | 82 -------------
pom.xml | 14 +++
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 | 3 -
transport/s3-transport/pom.xml | 60 ----------
.../org/apache/airavata/mft/transport/s3/Main.java | 86 --------------
.../airavata/mft/transport/s3/S3Constants.java | 30 -----
.../airavata/mft/transport/s3/S3Receiver.java | 45 -------
.../mft/transport/s3/S3ResourceIdentifier.java | 68 -----------
.../apache/airavata/mft/transport/s3/S3Sender.java | 40 -------
.../airavata/mft/transport/s3/S3SinkConnector.java | 87 --------------
.../mft/transport/s3/S3SourceConnector.java | 61 ----------
.../mft/transport/s3/S3TransportOperator.java | 36 ------
transport/scp-transport/pom.xml | 5 +
.../apache/airavata/mft/transport/scp/Main.java | 106 -----------------
.../mft/transport/scp/SCPMetadataCollector.java | 66 +++++++++++
.../airavata/mft/transport/scp/SCPReceiver.java | 64 +++++-----
.../airavata/mft/transport/scp/SCPSender.java | 90 ++++++--------
.../mft/transport/scp/SCPTransportOperator.java | 44 -------
.../mft/transport/scp/SSHResourceIdentifier.java | 1 +
.../mft/transport/scp/TransportMediator.java | 44 +++++++
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 --------
57 files changed, 304 insertions(+), 2466 deletions(-)
diff --git a/core/pom.xml b/core/pom.xml
index ef33495..2f5d241 100755
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -44,5 +44,13 @@
</plugins>
</build>
+ <dependencies>
+ <dependency>
+ <groupId>net.ladenthin</groupId>
+ <artifactId>streambuffer</artifactId>
+ <version>1.1.0</version>
+ </dependency>
+ </dependencies>
+
</project>
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/airavata/mft/core/ConnectorContext.java b/core/src/main/java/org/apache/airavata/mft/core/ConnectorContext.java
new file mode 100644
index 0000000..ba25cf7
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/ConnectorContext.java
@@ -0,0 +1,25 @@
+package org.apache.airavata.mft.core;
+
+import net.ladenthin.streambuffer.StreamBuffer;
+
+public class ConnectorContext {
+
+ private StreamBuffer streamBuffer;
+ private ResourceMetadata metadata;
+
+ public StreamBuffer getStreamBuffer() {
+ return streamBuffer;
+ }
+
+ public void setStreamBuffer(StreamBuffer streamBuffer) {
+ this.streamBuffer = streamBuffer;
+ }
+
+ public ResourceMetadata getMetadata() {
+ return metadata;
+ }
+
+ public void setMetadata(ResourceMetadata metadata) {
+ this.metadata = metadata;
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/ResourceMetadata.java b/core/src/main/java/org/apache/airavata/mft/core/ResourceMetadata.java
new file mode 100644
index 0000000..63763d4
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/ResourceMetadata.java
@@ -0,0 +1,41 @@
+package org.apache.airavata.mft.core;
+
+public class ResourceMetadata {
+
+ private long resourceSize;
+ private long createdTime;
+ private long updateTime;
+ private String md5sum;
+
+ public long getResourceSize() {
+ return resourceSize;
+ }
+
+ public void setResourceSize(long resourceSize) {
+ this.resourceSize = resourceSize;
+ }
+
+ public long getCreatedTime() {
+ return createdTime;
+ }
+
+ public void setCreatedTime(long createdTime) {
+ this.createdTime = createdTime;
+ }
+
+ public long getUpdateTime() {
+ return updateTime;
+ }
+
+ public void setUpdateTime(long updateTime) {
+ this.updateTime = updateTime;
+ }
+
+ public String getMd5sum() {
+ return md5sum;
+ }
+
+ public void setMd5sum(String md5sum) {
+ this.md5sum = md5sum;
+ }
+}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/TransferTask.java b/core/src/main/java/org/apache/airavata/mft/core/TransferTask.java
new file mode 100644
index 0000000..0f1ae0a
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/TransferTask.java
@@ -0,0 +1,23 @@
+package org.apache.airavata.mft.core;
+
+import org.apache.airavata.mft.core.ConnectorContext;
+import org.apache.airavata.mft.core.api.Connector;
+
+import java.util.concurrent.Callable;
+
+public class TransferTask implements Callable<Integer> {
+
+ private Connector connector;
+ private ConnectorContext context;
+
+ public TransferTask(Connector connector, ConnectorContext context) {
+ this.connector = connector;
+ this.context = context;
+ }
+
+ @Override
+ public Integer call() throws Exception {
+ this.connector.startStream(context);
+ return 0;
+ }
+}
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
deleted file mode 100755
index f80c398..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/api/CompletionCallback.java
+++ /dev/null
@@ -1,41 +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.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
old mode 100755
new mode 100644
index 191ccc0..57532fb
--- 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
@@ -1,43 +1,9 @@
-/*
- * 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.ConnectorContext;
-/**
- * This represents a connection between external source or sink
- */
public interface Connector {
-
-
- /**
- * This returns a {@link ConnectorChannel}
- *
- * @return Channel
- */
- 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;
+ public void init(String resourceId, String credentialToken);
+ public void destroy();
+ void startStream(ConnectorContext context) 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
deleted file mode 100755
index 5818ee8..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/api/ConnectorChannel.java
+++ /dev/null
@@ -1,64 +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.api;
-
-import java.io.IOException;
-import java.nio.channels.Channel;
-
-/**
- * An interface represents a channel of a {@Link Connector}
- */
-public interface ConnectorChannel {
- /**
- * get the NIO channel of ChannelConnector
- *
- * @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);
-
- /**
- * provides the initiated connector
- * @return Connector
- */
- Connector getSourceConnector();
-}
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
deleted file mode 100755
index 92c479a..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/api/Mediator.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.api;
-
-
-/**
- * This represent the interface for all mediators, which mediates
- * 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
- *
- * @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/MetadataCollector.java b/core/src/main/java/org/apache/airavata/mft/core/api/MetadataCollector.java
new file mode 100644
index 0000000..9f7a076
--- /dev/null
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/MetadataCollector.java
@@ -0,0 +1,7 @@
+package org.apache.airavata.mft.core.api;
+
+import org.apache.airavata.mft.core.ResourceMetadata;
+
+public interface MetadataCollector {
+ public ResourceMetadata getGetResourceMetadata(String resourceId, String credentialToken) 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
deleted file mode 100755
index da5137f..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/api/SinkConnector.java
+++ /dev/null
@@ -1,37 +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.api;
-
-
-/**
- * This represents the output connector, where to write
- * data from the application.
- */
-public interface SinkConnector extends Connector {
-
-
- /**
- * Verify content upload of the given channel is completed
- *
- * @param channel
- * @return true if succes else false
- */
- boolean verifyUpload(ConnectorChannel 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
deleted file mode 100755
index db9922e..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/api/SourceConnector.java
+++ /dev/null
@@ -1,31 +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.api;
-
-
-/**
- * This represents the input connector, where to read data
- * from the application. Use this connector to specify reader related
- * methods
- */
-public interface SourceConnector extends Connector {
-
-
-}
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
deleted file mode 100755
index 56a070c..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedReceiver.java
+++ /dev/null
@@ -1,26 +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.api;
-
-import org.apache.airavata.mft.core.streaming.TransportStream;
-
-public interface StreamedReceiver {
- 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
deleted file mode 100755
index 1638a08..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/api/StreamedSender.java
+++ /dev/null
@@ -1,26 +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.api;
-
-import org.apache.airavata.mft.core.streaming.TransportStream;
-
-public interface StreamedSender {
- 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
deleted file mode 100755
index cda2d5e..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/api/TransportOperator.java
+++ /dev/null
@@ -1,5 +0,0 @@
-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/bufferedImpl/ConnectorException.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java
deleted file mode 100755
index 6fa1371..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/ConnectorException.java
+++ /dev/null
@@ -1,37 +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;
-
-/**
- * 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
deleted file mode 100755
index ff7172d..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/Constants.java
+++ /dev/null
@@ -1,29 +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;
-
-/**
- * Contains all the constants used in core classes.
- */
-public final class Constants {
-
- 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/AbstractChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java
deleted file mode 100755
index 0c2c259..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractChannel.java
+++ /dev/null
@@ -1,54 +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.channel;
-
-import org.apache.airavata.mft.core.api.Connector;
-import org.apache.airavata.mft.core.api.ConnectorChannel;
-
-import java.util.HashMap;
-
-/**
- * A class to implement common method of In and Out Channels
- */
-public abstract class AbstractChannel implements ConnectorChannel {
-
- private HashMap<String, Object> contextAttributeMap = new HashMap<>();
-
- private Connector myConnector;
-
- public AbstractChannel(Connector myConnector) {
- this.myConnector = myConnector;
- }
-
- @Override
- public void addChannelAttribute(String key, Object value) {
- contextAttributeMap.put(key, value);
- }
-
- @Override
- 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
deleted file mode 100755
index b0916dc..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/AbstractConnector.java
+++ /dev/null
@@ -1,50 +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.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;
-
-
-/**
- * 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 {
-
- @Override
- public void closeChannel(ConnectorChannel channel) throws ConnectorException {
-
- try {
- channel.closeChannel();
-
- } catch (IOException e) {
- throw new ConnectorException("Error occurred while closing stream", e);
- }
-
- }
-
-
-}
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
deleted file mode 100755
index 7a402c6..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/ChannelUtils.java
+++ /dev/null
@@ -1,92 +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.channel;
-
-import org.apache.airavata.mft.core.bufferedImpl.Constants;
-
-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/channel/InChannel.java b/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
deleted file mode 100755
index 20ee77d..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/InChannel.java
+++ /dev/null
@@ -1,69 +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.channel;
-
-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;
-
-/**
- * A class which represents the channel of a {@Link SinkConnector}
- */
-public class InChannel extends AbstractChannel {
-
- private InputStream inputStream;
-
- private ReadableByteChannel readableByteChannel;
-
-
- public InChannel(InputStream inputStream, Connector sourceConnector) {
- super(sourceConnector);
- this.inputStream = inputStream;
- readableByteChannel = Channels.newChannel(inputStream);
-
- }
-
- public InChannel(ReadableByteChannel channel, Connector sourceConnector) {
- super(sourceConnector);
- readableByteChannel = channel;
-
- }
-
-
- @Override
- public Channel getChannel() {
- return readableByteChannel;
- }
-
- @Override
- public void closeChannel() throws IOException {
- 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
deleted file mode 100755
index 0bba57a..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/channel/OutChannel.java
+++ /dev/null
@@ -1,67 +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.channel;
-
-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;
-
-/**
- * A class which represents the channel in {@Link SinkConnector}
- */
-public class OutChannel extends AbstractChannel {
-
- private OutputStream outputStream;
-
- private WritableByteChannel writableByteChannel;
-
-
- public OutChannel(OutputStream outputStream, Connector sourceConnector) {
- super(sourceConnector);
- this.outputStream = outputStream;
- writableByteChannel = Channels.newChannel(outputStream);
- }
-
- public OutChannel(WritableByteChannel byteChannel, Connector sourceConnector) {
- super(sourceConnector);
- this.writableByteChannel = byteChannel;
- }
-
-
- @Override
- public Channel getChannel() {
- return writableByteChannel;
- }
-
- @Override
- public void closeChannel() throws IOException {
- if (outputStream != null) {
- outputStream.close();
- } else if (writableByteChannel != null) {
- writableByteChannel.close();
- }
- }
-
-
-}
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
deleted file mode 100755
index b9b10de..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/bufferedImpl/mediation/PassthroughMediator.java
+++ /dev/null
@@ -1,84 +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.mediation;
-
-import org.apache.airavata.mft.core.api.*;
-import org.apache.airavata.mft.core.bufferedImpl.ConnectorException;
-import org.apache.airavata.mft.core.bufferedImpl.channel.ChannelUtils;
-
-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);
- }
-
-
- Object obj = dst.getSourceConnector();
- if (obj != null && obj instanceof SinkConnector) {
- SinkConnector connector = (SinkConnector) obj;
- boolean success = connector.verifyUpload(dst);
- 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 = src.getSourceConnector();
- Connector sinkConnector = dst.getSourceConnector();
- try {
- sourceConnector.closeChannel(src);
- sinkConnector.closeChannel(dst);
- } catch (Exception ex) {
- String msg = "Error occurred while closing channels";
- ConnectorException connectorException = new ConnectorException(msg, ex);
- callback.onComplete(msg, connectorException);
- }
- }
-
- }
-}
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
deleted file mode 100755
index 12323e4..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayInputStream.java
+++ /dev/null
@@ -1,85 +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.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;
- private InputStream currentInputStream;
- private int readCount;
-
- public DoubleByteArrayInputStream(DoubleByteArrayOutputStream outputStream) {
- this.outputStream = outputStream;
- this.currentInputStream = outputStream.asInputStream();
- }
-
- @Override
- public int read() throws IOException {
- 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
- public int available() throws IOException {
- refresh();
- 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();
- 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
deleted file mode 100755
index 4aefc2e..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/DoubleByteArrayOutputStream.java
+++ /dev/null
@@ -1,130 +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.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();
- private ByteArrayOutputStream stream2 = new ByteArrayOutputStream();
-
- private ByteArrayOutputStream currentStream = stream1;
- private int activeStream = 1;
-
- private long maxBytesPerStream = 1 * 1000 * 1000;
- private long processedBytes = 0;
- private long totalBytes = 0;
- private boolean clearedNonActiveStream = false;
-
- private boolean isClosed = false;
-
- @Override
- public void write(int b) throws IOException {
- if (processedBytes > maxBytesPerStream) {
- swapBuffers();
- }
- this.currentStream.write(b);
- processedBytes += 1;
- totalBytes += 1;
- }
-
- @Override
- public void write(byte[] b, int off, int len) throws IOException {
- if (processedBytes > maxBytesPerStream) {
- swapBuffers();
- }
- this.currentStream.write(b, off, len);
- processedBytes += len;
- totalBytes += 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 {
- Thread.sleep(100);
- System.out.println("Receiver waiting until non active buffer gets emptied " + processedBytes);
- } 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);
- }
-
- @Override
- public void flush() throws IOException {
- this.currentStream.flush();
- }
-
- @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;
- }
-
- /**
- * 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;
- }
- ByteArrayInputStream inputStream = new ByteArrayInputStream(activeStream == 1 ? stream2.toByteArray(): stream1.toByteArray());
- this.clearedNonActiveStream = true;
- return inputStream;
- }
-
- public boolean isClosed() {
- return isClosed;
- }
-}
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
deleted file mode 100755
index 1442052..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportMetadata.java
+++ /dev/null
@@ -1,13 +0,0 @@
-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
deleted file mode 100755
index 5956ffe..0000000
--- a/core/src/main/java/org/apache/airavata/mft/core/streaming/TransportStream.java
+++ /dev/null
@@ -1,82 +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.streaming;
-
-import java.io.*;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-public class TransportStream {
-
- private TransportMetadata metadata;
- private AtomicBoolean streamCompleted = new AtomicBoolean(false);
-
- private OutputStream outputStream = new DoubleByteArrayOutputStream();
- private DoubleByteArrayInputStream inputStream = new DoubleByteArrayInputStream((DoubleByteArrayOutputStream) outputStream);
-
- 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() {
- return inputStream;
- }
-
- public OutputStream getOutputStream() {
- return outputStream;
- }
-
- 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 setSourceId(String sourceId) {
- this.sourceId = sourceId;
- }
-
- public String getDestId() {
- return destId;
- }
-
- public void setDestId(String destId) {
- this.destId = destId;
- }
-}
diff --git a/pom.xml b/pom.xml
index 983b1cc..f0d9e84 100755
--- a/pom.xml
+++ b/pom.xml
@@ -79,4 +79,18 @@
<url>https://issues.apache.org/jira/browse/AIRAVATA</url>
</issueManagement>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.8.1</version>
+ <configuration>
+ <source>11</source>
+ <target>11</target>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
</project>
\ No newline at end of file
diff --git a/transport/local-transport/pom.xml b/transport/local-transport/pom.xml
deleted file mode 100755
index b0c2c7d..0000000
--- a/transport/local-transport/pom.xml
+++ /dev/null
@@ -1,41 +0,0 @@
-<?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>mft-transport</artifactId>
- <groupId>org.apache.airavata</groupId>
- <version>0.18-SNAPSHOT</version>
- </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/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
deleted file mode 100755
index 1d529af..0000000
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileConstants.java
+++ /dev/null
@@ -1,29 +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.transport.local;
-
-/**
- * A class of represent local file transport constants
- */
-public final class FileConstants {
-
-
- public static final String OUTPUT_FILE = "OUT_FILE";
-}
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
deleted file mode 100755
index 99a6732..0000000
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/FileUtils.java
+++ /dev/null
@@ -1,52 +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.transport.local;
-
-import org.apache.airavata.mft.core.bufferedImpl.ConnectorException;
-
-import java.io.File;
-
-/**
- * A class having utility methods of local file manipulation
- */
-public class FileUtils {
-
- /**
- * Validates and provides the file
- * @param src
- * @return
- * @throws ConnectorException
- */
- 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);
- }
-
- }
-
-
-}
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
deleted file mode 100755
index cac3b94..0000000
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/ResourceIdentifier.java
+++ /dev/null
@@ -1,37 +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.transport.local;
-
-/**
- * A class used to identify the local file path. Path should be a
- * same node local path
- */
-public class ResourceIdentifier {
- private String filePath;
-
- public void setFilePath(String filePath) {
- this.filePath = filePath;
- }
-
- public String getFilePath() {
- return filePath;
- }
-
-}
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
deleted file mode 100755
index ab94295..0000000
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SinkFileConnector.java
+++ /dev/null
@@ -1,60 +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.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/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
deleted file mode 100755
index 6a2e1fc..0000000
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/SourceFileConnector.java
+++ /dev/null
@@ -1,50 +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.transport.local;
-
-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 java.io.File;
-import java.io.FileInputStream;
-
-
-/**
- * A class responsible for open up the connection to read from file
- */
-public class SourceFileConnector extends AbstractConnector implements SourceConnector {
-
- private ResourceIdentifier resourceIdentifier;
-
- 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 90bc015..3bcbefa 100755
--- a/transport/pom.xml
+++ b/transport/pom.xml
@@ -34,9 +34,6 @@
<packaging>pom</packaging>
<modules>
<module>scp-transport</module>
- <module>s3-transport</module>
- <module>local-transport</module>
- <module>tcp-transport</module>
</modules>
diff --git a/transport/s3-transport/pom.xml b/transport/s3-transport/pom.xml
deleted file mode 100755
index 8f50b85..0000000
--- a/transport/s3-transport/pom.xml
+++ /dev/null
@@ -1,60 +0,0 @@
-<?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>mft-transport</artifactId>
- <groupId>org.apache.airavata</groupId>
- <version>0.18-SNAPSHOT</version>
- </parent>
- <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>
- <groupId>com.amazonaws</groupId>
- <artifactId>aws-java-sdk</artifactId>
- <version>1.11.163</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/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
deleted file mode 100755
index 76c5ae5..0000000
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/Main.java
+++ /dev/null
@@ -1,86 +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.transport.s3;
-
-import org.apache.airavata.mft.core.api.ConnectorChannel;
-import org.apache.airavata.mft.core.bufferedImpl.channel.AbstractConnector;
-import org.apache.airavata.mft.core.bufferedImpl.mediation.PassthroughMediator;
-
-/**
- * For Testing
- */
-public class Main {
-
-
- public static void main(String[] args) {
-
- Main main = new Main();
- main.execute();
-
- }
-
- public void execute() {
-
- 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");
-
-
- try {
-
-
- AbstractConnector s3SourceConnector = new S3SourceConnector(src);
- ConnectorChannel srcChannel = s3SourceConnector.openChannel();
-
- AbstractConnector s3SinkConnector = new S3SinkConnector(dst);
- ConnectorChannel dstChannel = s3SinkConnector.openChannel();
-
- 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
deleted file mode 100755
index 5e7aec0..0000000
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Constants.java
+++ /dev/null
@@ -1,30 +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.transport.s3;
-
-/**
- * Includes constants related to S3 SDK
- */
-public final class S3Constants {
-
- 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/S3Receiver.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java
deleted file mode 100755
index d7c516a..0000000
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Receiver.java
+++ /dev/null
@@ -1,45 +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.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
deleted file mode 100755
index 085589d..0000000
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3ResourceIdentifier.java
+++ /dev/null
@@ -1,68 +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.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
deleted file mode 100755
index d90a34b..0000000
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
+++ /dev/null
@@ -1,40 +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.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/S3SinkConnector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
deleted file mode 100755
index b6cc3fd..0000000
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SinkConnector.java
+++ /dev/null
@@ -1,87 +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.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.Constants;
-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;
-
-/**
- * Connector class which connects to a given S3 destination and provides
- * S3 output Stream.
- */
-public class S3SinkConnector extends AbstractConnector implements SinkConnector {
- private AmazonS3 s3Client;
- private S3ResourceIdentifier s3ResourceIdentifier;
-
- public S3SinkConnector(S3ResourceIdentifier s3ResourceIdentifier) {
- this.s3ResourceIdentifier = s3ResourceIdentifier;
- this.s3Client = S3TransportUtil.getS3Client(s3ResourceIdentifier.getAccessKey(),
- s3ResourceIdentifier.getSecretKey(), s3ResourceIdentifier.getRegion());
- }
-
- @Override
- 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
- (s3ResourceIdentifier.getBucket(), s3ResourceIdentifier.getRemoteFile())
- .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,this);
- outChannel.addChannelAttribute(S3Constants.HTTP_CONNECTION, connection);
- return outChannel;
- }
-
- @Override
- 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) {
- 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
deleted file mode 100755
index c1dc2e9..0000000
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3SourceConnector.java
+++ /dev/null
@@ -1,61 +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.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.channel.AbstractConnector;
-import org.apache.airavata.mft.core.bufferedImpl.channel.InChannel;
-
-import java.io.InputStream;
-
-/**
- * Connector class which connects to a given S3 source and provides
- * S3 Input Stream.
- */
-public class S3SourceConnector extends AbstractConnector implements SourceConnector {
-
- private AmazonS3 s3Client;
-
- private S3ResourceIdentifier identifier;
-
- public S3SourceConnector(S3ResourceIdentifier identifier) {
- this.identifier = identifier;
- this.s3Client = S3TransportUtil.getS3Client(identifier.getAccessKey(),
- identifier.getSecretKey(), identifier.getAccessKey());
- }
-
- @Override
- public ConnectorChannel openChannel() {
- S3Object s3object = s3Client.getObject(identifier.getBucket(),
- identifier.getRemoteFile());
- InputStream inputStream;
- if (s3object != null && s3object.getObjectContent() != null) {
- inputStream = s3object.getObjectContent();
- InChannel inChannel = new InChannel(inputStream,this);
- return inChannel;
- }
- return null;
- }
-
-
-}
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
deleted file mode 100755
index 79db705..0000000
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3TransportOperator.java
+++ /dev/null
@@ -1,36 +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.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/scp-transport/pom.xml b/transport/scp-transport/pom.xml
index d3cbc4e..6480255 100755
--- a/transport/scp-transport/pom.xml
+++ b/transport/scp-transport/pom.xml
@@ -39,6 +39,11 @@
<version>0.1.55</version>
</dependency>
<dependency>
+ <groupId>com.hierynomus</groupId>
+ <artifactId>sshj</artifactId>
+ <version>0.23.0</version>
+ </dependency>
+ <dependency>
<groupId>org.apache.airavata</groupId>
<artifactId>mft-core</artifactId>
<version>0.18-SNAPSHOT</version>
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
deleted file mode 100755
index f0d63eb..0000000
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/Main.java
+++ /dev/null
@@ -1,106 +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.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;
-
-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("1"));
- final TransportStream stream = new TransportStream("1", "4", metadata);
-
- 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.isStreamCompleted()) {
- break;
- } else {
- try {
- Thread.sleep(100);
- System.out.println("Waiting " + read);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- }
- }
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
- };
-
- Runnable receiverRun = new Runnable() {
- @Override
- public void run() {
- SCPReceiver receiver = new SCPReceiver();
- //S3Receiver receiver = new S3Receiver();
- try {
- receiver.receive(stream);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- };
-
- Runnable senderRun = new Runnable() {
- @Override
- public void run() {
- //SCPSender sender = new SCPSender();
- S3Sender sender = new S3Sender();
- try {
- sender.send(stream);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- };
-
- //Thread t1 = new Thread(r1);
- //t1.start();
-
- Thread senderThread = new Thread(senderRun);
- senderThread.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/SCPMetadataCollector.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPMetadataCollector.java
new file mode 100644
index 0000000..88ad387
--- /dev/null
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPMetadataCollector.java
@@ -0,0 +1,66 @@
+package org.apache.airavata.mft.transport.scp;
+
+import net.schmizz.sshj.SSHClient;
+import net.schmizz.sshj.sftp.FileAttributes;
+import net.schmizz.sshj.sftp.SFTPClient;
+import net.schmizz.sshj.userauth.keyprovider.KeyProvider;
+import net.schmizz.sshj.userauth.method.AuthKeyboardInteractive;
+import net.schmizz.sshj.userauth.method.AuthMethod;
+import net.schmizz.sshj.userauth.method.AuthPublickey;
+import net.schmizz.sshj.userauth.method.ChallengeResponseProvider;
+import net.schmizz.sshj.userauth.password.Resource;
+import org.apache.airavata.mft.core.ResourceMetadata;
+import org.apache.airavata.mft.core.api.MetadataCollector;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+public class SCPMetadataCollector implements MetadataCollector {
+
+ public ResourceMetadata getGetResourceMetadata(String resourceId, String credentialToken) throws IOException {
+
+ SSHResourceIdentifier sshResourceIdentifier = SCPTransportUtil.getSSHResourceIdentifier(resourceId);
+
+ try (SSHClient sshClient = new SSHClient()) {
+ sshClient.addHostKeyVerifier((h, p, key) -> true);
+ KeyProvider keyProvider = sshClient.loadKeys(sshResourceIdentifier.getKeyFile(), sshResourceIdentifier.getKeyPassphrase());
+ final List<AuthMethod> am = new LinkedList<>();
+ am.add(new AuthPublickey(keyProvider));
+ am.add(new AuthKeyboardInteractive(new ChallengeResponseProvider() {
+ @Override
+ public List<String> getSubmethods() {
+ return new ArrayList<>();
+ }
+
+ @Override
+ public void init(Resource resource, String name, String instruction) {}
+
+ @Override
+ public char[] getResponse(String prompt, boolean echo) {
+ return new char[0];
+ }
+
+ @Override
+ public boolean shouldRetry() {
+ return false;
+ }
+ }));
+
+ sshClient.connect(sshResourceIdentifier.getHost(), sshResourceIdentifier.getPort());
+ sshClient.auth(sshResourceIdentifier.getUser(), am);
+
+ try (SFTPClient sftpClient = sshClient.newSFTPClient()) {
+ FileAttributes lstat = sftpClient.lstat(sshResourceIdentifier.getRemotePath());
+ sftpClient.close();
+
+ ResourceMetadata metadata = new ResourceMetadata();
+ metadata.setResourceSize(lstat.getSize());
+ metadata.setCreatedTime(lstat.getAtime());
+ metadata.setUpdateTime(lstat.getMtime());
+ return metadata;
+ }
+ }
+ }
+}
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 100755
new mode 100644
index d5871cb..091df74
--- 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,47 +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.transport.scp;
-import com.jcraft.jsch.*;
-import org.apache.airavata.mft.core.api.StreamedReceiver;
-import org.apache.airavata.mft.core.streaming.TransportStream;
+import com.jcraft.jsch.ChannelExec;
+import com.jcraft.jsch.Session;
+import net.ladenthin.streambuffer.StreamBuffer;
+import org.apache.airavata.mft.core.ConnectorContext;
+import org.apache.airavata.mft.core.api.Connector;
-import java.io.*;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
-public class SCPReceiver implements StreamedReceiver {
+public class SCPReceiver implements Connector {
+ private Session session;
+ private SSHResourceIdentifier sshResourceIdentifier;
- @Override
- public void receive(TransportStream stream) throws Exception {
- SSHResourceIdentifier sshResourceIdentifier = SCPTransportUtil.getSSHResourceIdentifier(stream.getSourceId());
- Session session = SCPTransportUtil.createSession(sshResourceIdentifier.getUser(), sshResourceIdentifier.getHost(),
+ public void init(String resourceId, String credentialToken) {
+ this.sshResourceIdentifier = SCPTransportUtil.getSSHResourceIdentifier(resourceId);
+ this.session = SCPTransportUtil.createSession(sshResourceIdentifier.getUser(), sshResourceIdentifier.getHost(),
sshResourceIdentifier.getPort(),
sshResourceIdentifier.getKeyFile(),
sshResourceIdentifier.getKeyPassphrase());
- transferRemoteToStream(session, sshResourceIdentifier.getRemotePath(), stream);
}
- private void transferRemoteToStream(Session session, String from, TransportStream stream) throws JSchException, IOException {
+ public void destroy() {
+ this.session.disconnect();
+ }
+
+ public void startStream(ConnectorContext context) throws Exception {
+ transferRemoteToStream(session, sshResourceIdentifier.getRemotePath(), context.getStreamBuffer());
+ }
+
+ private void transferRemoteToStream(Session session, String from, StreamBuffer streamBuffer) throws Exception {
+
+ OutputStream outputStream = streamBuffer.getOutputStream();
+
System.out.println("Starting scp receive");
// exec 'scp -f rfile' remotely
String command = "scp -f " + from;
- Channel channel = session.openChannel("exec");
+ com.jcraft.jsch.Channel channel = session.openChannel("exec");
((ChannelExec) channel).setCommand(command);
// get I/O streams for remote scp
@@ -101,8 +97,9 @@ public class SCPReceiver implements StreamedReceiver {
// error
break;
}
- stream.getOutputStream().write(buf, 0, bufSize);
- stream.getOutputStream().flush();
+ //System.out.println("Read " + bufSize);
+ outputStream.write(buf, 0, bufSize);
+ outputStream.flush();
filesize -= bufSize;
if (filesize == 0L) break;
@@ -118,7 +115,6 @@ public class SCPReceiver implements StreamedReceiver {
out.flush();
}
- stream.setStreamCompleted(true);
channel.disconnect();
session.disconnect();
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 100755
new mode 100644
index 6b4dfe7..75413ca
--- 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,51 +1,44 @@
-/*
- *
- * 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;
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(TransportStream stream) throws Exception {
- SSHResourceIdentifier sshResourceIdentifier = SCPTransportUtil.getSSHResourceIdentifier(stream.getDestId());
- Session session = SCPTransportUtil.createSession(sshResourceIdentifier.getUser(), sshResourceIdentifier.getHost(),
+import net.ladenthin.streambuffer.StreamBuffer;
+import org.apache.airavata.mft.core.ConnectorContext;
+import org.apache.airavata.mft.core.api.Connector;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+public class SCPSender implements Connector {
+ private SSHResourceIdentifier sshResourceIdentifier;
+ private Session session;
+ public void init(String resourceId, String credentialToken) {
+ this.sshResourceIdentifier = SCPTransportUtil.getSSHResourceIdentifier(resourceId);
+ this.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 {
+ public void destroy() {
+ this.session.disconnect();
+ }
+
+ public void startStream(ConnectorContext context) throws Exception {
+ copyLocalToRemote(this.session, sshResourceIdentifier.getRemotePath(), context.getStreamBuffer(), context.getMetadata().getResourceSize());
+ }
+
+ private void copyLocalToRemote(Session session, String to, StreamBuffer streamBuffer, long fileSize) throws JSchException, IOException {
+
+ InputStream inputStream = streamBuffer.getInputStream();
+
boolean ptimestamp = true;
// exec 'scp -t rfile' remotely
String command = "scp " + (ptimestamp ? "-p" : "") + " -t " + to;
- Channel channel = session.openChannel("exec");
+ com.jcraft.jsch.Channel channel = session.openChannel("exec");
((ChannelExec) channel).setCommand(command);
// get I/O streams for remote scp
@@ -71,8 +64,7 @@ public class SCPSender implements StreamedSender {
}
// send "C0644 filesize filename", where filename should not include '/'
- long filesize = stream.getMetadata().getLength();
- command = "C0644 " + filesize + " ";
+ command = "C0644 " + fileSize + " ";
if (to.lastIndexOf('/') > 0) {
command += to.substring(to.lastIndexOf('/') + 1);
} else {
@@ -91,20 +83,15 @@ public class SCPSender implements StreamedSender {
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());
+ int len = inputStream.read(buf, 0, buf.length);
+ if (len == -1) {
+ break;
} else {
out.write(buf, 0, len); //out.flush();
totalWritten += len;
+ if (totalWritten == fileSize) {
+ break;
+ }
}
}
@@ -118,17 +105,12 @@ public class SCPSender implements StreamedSender {
}
out.close();
- try {
- if (stream.getInputStream() != null) stream.getInputStream().close();
- } catch (Exception ex) {
- System.out.println(ex);
- }
-
channel.disconnect();
session.disconnect();
+ System.out.println("Done sending");
}
- public static int checkAck(InputStream in) throws IOException {
+ public int checkAck(InputStream in) throws IOException {
int b = in.read();
// b may be 0 for success,
// 1 for error,
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
deleted file mode 100755
index eccef2d..0000000
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportOperator.java
+++ /dev/null
@@ -1,44 +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.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/SSHResourceIdentifier.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SSHResourceIdentifier.java
index 571b0ab..dbab1b4 100755
--- 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
@@ -20,6 +20,7 @@
package org.apache.airavata.mft.transport.scp;
public class SSHResourceIdentifier {
+
private String id;
private String remotePath;
private String host;
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/TransportMediator.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/TransportMediator.java
new file mode 100644
index 0000000..7d92ef0
--- /dev/null
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/TransportMediator.java
@@ -0,0 +1,44 @@
+package org.apache.airavata.mft.transport.scp;
+
+import net.ladenthin.streambuffer.StreamBuffer;
+import org.apache.airavata.mft.core.ConnectorContext;
+import org.apache.airavata.mft.core.ResourceMetadata;
+import org.apache.airavata.mft.core.TransferTask;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+public class TransportMediator {
+
+ private ExecutorService executor = Executors.newFixedThreadPool(10);
+
+ public static void main(String args[]) throws Exception {
+ TransportMediator mediator = new TransportMediator();
+ SCPReceiver receiver = new SCPReceiver();
+ receiver.init("1", "");
+
+ SCPSender scpSender = new SCPSender();
+ scpSender.init("2", "");
+
+ SCPMetadataCollector metadataCollector = new SCPMetadataCollector();
+ ResourceMetadata metadata = metadataCollector.getGetResourceMetadata("1", "");
+ mediator.transfer(receiver, scpSender, metadata);
+ }
+
+ public void transfer(SCPReceiver receiver, SCPSender sender, ResourceMetadata metadata) throws Exception {
+
+ StreamBuffer streamBuffer = new StreamBuffer();
+ ConnectorContext context = new ConnectorContext();
+ context.setMetadata(metadata);
+ context.setStreamBuffer(streamBuffer);
+
+ TransferTask recvTask = new TransferTask(receiver, context);
+ TransferTask sendTask = new TransferTask(sender, context);
+
+ Future<Integer> recvFuture = executor.submit(recvTask);
+ Future<Integer> sendFuture = executor.submit(sendTask);
+
+ executor.shutdown();
+ }
+}
diff --git a/transport/tcp-transport/pom.xml b/transport/tcp-transport/pom.xml
deleted file mode 100644
index b17cc41..0000000
--- a/transport/tcp-transport/pom.xml
+++ /dev/null
@@ -1,41 +0,0 @@
-<?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>mft-transport</artifactId>
- <groupId>org.apache.airavata</groupId>
- <version>0.18-SNAPSHOT</version>
- </parent>
- <modelVersion>4.0.0</modelVersion>
-
- <artifactId>mft-tcp-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/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
deleted file mode 100644
index 715eb6c..0000000
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/Constants.java
+++ /dev/null
@@ -1,30 +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.transport.tcp;
-
-/**
- * A class which contains all Constants
- */
-public final class Constants {
-
- public static final String UNKNOWN_HOST = "UNKNOWN_HOST";
- 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/RemoteResourceIdentifier.java b/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
deleted file mode 100644
index 7fddfaf..0000000
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/RemoteResourceIdentifier.java
+++ /dev/null
@@ -1,62 +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.transport.tcp;
-
-import java.net.InetSocketAddress;
-
-/**
- * A class used to identify remote clients and servers
- */
-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);
- }
-
- 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;
- }
-
- /**
- * 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
deleted file mode 100644
index 31fe519..0000000
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/client/RemoteFileServerConnector.java
+++ /dev/null
@@ -1,54 +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.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 {
-
- private RemoteResourceIdentifier identifier;
-
- public RemoteFileServerConnector(RemoteResourceIdentifier remoteResourceIdentifier) {
- this.identifier = remoteResourceIdentifier;
- }
-
-
- @Override
- public boolean verifyUpload(ConnectorChannel channel) {
- //TODO Implement this later
- return true;
- }
-
- @Override
- public ConnectorChannel openChannel() throws Exception {
- 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
deleted file mode 100644
index 2089e69..0000000
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServer.java
+++ /dev/null
@@ -1,106 +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.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 {
-
- 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) {
- 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);
-
- serverSocketChannel.bind(new InetSocketAddress(port));
-
- 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
deleted file mode 100644
index aa617d5..0000000
--- a/transport/tcp-transport/src/main/java/org/apache/airavata/mft/transport/tcp/server/FileServerConnector.java
+++ /dev/null
@@ -1,51 +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.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 {
- SocketChannel socketChannel = this.acceptedChannelCache.take();
- InChannel inChannel = new InChannel(socketChannel, this);
- return inChannel;
-
- }
-
- public void addChannel(SocketChannel channel) throws InterruptedException {
- this.acceptedChannelCache.put(channel);
- }
-
-}