Making storage as first class entity and supporting http transport in the agent
diff --git a/admin/src/main/java/org/apache/airavata/mft/admin/models/TransferCommand.java b/admin/src/main/java/org/apache/airavata/mft/admin/models/TransferCommand.java
index 175e867..b74b90a 100644
--- a/admin/src/main/java/org/apache/airavata/mft/admin/models/TransferCommand.java
+++ b/admin/src/main/java/org/apache/airavata/mft/admin/models/TransferCommand.java
@@ -22,12 +22,14 @@
public class TransferCommand {
private String transferId;
- private String sourceId;
+ private String sourceStorageId;
+ private String sourcePath;
private String sourceType;
private String sourceToken;
private String sourceResourceBackend;
private String sourceCredentialBackend;
- private String destinationId;
+ private String destinationStorageId;
+ private String destinationPath;
private String destinationType;
private String destinationToken;
private String destResourceBackend;
@@ -42,12 +44,21 @@
return this;
}
- public String getSourceId() {
- return sourceId;
+ public String getSourceStorageId() {
+ return sourceStorageId;
}
- public TransferCommand setSourceId(String sourceId) {
- this.sourceId = sourceId;
+ public TransferCommand setSourceStorageId(String sourceStorageId) {
+ this.sourceStorageId = sourceStorageId;
+ return this;
+ }
+
+ public String getSourcePath() {
+ return sourcePath;
+ }
+
+ public TransferCommand setSourcePath(String sourcePath) {
+ this.sourcePath = sourcePath;
return this;
}
@@ -87,12 +98,21 @@
return this;
}
- public String getDestinationId() {
- return destinationId;
+ public String getDestinationStorageId() {
+ return destinationStorageId;
}
- public TransferCommand setDestinationId(String destinationId) {
- this.destinationId = destinationId;
+ public TransferCommand setDestinationStorageId(String destinationStorageId) {
+ this.destinationStorageId = destinationStorageId;
+ return this;
+ }
+
+ public String getDestinationPath() {
+ return destinationPath;
+ }
+
+ public TransferCommand setDestinationPath(String destinationPath) {
+ this.destinationPath = destinationPath;
return this;
}
diff --git a/admin/src/main/java/org/apache/airavata/mft/admin/models/TransferRequest.java b/admin/src/main/java/org/apache/airavata/mft/admin/models/TransferRequest.java
index ecd5cdc..8512713 100644
--- a/admin/src/main/java/org/apache/airavata/mft/admin/models/TransferRequest.java
+++ b/admin/src/main/java/org/apache/airavata/mft/admin/models/TransferRequest.java
@@ -21,12 +21,14 @@
public class TransferRequest {
- private String sourceId;
+ private String sourceStorageId;
+ private String sourcePath;
private String sourceType;
private String sourceToken;
private String sourceResourceBackend;
private String sourceCredentialBackend;
- private String destinationId;
+ private String destinationStorageId;
+ private String destinationPath;
private String destinationType;
private String destinationToken;
private String destResourceBackend;
@@ -34,12 +36,21 @@
private boolean affinityTransfer;
private Map<String, Integer> targetAgents;
- public String getSourceId() {
- return sourceId;
+ public String getSourceStorageId() {
+ return sourceStorageId;
}
- public TransferRequest setSourceId(String sourceId) {
- this.sourceId = sourceId;
+ public TransferRequest setSourceStorageId(String sourceStorageId) {
+ this.sourceStorageId = sourceStorageId;
+ return this;
+ }
+
+ public String getSourcePath() {
+ return sourcePath;
+ }
+
+ public TransferRequest setSourcePath(String sourcePath) {
+ this.sourcePath = sourcePath;
return this;
}
@@ -79,12 +90,21 @@
return this;
}
- public String getDestinationId() {
- return destinationId;
+ public String getDestinationStorageId() {
+ return destinationStorageId;
}
- public TransferRequest setDestinationId(String destinationId) {
- this.destinationId = destinationId;
+ public TransferRequest setDestinationStorageId(String destinationStorageId) {
+ this.destinationStorageId = destinationStorageId;
+ return this;
+ }
+
+ public String getDestinationPath() {
+ return destinationPath;
+ }
+
+ public TransferRequest setDestinationPath(String destinationPath) {
+ this.destinationPath = destinationPath;
return this;
}
diff --git a/agent/pom.xml b/agent/pom.xml
index 7ca3753..7d6765e 100644
--- a/agent/pom.xml
+++ b/agent/pom.xml
@@ -99,6 +99,11 @@
<artifactId>log4j-over-slf4j</artifactId>
<version>${log4j.over.slf4j}</version>
</dependency>
+ <dependency>
+ <groupId>com.sun.activation</groupId>
+ <artifactId>javax.activation</artifactId>
+ <version>1.2.0</version>
+ </dependency>
</dependencies>
<build>
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/AppConfig.java b/agent/src/main/java/org/apache/airavata/mft/agent/AppConfig.java
index 4409232..49b3ddf 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/AppConfig.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/AppConfig.java
@@ -18,10 +18,10 @@
package org.apache.airavata.mft.agent;
import org.apache.airavata.mft.admin.MFTConsulClient;
+import org.apache.airavata.mft.agent.http.HttpTransferRequestsStore;
import org.apache.airavata.mft.agent.rpc.RPCParser;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.PropertySource;
@Configuration
public class AppConfig {
@@ -41,4 +41,9 @@
public RPCParser rpcParser() {
return new RPCParser();
}
+
+ @Bean
+ public HttpTransferRequestsStore transferRequestStore() {
+ return new HttpTransferRequestsStore();
+ }
}
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java b/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
index c7bd8f7..19a0dfb 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
@@ -30,6 +30,8 @@
import org.apache.airavata.mft.admin.models.TransferCommand;
import org.apache.airavata.mft.admin.models.TransferState;
import org.apache.airavata.mft.admin.models.rpc.SyncRPCRequest;
+import org.apache.airavata.mft.agent.http.HttpServer;
+import org.apache.airavata.mft.agent.http.HttpTransferRequestsStore;
import org.apache.airavata.mft.agent.rpc.RPCParser;
import org.apache.airavata.mft.core.ConnectorResolver;
import org.apache.airavata.mft.core.MetadataCollectorResolver;
@@ -101,6 +103,9 @@
@Autowired
private MFTConsulClient mftConsulClient;
+ @Autowired
+ private HttpTransferRequestsStore transferRequestsStore;
+
public void init() {
transferMessageCache = KVCache.newCache(mftConsulClient.getKvClient(), MFTConsulClient.AGENTS_TRANSFER_REQUEST_MESSAGE_PATH + agentId );
rpcMessageCache = KVCache.newCache(mftConsulClient.getKvClient(), MFTConsulClient.AGENTS_RPC_REQUEST_MESSAGE_PATH + agentId );
@@ -147,11 +152,11 @@
Optional<Connector> inConnectorOpt = ConnectorResolver.resolveConnector(request.getSourceType(), "IN");
Connector inConnector = inConnectorOpt.orElseThrow(() -> new Exception("Could not find an in connector for given input"));
- inConnector.init(request.getSourceId(), request.getSourceToken(), resourceServiceHost, resourceServicePort, secretServiceHost, secretServicePort);
+ inConnector.init(request.getSourceStorageId(), request.getSourceToken(), resourceServiceHost, resourceServicePort, secretServiceHost, secretServicePort);
Optional<Connector> outConnectorOpt = ConnectorResolver.resolveConnector(request.getDestinationType(), "OUT");
Connector outConnector = outConnectorOpt.orElseThrow(() -> new Exception("Could not find an out connector for given input"));
- outConnector.init(request.getDestinationId(), request.getDestinationToken(), resourceServiceHost, resourceServicePort, secretServiceHost, secretServicePort);
+ outConnector.init(request.getDestinationStorageId(), request.getDestinationToken(), resourceServiceHost, resourceServicePort, secretServiceHost, secretServicePort);
Optional<MetadataCollector> srcMetadataCollectorOp = MetadataCollectorResolver.resolveMetadataCollector(request.getSourceType());
MetadataCollector srcMetadataCollector = srcMetadataCollectorOp.orElseThrow(() -> new Exception("Could not find a metadata collector for source"));
@@ -221,6 +226,19 @@
transferMessageCache.start();
}
+ private void acceptHTTPRequests() {
+ logger.info("Starting the HTTP front end");
+
+ new Thread(() -> {
+ HttpServer httpServer = new HttpServer(3333, false, transferRequestsStore);
+ try {
+ httpServer.run();
+ } catch (Exception e) {
+ logger.error("Http frontend server start failed", e);
+ }
+ }).start();
+ }
+
private boolean connectAgent() throws MFTConsulClientException {
final ImmutableSession session = ImmutableSession.builder()
.name(agentId)
@@ -316,6 +334,7 @@
acceptTransferRequests();
acceptRPCRequests();
+ acceptHTTPRequests();
}
@Override
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java b/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
index cd90503..c1ac253 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
@@ -52,7 +52,10 @@
MetadataCollector destMetadataCollector, BiConsumer<String, TransferState> onStatusCallback,
BiConsumer<String, Boolean> exitingCallback) throws Exception {
- FileResourceMetadata srcMetadata = srcMetadataCollector.getFileResourceMetadata(command.getSourceId(), command.getSourceToken());
+ FileResourceMetadata srcMetadata = srcMetadataCollector.getFileResourceMetadata(
+ command.getSourceStorageId(),
+ command.getSourcePath(),
+ command.getSourceToken());
final long resourceSize = srcMetadata.getResourceSize();
logger.debug("Source file size {}. MD5 {}", resourceSize, srcMetadata.getMd5sum());
@@ -65,8 +68,8 @@
context.setStreamBuffer(streamBuffer);
context.setTransferId(command.getTransferId());
- TransferTask recvTask = new TransferTask(inConnector, context);
- TransferTask sendTask = new TransferTask(outConnector, context);
+ TransferTask recvTask = new TransferTask(inConnector, context, command.getSourcePath());
+ TransferTask sendTask = new TransferTask(outConnector, context, command.getDestinationPath());
List<Future<Integer>> futureList = new ArrayList<>();
ExecutorCompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);
@@ -121,14 +124,19 @@
}
if (!transferErrored) {
- Boolean transferred = destMetadataCollector.isAvailable(command.getDestinationId(), command.getDestinationToken());
+ Boolean transferred = destMetadataCollector.isAvailable(
+ command.getDestinationStorageId(),
+ command.getDestinationPath(),
+ command.getDestinationToken());
if (!transferred) {
logger.error("Transfer completed but resource is not available in destination");
throw new Exception("Transfer completed but resource is not available in destination");
}
- FileResourceMetadata destMetadata = destMetadataCollector.getFileResourceMetadata(command.getDestinationId(),
+ FileResourceMetadata destMetadata = destMetadataCollector.getFileResourceMetadata(
+ command.getDestinationStorageId(),
+ command.getDestinationPath(),
command.getDestinationToken());
boolean doIntegrityVerify = true;
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/http/ConnectorParams.java b/agent/src/main/java/org/apache/airavata/mft/agent/http/ConnectorParams.java
new file mode 100644
index 0000000..bb61ebd
--- /dev/null
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/http/ConnectorParams.java
@@ -0,0 +1,78 @@
+/*
+ * 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.agent.http;
+
+public class ConnectorParams {
+
+ private String storageId, credentialToken, resourceServiceHost, secretServiceHost;
+ private int resourceServicePort, secretServicePort;
+
+ public String getStorageId() {
+ return storageId;
+ }
+
+ public ConnectorParams setStorageId(String storageId) {
+ this.storageId = storageId;
+ return this;
+ }
+
+ public String getCredentialToken() {
+ return credentialToken;
+ }
+
+ public ConnectorParams setCredentialToken(String credentialToken) {
+ this.credentialToken = credentialToken;
+ return this;
+ }
+
+ public String getResourceServiceHost() {
+ return resourceServiceHost;
+ }
+
+ public ConnectorParams setResourceServiceHost(String resourceServiceHost) {
+ this.resourceServiceHost = resourceServiceHost;
+ return this;
+ }
+
+ public String getSecretServiceHost() {
+ return secretServiceHost;
+ }
+
+ public ConnectorParams setSecretServiceHost(String secretServiceHost) {
+ this.secretServiceHost = secretServiceHost;
+ return this;
+ }
+
+ public int getResourceServicePort() {
+ return resourceServicePort;
+ }
+
+ public ConnectorParams setResourceServicePort(int resourceServicePort) {
+ this.resourceServicePort = resourceServicePort;
+ return this;
+ }
+
+ public int getSecretServicePort() {
+ return secretServicePort;
+ }
+
+ public ConnectorParams setSecretServicePort(int secretServicePort) {
+ this.secretServicePort = secretServicePort;
+ return this;
+ }
+}
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpDownloadRequest.java b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpDownloadRequest.java
new file mode 100644
index 0000000..2cf56af
--- /dev/null
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpDownloadRequest.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.airavata.mft.agent.http;
+
+import org.apache.airavata.mft.core.api.Connector;
+import org.apache.airavata.mft.core.api.MetadataCollector;
+
+public class HttpDownloadRequest {
+
+ private ConnectorParams connectorParams;
+ private Connector srcConnector;
+ private MetadataCollector srcMetadataCollector;
+ private String srcResourceId;
+ private String srcToken;
+
+ public ConnectorParams getConnectorParams() {
+ return connectorParams;
+ }
+
+ public HttpDownloadRequest setConnectorParams(ConnectorParams connectorParams) {
+ this.connectorParams = connectorParams;
+ return this;
+ }
+
+ public Connector getSrcConnector() {
+ return srcConnector;
+ }
+
+ public HttpDownloadRequest setSrcConnector(Connector srcConnector) {
+ this.srcConnector = srcConnector;
+ return this;
+ }
+
+ public MetadataCollector getSrcMetadataCollector() {
+ return srcMetadataCollector;
+ }
+
+ public HttpDownloadRequest setSrcMetadataCollector(MetadataCollector srcMetadataCollector) {
+ this.srcMetadataCollector = srcMetadataCollector;
+ return this;
+ }
+
+ public String getSrcResourceId() {
+ return srcResourceId;
+ }
+
+ public HttpDownloadRequest setSrcResourceId(String srcResourceId) {
+ this.srcResourceId = srcResourceId;
+ return this;
+ }
+
+ public String getSrcToken() {
+ return srcToken;
+ }
+
+ public HttpDownloadRequest setSrcToken(String srcToken) {
+ this.srcToken = srcToken;
+ return this;
+ }
+}
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServer.java b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServer.java
new file mode 100644
index 0000000..4cc0d8a
--- /dev/null
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServer.java
@@ -0,0 +1,73 @@
+/*
+ * 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.agent.http;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslProvider;
+import io.netty.handler.ssl.util.SelfSignedCertificate;
+
+public class HttpServer {
+
+ private final int port;
+ private final boolean enableSSL;
+ private HttpTransferRequestsStore transferRequestsStore;
+
+ public HttpServer(int port, boolean enableSSL, HttpTransferRequestsStore transferRequestsStore) {
+ this.port = port;
+ this.enableSSL = enableSSL;
+ this.transferRequestsStore = transferRequestsStore;
+ }
+
+ public void run() throws Exception {
+ // Configure SSL.
+ final SslContext sslCtx;
+ if (enableSSL) {
+ SelfSignedCertificate ssc = new SelfSignedCertificate();
+ sslCtx = SslContext.newServerContext(SslProvider.JDK, ssc.certificate(), ssc.privateKey());
+ } else {
+ sslCtx = null;
+ }
+
+ EventLoopGroup bossGroup = new NioEventLoopGroup(1);
+ EventLoopGroup workerGroup = new NioEventLoopGroup();
+ try {
+ ServerBootstrap b = new ServerBootstrap();
+ b.group(bossGroup, workerGroup)
+ .channel(NioServerSocketChannel.class)
+ .handler(new LoggingHandler(LogLevel.INFO))
+ .childHandler(new HttpServerInitializer(sslCtx, transferRequestsStore));
+
+ Channel ch = b.bind(port).sync().channel();
+
+ System.err.println("Open your web browser and navigate to " +
+ (enableSSL? "https" : "http") + "://127.0.0.1:" + port + '/');
+
+ ch.closeFuture().sync();
+ } finally {
+ bossGroup.shutdownGracefully();
+ workerGroup.shutdownGracefully();
+ }
+ }
+}
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServerHandler.java b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServerHandler.java
new file mode 100644
index 0000000..57ee4f0
--- /dev/null
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServerHandler.java
@@ -0,0 +1,193 @@
+/*
+ * 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.agent.http;
+
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelProgressiveFuture;
+import io.netty.channel.ChannelProgressiveFutureListener;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.*;
+import io.netty.handler.stream.ChunkedStream;
+import io.netty.util.CharsetUtil;
+import org.apache.airavata.mft.core.ConnectorContext;
+import org.apache.airavata.mft.core.DoubleStreamingBuffer;
+import org.apache.airavata.mft.core.FileResourceMetadata;
+import org.apache.airavata.mft.core.TransferTask;
+import org.apache.airavata.mft.core.api.Connector;
+import org.apache.airavata.mft.core.api.MetadataCollector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.activation.MimetypesFileTypeMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import static io.netty.handler.codec.http.HttpMethod.*;
+import static io.netty.handler.codec.http.HttpResponseStatus.*;
+import static io.netty.handler.codec.http.HttpVersion.*;
+
+public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+
+ private static final Logger logger = LoggerFactory.getLogger(HttpServerHandler.class);
+
+ private final HttpTransferRequestsStore transferRequestsStore;
+ private final ExecutorService executor = Executors.newFixedThreadPool(10);
+
+ public HttpServerHandler(HttpTransferRequestsStore transferRequestsStore) {
+ this.transferRequestsStore = transferRequestsStore;
+ }
+
+ @Override
+ public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
+ if (!request.decoderResult().isSuccess()) {
+ sendError(ctx, BAD_REQUEST);
+ return;
+ }
+
+ if (request.method() != GET) {
+ sendError(ctx, METHOD_NOT_ALLOWED);
+ return;
+ }
+
+ final String uri = request.uri().substring(1);
+ logger.info("Received download request through url {}", uri);
+
+ HttpTransferRequest httpTransferRequest = transferRequestsStore.getDownloadRequest(uri);
+
+ if (httpTransferRequest == null) {
+ logger.error("Couldn't find transfer request for uri {}", uri);
+ sendError(ctx, NOT_FOUND);
+ return;
+ }
+
+ Connector connector = httpTransferRequest.getOtherConnector();
+ MetadataCollector metadataCollector = httpTransferRequest.getOtherMetadataCollector();
+
+ ConnectorParams params = httpTransferRequest.getConnectorParams();
+
+ connector.init(params.getStorageId(), params.getCredentialToken(), params.getResourceServiceHost(),
+ params.getResourceServicePort(), params.getSecretServiceHost(), params.getSecretServicePort());
+
+ metadataCollector.init(params.getResourceServiceHost(), params.getResourceServicePort(),
+ params.getSecretServiceHost(), params.getSecretServicePort());
+
+ Boolean available = metadataCollector.isAvailable(params.getStorageId(),
+ httpTransferRequest.getTargetResourcePath(),
+ params.getCredentialToken());
+
+
+ if (!available) {
+ logger.error("File {} is not available in store {}", httpTransferRequest.getTargetResourcePath(), params.getStorageId());
+ sendError(ctx, NOT_FOUND);
+ return;
+ }
+
+ FileResourceMetadata fileResourceMetadata = metadataCollector.getFileResourceMetadata(params.getStorageId(),
+ httpTransferRequest.getTargetResourcePath(),
+ params.getCredentialToken());
+
+ long fileLength = fileResourceMetadata.getResourceSize();
+
+ HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
+ HttpUtil.setContentLength(response, fileLength);
+ setContentTypeHeader(response, httpTransferRequest.getTargetResourcePath());
+
+ if (HttpUtil.isKeepAlive(request)) {
+ response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
+ }
+
+ // Write the initial line and the header.
+ ctx.write(response);
+
+ // Write the content.
+ ChannelFuture sendFileFuture;
+ ChannelFuture lastContentFuture;
+
+ ConnectorContext connectorContext = new ConnectorContext();
+ connectorContext.setStreamBuffer(new DoubleStreamingBuffer());
+ connectorContext.setTransferId(uri);
+ connectorContext.setMetadata(new FileResourceMetadata()); // TODO Resolve
+
+ TransferTask pullTask = new TransferTask(connector, connectorContext, httpTransferRequest.getTargetResourcePath());
+
+ // TODO aggregate pullStatusFuture and sendFileFuture for keepalive test
+ Future<Integer> pullStatusFuture = executor.submit(pullTask);
+
+ sendFileFuture = ctx.writeAndFlush(new HttpChunkedInput(new ChunkedStream(connectorContext.getStreamBuffer().getInputStream())),
+ ctx.newProgressivePromise());
+
+ // HttpChunkedInput will write the end marker (LastHttpContent) for us.
+ lastContentFuture = sendFileFuture;
+
+ sendFileFuture.addListener(new ChannelProgressiveFutureListener() {
+ @Override
+ public void operationProgressed(ChannelProgressiveFuture future, long progress, long total) {
+ if (total < 0) { // total unknown
+ System.err.println(future.channel() + " Transfer progress: " + progress);
+ } else {
+ System.err.println(future.channel() + " Transfer progress: " + progress + " / " + total);
+ }
+ }
+
+ @Override
+ public void operationComplete(ChannelProgressiveFuture future) {
+ System.err.println(future.channel() + " Transfer complete.");
+ }
+ });
+
+ // Decide whether to close the connection or not.
+ if (!HttpUtil.isKeepAlive(request)) {
+ // Close the connection when the whole content is written out.
+ lastContentFuture.addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ cause.printStackTrace();
+ if (ctx.channel().isActive()) {
+ sendError(ctx, INTERNAL_SERVER_ERROR);
+ }
+ }
+
+ private static void sendRedirect(ChannelHandlerContext ctx, String newUri) {
+ FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, FOUND);
+ response.headers().set(HttpHeaderNames.LOCATION, newUri);
+
+ // Close the connection as soon as the error message is sent.
+ ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ }
+
+ private static void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
+ FullHttpResponse response = new DefaultFullHttpResponse(
+ HTTP_1_1, status, Unpooled.copiedBuffer("Failure: " + status + "\r\n", CharsetUtil.UTF_8));
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
+
+ // Close the connection as soon as the error message is sent.
+ ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+ }
+
+ private static void setContentTypeHeader(HttpResponse response, String path) {
+ MimetypesFileTypeMap mimeTypesMap = new MimetypesFileTypeMap();
+ response.headers().set(HttpHeaderNames.CONTENT_TYPE, path);
+ }
+}
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServerInitializer.java b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServerInitializer.java
new file mode 100644
index 0000000..f94d067
--- /dev/null
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpServerInitializer.java
@@ -0,0 +1,49 @@
+/*
+ * 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.agent.http;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.stream.ChunkedWriteHandler;
+
+public class HttpServerInitializer extends ChannelInitializer<SocketChannel> {
+
+ private final SslContext sslCtx;
+ private final HttpTransferRequestsStore transferRequestsStore;
+
+ public HttpServerInitializer(SslContext sslCtx, HttpTransferRequestsStore transferRequestsStore) {
+ this.sslCtx = sslCtx;
+ this.transferRequestsStore = transferRequestsStore;
+ }
+
+ @Override
+ public void initChannel(SocketChannel ch) {
+ ChannelPipeline pipeline = ch.pipeline();
+ if (sslCtx != null) {
+ pipeline.addLast(sslCtx.newHandler(ch.alloc()));
+ }
+ pipeline.addLast(new HttpServerCodec());
+ pipeline.addLast(new HttpObjectAggregator(65536));
+ pipeline.addLast(new ChunkedWriteHandler());
+ pipeline.addLast(new HttpServerHandler(transferRequestsStore));
+ }
+}
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpTransferRequest.java b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpTransferRequest.java
new file mode 100644
index 0000000..dbc807f
--- /dev/null
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpTransferRequest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.agent.http;
+
+import org.apache.airavata.mft.core.api.Connector;
+import org.apache.airavata.mft.core.api.MetadataCollector;
+
+public class HttpTransferRequest {
+ private Connector otherConnector;
+ private MetadataCollector otherMetadataCollector;
+ private ConnectorParams connectorParams;
+ private String targetResourcePath;
+
+ public Connector getOtherConnector() {
+ return otherConnector;
+ }
+
+ public HttpTransferRequest setOtherConnector(Connector otherConnector) {
+ this.otherConnector = otherConnector;
+ return this;
+ }
+
+ public MetadataCollector getOtherMetadataCollector() {
+ return otherMetadataCollector;
+ }
+
+ public HttpTransferRequest setOtherMetadataCollector(MetadataCollector otherMetadataCollector) {
+ this.otherMetadataCollector = otherMetadataCollector;
+ return this;
+ }
+
+ public String getTargetResourcePath() {
+ return targetResourcePath;
+ }
+
+ public HttpTransferRequest setTargetResourcePath(String targetResourcePath) {
+ this.targetResourcePath = targetResourcePath;
+ return this;
+ }
+
+ public ConnectorParams getConnectorParams() {
+ return connectorParams;
+ }
+
+ public HttpTransferRequest setConnectorParams(ConnectorParams connectorParams) {
+ this.connectorParams = connectorParams;
+ return this;
+ }
+}
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpTransferRequestsStore.java b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpTransferRequestsStore.java
new file mode 100644
index 0000000..fc38701
--- /dev/null
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/http/HttpTransferRequestsStore.java
@@ -0,0 +1,54 @@
+/*
+ * 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.agent.http;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+public class HttpTransferRequestsStore {
+
+ final private Map<String, HttpTransferRequest> downloadRequestStore = new HashMap<>();
+ final private Map<String, HttpTransferRequest> uploadRequestStore = new HashMap<>();
+
+ public String addDownloadRequest(HttpTransferRequest request) {
+ String randomUrl = UUID.randomUUID().toString();
+ downloadRequestStore.put(randomUrl, request);
+ return randomUrl;
+ }
+
+ public HttpTransferRequest getDownloadRequest(String url) {
+
+ //TODO Need to block concurrent calls to same url as connectors are not thread safe
+ HttpTransferRequest request = downloadRequestStore.get(url);
+ return request;
+ }
+
+ public String addUploadRequest(HttpTransferRequest request) {
+ String randomUrl = UUID.randomUUID().toString();
+ uploadRequestStore.put(randomUrl, request);
+ return randomUrl;
+ }
+
+ public HttpTransferRequest getUploadRequest(String url) {
+
+ //TODO Need to block concurrent calls to same url as connectors are not thread safe
+ HttpTransferRequest request = downloadRequestStore.get(url);
+ return request;
+ }
+}
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/rpc/RPCParser.java b/agent/src/main/java/org/apache/airavata/mft/agent/rpc/RPCParser.java
index f136704..dbcc48a 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/rpc/RPCParser.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/rpc/RPCParser.java
@@ -20,12 +20,18 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.airavata.mft.admin.models.rpc.SyncRPCRequest;
import org.apache.airavata.mft.admin.models.rpc.SyncRPCResponse;
+import org.apache.airavata.mft.agent.http.ConnectorParams;
+import org.apache.airavata.mft.agent.http.HttpTransferRequest;
+import org.apache.airavata.mft.agent.http.HttpTransferRequestsStore;
+import org.apache.airavata.mft.core.ConnectorResolver;
import org.apache.airavata.mft.core.DirectoryResourceMetadata;
import org.apache.airavata.mft.core.FileResourceMetadata;
import org.apache.airavata.mft.core.MetadataCollectorResolver;
+import org.apache.airavata.mft.core.api.Connector;
import org.apache.airavata.mft.core.api.MetadataCollector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
import java.util.Optional;
@@ -45,6 +51,9 @@
@org.springframework.beans.factory.annotation.Value("${secret.service.port}")
private int secretServicePort;
+ @Autowired
+ private HttpTransferRequestsStore httpTransferRequestsStore;
+
public String resolveRPCRequest(SyncRPCRequest request) throws Exception {
// TODO implement using the reflection
ObjectMapper mapper = new ObjectMapper();
@@ -64,6 +73,7 @@
return mapper.writeValueAsString(fileResourceMetadata);
}
break;
+
case "getChildFileResourceMetadata":
resourceId = request.getParameters().get("resourceId");
resourceType = request.getParameters().get("resourceType");
@@ -79,6 +89,7 @@
return mapper.writeValueAsString(fileResourceMetadata);
}
break;
+
case "getDirectoryResourceMetadata":
resourceId = request.getParameters().get("resourceId");
resourceType = request.getParameters().get("resourceType");
@@ -93,6 +104,7 @@
return mapper.writeValueAsString(dirResourceMetadata);
}
break;
+
case "getChildDirectoryResourceMetadata":
resourceId = request.getParameters().get("resourceId");
resourceType = request.getParameters().get("resourceType");
@@ -108,6 +120,32 @@
return mapper.writeValueAsString(dirResourceMetadata);
}
break;
+
+ case "submitHttpDownload":
+ String storeId = request.getParameters().get("storeId");
+ String sourcePath = request.getParameters().get("sourcePath");
+ String sourceToken = request.getParameters().get("sourceToken");
+ String storeType = request.getParameters().get("storeType");
+ mftAuthorizationToken = request.getParameters().get("mftAuthorizationToken");
+
+ metadataCollectorOp = MetadataCollectorResolver.resolveMetadataCollector(storeType);
+ Optional<Connector> connectorOp = ConnectorResolver.resolveConnector(storeType, "IN");
+
+ if (metadataCollectorOp.isPresent() && connectorOp.isPresent()) {
+ HttpTransferRequest transferRequest = new HttpTransferRequest();
+ transferRequest.setConnectorParams(new ConnectorParams()
+ .setResourceServiceHost(resourceServiceHost)
+ .setResourceServicePort(resourceServicePort)
+ .setSecretServiceHost(secretServiceHost)
+ .setSecretServicePort(secretServicePort)
+ .setStorageId(storeId).setCredentialToken(sourceToken));
+ transferRequest.setTargetResourcePath(sourcePath);
+ transferRequest.setOtherMetadataCollector(metadataCollectorOp.get());
+ transferRequest.setOtherConnector(connectorOp.get());
+ String url = httpTransferRequestsStore.addDownloadRequest(transferRequest);
+ return url;
+ }
+ break;
}
logger.error("Unknown method type specified {}", request.getMethod());
throw new Exception("Unknown method " + request.getMethod());
diff --git a/api/service/src/main/java/org/apache/airavata/mft/api/handler/MFTApiHandler.java b/api/service/src/main/java/org/apache/airavata/mft/api/handler/MFTApiHandler.java
index 108bd30..1a6a8b9 100644
--- a/api/service/src/main/java/org/apache/airavata/mft/api/handler/MFTApiHandler.java
+++ b/api/service/src/main/java/org/apache/airavata/mft/api/handler/MFTApiHandler.java
@@ -92,6 +92,51 @@
}
@Override
+ public void submitHttpUpload(HttpUploadApiRequest request, StreamObserver<HttpUploadApiResponse> responseObserver) {
+ super.submitHttpUpload(request, responseObserver);
+ }
+
+ @Override
+ public void submitHttpDownload(HttpDownloadApiRequest request, StreamObserver<HttpDownloadApiResponse> responseObserver) {
+ try {
+
+ // TODO : Automatically derive agent if the target agent is empty
+ SyncRPCRequest.SyncRPCRequestBuilder requestBuilder = SyncRPCRequest.SyncRPCRequestBuilder.builder()
+ .withAgentId(request.getTargetAgent())
+ .withMessageId(UUID.randomUUID().toString())
+ .withMethod("submitHttpDownload")
+ .withParameter("storeId", request.getSourceStoreId())
+ .withParameter("sourcePath", request.getSourcePath())
+ .withParameter("sourceToken", request.getSourceToken())
+ .withParameter("storeType", request.getSourceType())
+ .withParameter("mftAuthorizationToken", request.getMftAuthorizationToken());
+
+ SyncRPCResponse rpcResponse = agentRPCClient.sendSyncRequest(requestBuilder.build());
+
+ switch (rpcResponse.getResponseStatus()) {
+ case SUCCESS:
+ String url = rpcResponse.getResponseAsStr();
+ HttpDownloadApiResponse downloadResponse = HttpDownloadApiResponse.newBuilder()
+ .setUrl(url)
+ .setTargetAgent(request.getTargetAgent()).build();
+ responseObserver.onNext(downloadResponse);
+ responseObserver.onCompleted();
+ return;
+ case FAIL:
+ logger.error("Errored while processing the download request to resource {} in store {}. Error msg : {}",
+ request.getSourcePath(), request.getSourceStoreId(), rpcResponse.getErrorAsStr());
+ responseObserver.onError(new Exception("Errored while processing the the fetch file metadata response. Error msg : " +
+ rpcResponse.getErrorAsStr()));
+ }
+
+ } catch (Exception e) {
+ logger.error("Error while submitting http download request to path {} in store {}",
+ request.getSourcePath(), request.getSourceStoreId() , e);
+ responseObserver.onError(new Exception("Failed to submit http download request", e));
+ }
+ }
+
+ @Override
public void getTransferStates(TransferStateApiRequest request, StreamObserver<TransferStateApiResponse> responseObserver) {
try {
List<TransferState> states = mftConsulClient.getTransferStates(request.getTransferId());
diff --git a/api/stub/src/main/proto/MFTApi.proto b/api/stub/src/main/proto/MFTApi.proto
index dc003ef..891413c 100644
--- a/api/stub/src/main/proto/MFTApi.proto
+++ b/api/stub/src/main/proto/MFTApi.proto
@@ -7,25 +7,55 @@
import "google/protobuf/empty.proto";
message TransferApiRequest {
- string sourceId = 1;
- string sourceType = 2;
- string sourceToken = 3;
- string sourceResourceBackend = 4;
- string sourceCredentialBackend = 5;
- string destinationId = 6;
- string destinationType = 7;
- string destinationToken = 8;
- string destResourceBackend = 9;
- string destCredentialBackend = 10;
- bool affinityTransfer = 11;
- map<string, int32> targetAgents = 12;
- string mftAuthorizationToken = 13;
+ string sourceStorageId = 1;
+ string sourcePath = 2;
+ string sourceType = 3;
+ string sourceToken = 4;
+ string sourceResourceBackend = 5;
+ string sourceCredentialBackend = 6;
+ string destinationStorageId = 7;
+ string destinationPath = 8;
+ string destinationType = 9;
+ string destinationToken = 10;
+ string destResourceBackend = 11;
+ string destCredentialBackend = 12;
+ bool affinityTransfer = 13;
+ map<string, int32> targetAgents = 14;
+ string mftAuthorizationToken = 15;
}
message TransferApiResponse {
string transferId = 1;
}
+message HttpUploadApiRequest {
+ string destinationStoreId = 1;
+ string destinationPath = 2;
+ string destinationToken = 3;
+ string destinationType = 4;
+ string targetAgent = 5;
+ string mftAuthorizationToken = 6;
+}
+
+message HttpUploadApiResponse {
+ string url = 1;
+ string targetAgent = 2;
+}
+
+message HttpDownloadApiRequest {
+ string sourceStoreId = 1;
+ string sourcePath = 2;
+ string sourceToken = 3;
+ string sourceType = 4;
+ string targetAgent = 5;
+ string mftAuthorizationToken = 6;
+}
+
+message HttpDownloadApiResponse {
+ string url = 1;
+ string targetAgent = 2;
+}
+
message TransferStateApiRequest {
string transferId = 1;
string mftAuthorizationToken = 2;
@@ -93,6 +123,18 @@
};
}
+ rpc submitHttpUpload(HttpUploadApiRequest) returns (HttpUploadApiResponse) {
+ option (google.api.http) = {
+ post: "/v1.0/api/http-upload"
+ };
+ }
+
+ rpc submitHttpDownload(HttpDownloadApiRequest) returns (HttpDownloadApiResponse) {
+ option (google.api.http) = {
+ post: "/v1.0/api/http-download"
+ };
+ }
+
rpc getTransferStates(TransferStateApiRequest) returns (stream TransferStateApiResponse) {
option (google.api.http) = {
get: "/v1.0/api/transfer/states"
diff --git a/controller/src/main/java/org/apache/airavata/mft/controller/MFTController.java b/controller/src/main/java/org/apache/airavata/mft/controller/MFTController.java
index 0b0f2cd..ec4240c 100644
--- a/controller/src/main/java/org/apache/airavata/mft/controller/MFTController.java
+++ b/controller/src/main/java/org/apache/airavata/mft/controller/MFTController.java
@@ -221,12 +221,14 @@
private TransferCommand convertRequestToCommand(String transferId, TransferRequest transferRequest) {
TransferCommand transferCommand = new TransferCommand();
- transferCommand.setSourceId(transferRequest.getSourceId())
+ transferCommand.setSourceStorageId(transferRequest.getSourceStorageId())
+ .setSourcePath(transferRequest.getSourcePath())
.setSourceToken(transferRequest.getSourceToken())
.setSourceType(transferRequest.getSourceType())
.setSourceResourceBackend(transferRequest.getSourceResourceBackend())
.setSourceCredentialBackend(transferRequest.getSourceCredentialBackend())
- .setDestinationId(transferRequest.getDestinationId())
+ .setDestinationStorageId(transferRequest.getDestinationStorageId())
+ .setDestinationPath(transferRequest.getDestinationPath())
.setDestinationToken(transferRequest.getDestinationToken())
.setDestinationType(transferRequest.getDestinationType())
.setDestResourceBackend(transferRequest.getDestResourceBackend())
diff --git a/core/src/main/java/org/apache/airavata/mft/core/DirectoryResourceMetadata.java b/core/src/main/java/org/apache/airavata/mft/core/DirectoryResourceMetadata.java
index 5eb9813..5ff3b5f 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/DirectoryResourceMetadata.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/DirectoryResourceMetadata.java
@@ -26,8 +26,6 @@
private long createdTime;
private long updateTime;
private String resourcePath;
- private String parentResourceId;
- private String parentResourceType;
private List<DirectoryResourceMetadata> directories = new ArrayList<>();
private List<FileResourceMetadata> files = new ArrayList<>();
private boolean lazyInitialized = true;
@@ -68,24 +66,6 @@
return this;
}
- public String getParentResourceId() {
- return parentResourceId;
- }
-
- public DirectoryResourceMetadata setParentResourceId(String parentResourceId) {
- this.parentResourceId = parentResourceId;
- return this;
- }
-
- public String getParentResourceType() {
- return parentResourceType;
- }
-
- public DirectoryResourceMetadata setParentResourceType(String parentResourceType) {
- this.parentResourceType = parentResourceType;
- return this;
- }
-
public List<DirectoryResourceMetadata> getDirectories() {
return directories;
}
@@ -119,8 +99,6 @@
private long createdTime;
private long updateTime;
private String resourcePath;
- private String parentResourceId;
- private String parentResourceType;
private List<DirectoryResourceMetadata> directories = new ArrayList<>();
private List<FileResourceMetadata> files = new ArrayList<>();
private boolean lazyInitialized = true;
@@ -152,16 +130,6 @@
return this;
}
- public Builder withParentResourceId(String parentResourceId) {
- this.parentResourceId = parentResourceId;
- return this;
- }
-
- public Builder withParentResourceType(String parentResourceType) {
- this.parentResourceType = parentResourceType;
- return this;
- }
-
public Builder withDirectories(List<DirectoryResourceMetadata> directories) {
this.directories = directories;
return this;
@@ -193,8 +161,6 @@
directoryResourceMetadata.setCreatedTime(createdTime);
directoryResourceMetadata.setUpdateTime(updateTime);
directoryResourceMetadata.setResourcePath(resourcePath);
- directoryResourceMetadata.setParentResourceId(parentResourceId);
- directoryResourceMetadata.setParentResourceType(parentResourceType);
directoryResourceMetadata.setDirectories(directories);
directoryResourceMetadata.setFiles(files);
directoryResourceMetadata.setLazyInitialized(lazyInitialized);
diff --git a/core/src/main/java/org/apache/airavata/mft/core/FileResourceMetadata.java b/core/src/main/java/org/apache/airavata/mft/core/FileResourceMetadata.java
index 0b2372e..d61d743 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/FileResourceMetadata.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/FileResourceMetadata.java
@@ -25,8 +25,6 @@
private long updateTime;
private String md5sum;
private String resourcePath;
- private String parentResourceId;
- private String parentResourceType;
public String getFriendlyName() {
return friendlyName;
@@ -82,24 +80,6 @@
return this;
}
- public String getParentResourceId() {
- return parentResourceId;
- }
-
- public FileResourceMetadata setParentResourceId(String parentResourceId) {
- this.parentResourceId = parentResourceId;
- return this;
- }
-
- public String getParentResourceType() {
- return parentResourceType;
- }
-
- public FileResourceMetadata setParentResourceType(String parentResourceType) {
- this.parentResourceType = parentResourceType;
- return this;
- }
-
public static final class Builder {
private String friendlyName;
private long resourceSize;
@@ -107,8 +87,6 @@
private long updateTime;
private String md5sum;
private String resourcePath;
- private String parentResourceId;
- private String parentResourceType;
private Builder() {
}
@@ -147,15 +125,6 @@
return this;
}
- public Builder withParentResourceId(String parentResourceId) {
- this.parentResourceId = parentResourceId;
- return this;
- }
-
- public Builder withParentResourceType(String parentResourceType) {
- this.parentResourceType = parentResourceType;
- return this;
- }
public FileResourceMetadata build() {
FileResourceMetadata fileResourceMetadata = new FileResourceMetadata();
@@ -165,8 +134,6 @@
fileResourceMetadata.setUpdateTime(updateTime);
fileResourceMetadata.setMd5sum(md5sum);
fileResourceMetadata.setResourcePath(resourcePath);
- fileResourceMetadata.setParentResourceId(parentResourceId);
- fileResourceMetadata.setParentResourceType(parentResourceType);
return fileResourceMetadata;
}
}
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
index b90f4cc..860e9f7 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/TransferTask.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/TransferTask.java
@@ -17,7 +17,6 @@
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;
@@ -26,15 +25,17 @@
private Connector connector;
private ConnectorContext context;
+ private String resourcePath;
- public TransferTask(Connector connector, ConnectorContext context) {
+ public TransferTask(Connector connector, ConnectorContext context, String resourcePath) {
this.connector = connector;
this.context = context;
+ this.resourcePath = resourcePath;
}
@Override
public Integer call() throws Exception {
- this.connector.startStream(context);
+ this.connector.startStream(resourcePath, context);
return 0;
}
}
diff --git a/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
index 772d90e..8264d58 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/api/Connector.java
@@ -20,8 +20,8 @@
import org.apache.airavata.mft.core.ConnectorContext;
public interface Connector {
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort,
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort,
String secretServiceHost, int secretServicePort) throws Exception;
public void destroy();
- void startStream(ConnectorContext context) throws Exception;
+ void startStream(String targetPath, ConnectorContext context) throws Exception;
}
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
index e784867..38a5529 100644
--- 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
@@ -43,17 +43,15 @@
public FileResourceMetadata getFileResourceMetadata(String resourceId, String credentialToken) throws Exception;
/**
- * Fetches a metadata of given File Resource inside a registered directory resource. Target file might be living in
- * multiple level below the parent directory
+ * Fetches a metadata of given File Resource
*
- * @param parentResourceId parent directory resource id
- * @param resourcePath path of the target resource. This should be a child path of the parent resource
+ * @param storageId id of the storage resource
+ * @param resourcePath resource path
* @param credentialToken credential token for the resource
* @return an object of {@link FileResourceMetadata}
- * @throws Exception if the parent resource is not a Directory resource or the target resource is not a File Resource type
- * or the resource can't be fetched from the resource service
+ * @throws Exception if the resource id is not a File Resource type or the resource can't be fetched from the resource service
*/
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception;
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception;
/**
* Fetches a metadata of given Directory Resource
@@ -66,17 +64,14 @@
public DirectoryResourceMetadata getDirectoryResourceMetadata(String resourceId, String credentialToken) throws Exception;
/**
- * Fetches a metadata of given Directory Resource inside a registered directory resource. Target directory might be living in
- * multiple level below the parent directory
+ * Fetches a metadata of given Directory Resource
*
- * @param parentResourceId parent directory resource id
- * @param resourcePath path of the target resource. This should be a child path of the parent resource
- * @param credentialToken credential token for the resource
+ * @param storageId id of the storage resource
+ * @param resourcePath resource path
* @return an object of {@link DirectoryResourceMetadata}
- * @throws Exception if the parent resource is not a Directory resource or the target resource is not a Directory Resource type
- * or the resource can't be fetched from the resource service
+ * @throws Exception if the resource id is not a Directory Resource type or the resource can't be fetched from the resource service
*/
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception;
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception;
/**
* Check whether the resource is available in the actual storage
@@ -87,4 +82,15 @@
* @throws Exception if the resource details can not be fetched from the resource service
*/
public Boolean isAvailable(String resourceId, String credentialToken) throws Exception;
+
+ /**
+ * Check whether the resource is available in the actual storage
+ *
+ * @param storageId id of the storage
+ * @param resourcePath resource path
+ * @param credentialToken credential token for the resource
+ * @return true of the resource is available false otherwise
+ * @throws Exception if the resource details can not be fetched from the resource service
+ */
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception;
}
diff --git a/examples/src/main/java/org/apache/airavata/mft/examples/http/DownloadExample.java b/examples/src/main/java/org/apache/airavata/mft/examples/http/DownloadExample.java
new file mode 100644
index 0000000..93b5d83
--- /dev/null
+++ b/examples/src/main/java/org/apache/airavata/mft/examples/http/DownloadExample.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.airavata.mft.examples.http;
+
+import org.apache.airavata.mft.api.client.MFTApiClient;
+import org.apache.airavata.mft.api.service.HttpDownloadApiRequest;
+import org.apache.airavata.mft.api.service.HttpDownloadApiResponse;
+import org.apache.airavata.mft.api.service.MFTApiServiceGrpc;
+
+public class DownloadExample {
+ public static void main(String args[]) {
+ MFTApiServiceGrpc.MFTApiServiceBlockingStub client = MFTApiClient.buildClient("localhost", 7004);
+ HttpDownloadApiResponse httpDownloadApiResponse = client.submitHttpDownload(HttpDownloadApiRequest.newBuilder()
+ .setTargetAgent("agent0")
+ .setSourcePath("/tmp/a.txt")
+ .setSourceStoreId("remote-ssh-storage")
+ .setSourceToken("local-ssh-cred")
+ .setSourceType("SCP")
+ .setMftAuthorizationToken("")
+ .build());
+
+ System.out.println(httpDownloadApiResponse);
+ }
+}
diff --git a/examples/src/main/java/org/apache/airavata/mft/examples/transfer/LocalExample.java b/examples/src/main/java/org/apache/airavata/mft/examples/transfer/LocalExample.java
index 53bfa71..af73d7c 100644
--- a/examples/src/main/java/org/apache/airavata/mft/examples/transfer/LocalExample.java
+++ b/examples/src/main/java/org/apache/airavata/mft/examples/transfer/LocalExample.java
@@ -26,18 +26,22 @@
public static void main(String args[]) throws Exception {
MFTApiServiceGrpc.MFTApiServiceBlockingStub client = MFTApiClient.buildClient("localhost", 7004);
- String sourceId = "remote-ssh-resource2";
+ String sourceStorageId = "remote-ssh-storage";
+ String sourceResourcePath = "/tmp/1mb.txt";
String sourceToken = "local-ssh-cred";
- String destId = "10mb-file";
+ String destStorageId = "local-storage-1";
+ String destResourcePath = "/tmp/1mb-copy.txt";
String destToken = "";
String mftAuthorizationToken = "43ff79ac-e4f2-473c-9ea1-04eee9509a53";
TransferApiRequest request = TransferApiRequest.newBuilder()
.setMftAuthorizationToken(mftAuthorizationToken)
- .setSourceId(sourceId)
+ .setSourceStorageId(sourceStorageId)
+ .setSourcePath(sourceResourcePath)
.setSourceToken(sourceToken)
.setSourceType("SCP")
- .setDestinationId(destId)
+ .setDestinationStorageId(destStorageId)
+ .setDestinationPath(destResourcePath)
.setDestinationToken(destToken)
.setDestinationType("LOCAL")
.setAffinityTransfer(false).build();
diff --git a/examples/src/main/java/org/apache/airavata/mft/examples/transfer/S3Example.java b/examples/src/main/java/org/apache/airavata/mft/examples/transfer/S3Example.java
index 8443cf2..9df81e2 100644
--- a/examples/src/main/java/org/apache/airavata/mft/examples/transfer/S3Example.java
+++ b/examples/src/main/java/org/apache/airavata/mft/examples/transfer/S3Example.java
@@ -26,18 +26,22 @@
public static void main(String args[]) throws Exception {
MFTApiServiceGrpc.MFTApiServiceBlockingStub client = MFTApiClient.buildClient("localhost", 7004);
- String sourceId = "remote-ssh-resource2";
- String sourceToken = "local-ssh-cred";
- String destId = "s3-file";
+ String sourceStorageId = "remote-ssh-storage";
+ String sourceResourcePath = "/tmp/1mb.txt";
+ String sourceToken = "ssh-cred";
+ String destStorageId = "s3-storage-1";
+ String destResourcePath = "1mb-copy.txt";
String destToken = "s3-cred";
String mftAuthorizationToken = "43ff79ac-e4f2-473c-9ea1-04eee9509a53";
TransferApiRequest request = TransferApiRequest.newBuilder()
.setMftAuthorizationToken(mftAuthorizationToken)
- .setSourceId(sourceId)
+ .setSourceStorageId(sourceStorageId)
+ .setSourcePath(sourceResourcePath)
.setSourceToken(sourceToken)
.setSourceType("SCP")
- .setDestinationId(destId)
+ .setDestinationStorageId(destStorageId)
+ .setDestinationPath(destResourcePath)
.setDestinationToken(destToken)
.setDestinationType("S3")
.setAffinityTransfer(false).build();
diff --git a/examples/src/main/java/org/apache/airavata/mft/examples/transfer/SCPExample.java b/examples/src/main/java/org/apache/airavata/mft/examples/transfer/SCPExample.java
index 5977598..72ffac4 100644
--- a/examples/src/main/java/org/apache/airavata/mft/examples/transfer/SCPExample.java
+++ b/examples/src/main/java/org/apache/airavata/mft/examples/transfer/SCPExample.java
@@ -26,18 +26,22 @@
public static void main(String args[]) throws Exception {
MFTApiServiceGrpc.MFTApiServiceBlockingStub client = MFTApiClient.buildClient("localhost", 7004);
- String sourceId = "remote-ssh-resource2";
- String sourceToken = "local-ssh-cred";
- String destId = "remote-ssh-resource";
- String destToken = "local-ssh-cred";
+ String sourceStorageId = "remote-ssh-storage-1";
+ String sourceResourcePath = "/tmp/1mb.txt";
+ String sourceToken = "ssh-cred-1";
+ String destStorageId = "remote-ssh-storage-2";
+ String destResourcePath = "/tmp/1mb-copy.txt";
+ String destToken = "ssh-cred-2";
String mftAuthorizationToken = "43ff79ac-e4f2-473c-9ea1-04eee9509a53";
TransferApiRequest request = TransferApiRequest.newBuilder()
.setMftAuthorizationToken(mftAuthorizationToken)
- .setSourceId(sourceId)
+ .setSourceStorageId(sourceStorageId)
+ .setSourcePath(sourceResourcePath)
.setSourceToken(sourceToken)
.setSourceType("SCP")
- .setDestinationId(destId)
+ .setDestinationStorageId(destStorageId)
+ .setDestinationPath(destResourcePath)
.setDestinationToken(destToken)
.setDestinationType("SCP")
.setAffinityTransfer(false).build();
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/ResourceBackend.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/ResourceBackend.java
index 2e3a550..d924995 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/ResourceBackend.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/ResourceBackend.java
@@ -18,13 +18,19 @@
package org.apache.airavata.mft.resource.server.backend;
import org.apache.airavata.mft.resource.stubs.azure.resource.*;
+import org.apache.airavata.mft.resource.stubs.azure.storage.*;
import org.apache.airavata.mft.resource.stubs.box.resource.*;
+import org.apache.airavata.mft.resource.stubs.box.storage.*;
import org.apache.airavata.mft.resource.stubs.dropbox.resource.*;
+import org.apache.airavata.mft.resource.stubs.dropbox.storage.*;
import org.apache.airavata.mft.resource.stubs.ftp.resource.*;
import org.apache.airavata.mft.resource.stubs.ftp.storage.*;
import org.apache.airavata.mft.resource.stubs.gcs.resource.*;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.*;
import org.apache.airavata.mft.resource.stubs.local.resource.*;
+import org.apache.airavata.mft.resource.stubs.local.storage.*;
import org.apache.airavata.mft.resource.stubs.s3.resource.*;
+import org.apache.airavata.mft.resource.stubs.s3.storage.*;
import org.apache.airavata.mft.resource.stubs.scp.resource.*;
import org.apache.airavata.mft.resource.stubs.scp.storage.*;
@@ -45,31 +51,61 @@
public boolean updateSCPResource(SCPResourceUpdateRequest request) throws Exception;
public boolean deleteSCPResource(SCPResourceDeleteRequest request) throws Exception;
+ public Optional<LocalStorage> getLocalStorage(LocalStorageGetRequest request) throws Exception;
+ public LocalStorage createLocalStorage(LocalStorageCreateRequest request) throws Exception;
+ public boolean updateLocalStorage(LocalStorageUpdateRequest request) throws Exception;
+ public boolean deleteLocalStorage(LocalStorageDeleteRequest request) throws Exception;
+
public Optional<LocalResource> getLocalResource(LocalResourceGetRequest request) throws Exception;
public LocalResource createLocalResource(LocalResourceCreateRequest request) throws Exception;
public boolean updateLocalResource(LocalResourceUpdateRequest request) throws Exception;
public boolean deleteLocalResource(LocalResourceDeleteRequest request) throws Exception;
+ public Optional<S3Storage> getS3Storage(S3StorageGetRequest request) throws Exception;
+ public S3Storage createS3Storage(S3StorageCreateRequest request) throws Exception;
+ public boolean updateS3Storage(S3StorageUpdateRequest request) throws Exception;
+ public boolean deleteS3Storage(S3StorageDeleteRequest request) throws Exception;
+
public Optional<S3Resource> getS3Resource(S3ResourceGetRequest request) throws Exception;
public S3Resource createS3Resource(S3ResourceCreateRequest request) throws Exception;
public boolean updateS3Resource(S3ResourceUpdateRequest request) throws Exception;
public boolean deleteS3Resource(S3ResourceDeleteRequest request) throws Exception;
+ public Optional<BoxStorage> getBoxStorage(BoxStorageGetRequest request) throws Exception;
+ public BoxStorage createBoxStorage(BoxStorageCreateRequest request) throws Exception;
+ public boolean updateBoxStorage(BoxStorageUpdateRequest request) throws Exception;
+ public boolean deleteBoxStorage(BoxStorageDeleteRequest request) throws Exception;
+
public Optional<BoxResource> getBoxResource(BoxResourceGetRequest request) throws Exception;
public BoxResource createBoxResource(BoxResourceCreateRequest request) throws Exception;
public boolean updateBoxResource(BoxResourceUpdateRequest request) throws Exception;
public boolean deleteBoxResource(BoxResourceDeleteRequest request) throws Exception;
+ public Optional<AzureStorage> getAzureStorage(AzureStorageGetRequest request) throws Exception;
+ public AzureStorage createAzureStorage(AzureStorageCreateRequest request) throws Exception;
+ public boolean updateAzureStorage(AzureStorageUpdateRequest request) throws Exception;
+ public boolean deleteAzureStorage(AzureStorageDeleteRequest request) throws Exception;
+
public Optional<AzureResource> getAzureResource(AzureResourceGetRequest request) throws Exception;
public AzureResource createAzureResource(AzureResourceCreateRequest request) throws Exception;
public boolean updateAzureResource(AzureResourceUpdateRequest request) throws Exception;
public boolean deleteAzureResource(AzureResourceDeleteRequest request) throws Exception;
+ public Optional<GCSStorage> getGCSStorage(GCSStorageGetRequest request) throws Exception;
+ public GCSStorage createGCSStorage(GCSStorageCreateRequest request) throws Exception;
+ public boolean updateGCSStorage(GCSStorageUpdateRequest request) throws Exception;
+ public boolean deleteGCSStorage(GCSStorageDeleteRequest request) throws Exception;
+
public Optional<GCSResource> getGCSResource(GCSResourceGetRequest request) throws Exception;
public GCSResource createGCSResource(GCSResourceCreateRequest request) throws Exception;
public boolean updateGCSResource(GCSResourceUpdateRequest request) throws Exception;
public boolean deleteGCSResource(GCSResourceDeleteRequest request) throws Exception;
+ public Optional<DropboxStorage> getDropboxStorage(DropboxStorageGetRequest request) throws Exception;
+ public DropboxStorage createDropboxStorage(DropboxStorageCreateRequest request) throws Exception;
+ public boolean updateDropboxStorage(DropboxStorageUpdateRequest request) throws Exception;
+ public boolean deleteDropboxStorage(DropboxStorageDeleteRequest request) throws Exception;
+
public Optional<DropboxResource> getDropboxResource(DropboxResourceGetRequest request) throws Exception;
public DropboxResource createDropboxResource(DropboxResourceCreateRequest request) throws Exception;
public boolean updateDropboxResource(DropboxResourceUpdateRequest request) throws Exception;
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/airavata/AiravataResourceBackend.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/airavata/AiravataResourceBackend.java
index 6299a07..a5a78f1 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/airavata/AiravataResourceBackend.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/airavata/AiravataResourceBackend.java
@@ -19,14 +19,20 @@
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.stubs.azure.resource.*;
+import org.apache.airavata.mft.resource.stubs.azure.storage.*;
import org.apache.airavata.mft.resource.stubs.box.resource.*;
+import org.apache.airavata.mft.resource.stubs.box.storage.*;
import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.resource.stubs.dropbox.resource.*;
+import org.apache.airavata.mft.resource.stubs.dropbox.storage.*;
import org.apache.airavata.mft.resource.stubs.ftp.resource.*;
import org.apache.airavata.mft.resource.stubs.ftp.storage.*;
import org.apache.airavata.mft.resource.stubs.gcs.resource.*;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.*;
import org.apache.airavata.mft.resource.stubs.local.resource.*;
+import org.apache.airavata.mft.resource.stubs.local.storage.*;
import org.apache.airavata.mft.resource.stubs.s3.resource.*;
+import org.apache.airavata.mft.resource.stubs.s3.storage.*;
import org.apache.airavata.mft.resource.stubs.scp.resource.*;
import org.apache.airavata.mft.resource.stubs.scp.storage.*;
import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
@@ -155,6 +161,26 @@
}
@Override
+ public Optional<LocalStorage> getLocalStorage(LocalStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public LocalStorage createLocalStorage(LocalStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateLocalStorage(LocalStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteLocalStorage(LocalStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<LocalResource> getLocalResource(LocalResourceGetRequest request) {
throw new UnsupportedOperationException("Operation is not supported in backend");
@@ -179,6 +205,26 @@
}
@Override
+ public Optional<S3Storage> getS3Storage(S3StorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public S3Storage createS3Storage(S3StorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateS3Storage(S3StorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteS3Storage(S3StorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<S3Resource> getS3Resource(S3ResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
@@ -203,6 +249,26 @@
}
@Override
+ public Optional<BoxStorage> getBoxStorage(BoxStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public BoxStorage createBoxStorage(BoxStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateBoxStorage(BoxStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteBoxStorage(BoxStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<BoxResource> getBoxResource(BoxResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
@@ -223,6 +289,26 @@
}
@Override
+ public Optional<AzureStorage> getAzureStorage(AzureStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public AzureStorage createAzureStorage(AzureStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateAzureStorage(AzureStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteAzureStorage(AzureStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<AzureResource> getAzureResource(AzureResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
@@ -243,6 +329,26 @@
}
@Override
+ public Optional<GCSStorage> getGCSStorage(GCSStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public GCSStorage createGCSStorage(GCSStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateGCSStorage(GCSStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteGCSStorage(GCSStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<GCSResource> getGCSResource(GCSResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
@@ -261,6 +367,27 @@
public boolean deleteGCSResource(GCSResourceDeleteRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
+
+ @Override
+ public Optional<DropboxStorage> getDropboxStorage(DropboxStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public DropboxStorage createDropboxStorage(DropboxStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateDropboxStorage(DropboxStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteDropboxStorage(DropboxStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
@Override
public Optional<DropboxResource> getDropboxResource(DropboxResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/file/FileBasedResourceBackend.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/file/FileBasedResourceBackend.java
index 3841409..eace41a 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/file/FileBasedResourceBackend.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/file/FileBasedResourceBackend.java
@@ -19,21 +19,21 @@
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.stubs.azure.resource.*;
-import org.apache.airavata.mft.resource.stubs.azure.storage.AzureStorage;
+import org.apache.airavata.mft.resource.stubs.azure.storage.*;
import org.apache.airavata.mft.resource.stubs.box.resource.*;
-import org.apache.airavata.mft.resource.stubs.box.storage.BoxStorage;
+import org.apache.airavata.mft.resource.stubs.box.storage.*;
import org.apache.airavata.mft.resource.stubs.common.DirectoryResource;
import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.resource.stubs.dropbox.resource.*;
-import org.apache.airavata.mft.resource.stubs.dropbox.storage.DropboxStorage;
+import org.apache.airavata.mft.resource.stubs.dropbox.storage.*;
import org.apache.airavata.mft.resource.stubs.ftp.resource.*;
import org.apache.airavata.mft.resource.stubs.ftp.storage.*;
import org.apache.airavata.mft.resource.stubs.gcs.resource.*;
-import org.apache.airavata.mft.resource.stubs.gcs.storage.GCSStorage;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.*;
import org.apache.airavata.mft.resource.stubs.local.resource.*;
-import org.apache.airavata.mft.resource.stubs.local.storage.LocalStorage;
+import org.apache.airavata.mft.resource.stubs.local.storage.*;
import org.apache.airavata.mft.resource.stubs.s3.resource.*;
-import org.apache.airavata.mft.resource.stubs.s3.storage.S3Storage;
+import org.apache.airavata.mft.resource.stubs.s3.storage.*;
import org.apache.airavata.mft.resource.stubs.scp.resource.*;
import org.apache.airavata.mft.resource.stubs.scp.storage.*;
import org.json.simple.JSONArray;
@@ -57,6 +57,9 @@
@org.springframework.beans.factory.annotation.Value("${file.backend.resource.file}")
private String resourceFile;
+ @org.springframework.beans.factory.annotation.Value("${file.backend.storage.file}")
+ private String storageFile;
+
@Override
public void init() {
logger.info("Initializing file based resource backend");
@@ -69,7 +72,33 @@
@Override
public Optional<SCPStorage> getSCPStorage(SCPStorageGetRequest request) throws Exception {
- throw new UnsupportedOperationException("Operation is not supported in backend");
+ InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(storageFile);
+
+ JSONParser jsonParser = new JSONParser();
+
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+
+ Object obj = jsonParser.parse(reader);
+
+ JSONArray resourceList = (JSONArray) obj;
+
+ List<SCPStorage> scpStorages = (List<SCPStorage>) resourceList.stream()
+ .filter(resource -> "SCP".equals(((JSONObject) resource).get("type").toString()))
+ .map(st -> {
+ JSONObject s = (JSONObject) st;
+
+ SCPStorage storage = SCPStorage.newBuilder()
+ .setStorageId(s.get("storageId").toString())
+ .setHost(s.get("host").toString())
+ .setUser(s.get("user").toString())
+ .setPort(Integer.parseInt(s.get("port").toString())).build();
+
+ return storage;
+
+ }).collect(Collectors.toList());
+
+ return scpStorages.stream().filter(s -> request.getStorageId().equals(s.getStorageId())).findFirst();
+ }
}
@Override
@@ -108,15 +137,9 @@
.map(resource -> {
JSONObject r = (JSONObject) resource;
- SCPStorage storage = SCPStorage.newBuilder()
- .setStorageId(((JSONObject)r.get("scpStorage")).get("storageId").toString())
- .setHost(((JSONObject)r.get("scpStorage")).get("host").toString())
- .setUser(((JSONObject)r.get("scpStorage")).get("user").toString())
- .setPort(Integer.parseInt(((JSONObject)r.get("scpStorage")).get("port").toString())).build();
-
SCPResource.Builder builder = SCPResource.newBuilder()
.setResourceId(r.get("resourceId").toString())
- .setScpStorage(storage);
+ .setScpStorage(SCPStorage.newBuilder().setStorageId(r.get("storageId").toString()).getDefaultInstanceForType());
switch (r.get("resourceMode").toString()) {
case "FILE":
@@ -156,6 +179,47 @@
}
@Override
+ public Optional<LocalStorage> getLocalStorage(LocalStorageGetRequest request) throws Exception {
+ JSONParser jsonParser = new JSONParser();
+ InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(storageFile);
+
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+ Object obj = jsonParser.parse(reader);
+
+ JSONArray storageList = (JSONArray) obj;
+
+ List<LocalStorage> localStorages = (List<LocalStorage>) storageList.stream()
+ .filter(storage -> "LOCAL".equals(((JSONObject) storage).get("type").toString()))
+ .map( storage -> {
+ JSONObject s = (JSONObject) storage;
+
+ LocalStorage st = LocalStorage.newBuilder()
+ .setStorageId(s.get("storageId").toString())
+ .setAgentId(s.get("agentId").toString())
+ .build();
+
+ return st;
+ }).collect(Collectors.toList());
+ return localStorages.stream().filter(s -> request.getStorageId().equals(s.getStorageId())).findFirst();
+ }
+ }
+
+ @Override
+ public LocalStorage createLocalStorage(LocalStorageCreateRequest request) throws Exception {
+ return null;
+ }
+
+ @Override
+ public boolean updateLocalStorage(LocalStorageUpdateRequest request) throws Exception {
+ return false;
+ }
+
+ @Override
+ public boolean deleteLocalStorage(LocalStorageDeleteRequest request) throws Exception {
+ return false;
+ }
+
+ @Override
public Optional<LocalResource> getLocalResource(LocalResourceGetRequest request) throws Exception {
JSONParser jsonParser = new JSONParser();
InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(resourceFile);
@@ -214,6 +278,48 @@
}
@Override
+ public Optional<S3Storage> getS3Storage(S3StorageGetRequest request) throws Exception {
+ JSONParser jsonParser = new JSONParser();
+ InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(storageFile);
+
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+ Object obj = jsonParser.parse(reader);
+
+ JSONArray storagesList = (JSONArray) obj;
+
+ List<S3Storage> s3Storages = (List<S3Storage>) storagesList.stream()
+ .filter(storage -> "S3".equals(((JSONObject) storage).get("type").toString()))
+ .map(storage -> {
+ JSONObject s = (JSONObject) storage;
+
+ S3Storage st = S3Storage.newBuilder()
+ .setStorageId(s.get("storageId").toString())
+ .setRegion(s.get("region").toString())
+ .setRegion(s.get("bucketName").toString())
+ .build();
+
+ return st;
+ }).collect(Collectors.toList());
+ return s3Storages.stream().filter(s -> request.getStorageId().equals(s.getStorageId())).findFirst();
+ }
+ }
+
+ @Override
+ public S3Storage createS3Storage(S3StorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateS3Storage(S3StorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteS3Storage(S3StorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<S3Resource> getS3Resource(S3ResourceGetRequest request) throws Exception {
JSONParser jsonParser = new JSONParser();
InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(resourceFile);
@@ -274,6 +380,47 @@
}
@Override
+ public Optional<BoxStorage> getBoxStorage(BoxStorageGetRequest request) throws Exception {
+ JSONParser jsonParser = new JSONParser();
+ InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(storageFile);
+
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+ Object obj = jsonParser.parse(reader);
+
+ JSONArray storageList = (JSONArray) obj;
+
+ System.out.println("All resources ");
+ List<BoxStorage> boxStorages = (List<BoxStorage>) storageList.stream()
+ .filter(storage -> "BOX".equals(((JSONObject) storage).get("type").toString()))
+ .map(storage -> {
+ JSONObject s = (JSONObject) storage;
+
+ BoxStorage st = BoxStorage.newBuilder()
+ .setStorageId(s.get("storageId").toString())
+ .build();
+
+ return st;
+ }).collect(Collectors.toList());
+ return boxStorages.stream().filter(s -> request.getStorageId().equals(s.getStorageId())).findFirst();
+ }
+ }
+
+ @Override
+ public BoxStorage createBoxStorage(BoxStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateBoxStorage(BoxStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteBoxStorage(BoxStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<BoxResource> getBoxResource(BoxResourceGetRequest request) throws Exception {
JSONParser jsonParser = new JSONParser();
InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(resourceFile);
@@ -330,6 +477,47 @@
}
@Override
+ public Optional<AzureStorage> getAzureStorage(AzureStorageGetRequest request) throws Exception {
+ JSONParser jsonParser = new JSONParser();
+ InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(storageFile);
+
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+ Object obj = jsonParser.parse(reader);
+
+ JSONArray storageList = (JSONArray) obj;
+
+ List<AzureStorage> azureStorages = (List<AzureStorage>) storageList.stream()
+ .filter(storage -> "AZURE".equals(((JSONObject) storage).get("type").toString()))
+ .map(storage -> {
+ JSONObject s = (JSONObject) storage;
+
+ AzureStorage st = AzureStorage.newBuilder()
+ .setStorageId(s.get("storageId").toString())
+ .setContainer(s.get("container").toString())
+ .build();
+
+ return st;
+ }).collect(Collectors.toList());
+ return azureStorages.stream().filter(s -> request.getStorageId().equals(s.getStorageId())).findFirst();
+ }
+ }
+
+ @Override
+ public AzureStorage createAzureStorage(AzureStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateAzureStorage(AzureStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteAzureStorage(AzureStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<AzureResource> getAzureResource(AzureResourceGetRequest request) throws Exception {
JSONParser jsonParser = new JSONParser();
InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(resourceFile);
@@ -386,6 +574,47 @@
}
@Override
+ public Optional<GCSStorage> getGCSStorage(GCSStorageGetRequest request) throws Exception {
+ JSONParser jsonParser = new JSONParser();
+ InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(storageFile);
+
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+ Object obj = jsonParser.parse(reader);
+
+ JSONArray storageList = (JSONArray) obj;
+
+ List<GCSStorage> gcsStorages = (List<GCSStorage>) storageList.stream()
+ .filter(storage -> "GCS".equals(((JSONObject) storage).get("type").toString()))
+ .map(storage -> {
+ JSONObject s = (JSONObject) storage;
+
+ GCSStorage st = GCSStorage.newBuilder()
+ .setStorageId(s.get("storageId").toString())
+ .setBucketName(s.get("bucketName").toString())
+ .build();
+
+ return st;
+ }).collect(Collectors.toList());
+ return gcsStorages.stream().filter(s -> request.getStorageId().equals(s.getStorageId())).findFirst();
+ }
+ }
+
+ @Override
+ public GCSStorage createGCSStorage(GCSStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateGCSStorage(GCSStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteGCSStorage(GCSStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<GCSResource> getGCSResource(GCSResourceGetRequest request) throws Exception {
JSONParser jsonParser = new JSONParser();
InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(resourceFile);
@@ -440,6 +669,64 @@
public boolean deleteGCSResource(GCSResourceDeleteRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
+
+ @Override
+ public Optional<DropboxStorage> getDropboxStorage(DropboxStorageGetRequest request) throws Exception {
+ JSONParser jsonParser = new JSONParser();
+ InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(resourceFile);
+
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+ Object obj = jsonParser.parse(reader);
+
+ JSONArray storageList = (JSONArray) obj;
+
+ List<DropboxStorage> dropboxStorages = (List<DropboxStorage>) storageList.stream()
+ .filter(resource -> "DROPBOX".equals(((JSONObject) resource).get("type").toString()))
+ .map(resource -> {
+ JSONObject r = (JSONObject) resource;
+ String resourcePath = r.get("resourcePath").toString();
+ resourcePath = resourcePath.startsWith("/") ? resourcePath : "/" + resourcePath;
+
+ DropboxStorage storage = DropboxStorage.newBuilder()
+ .setStorageId(((JSONObject)r.get("dropboxStorage")).get("storageId").toString())
+ .build();
+
+ DropboxResource.Builder builder = DropboxResource.newBuilder()
+ .setResourceId(r.get("resourceId").toString())
+ .setDropboxStorage(storage);
+
+ switch (r.get("resourceMode").toString()) {
+ case "FILE":
+ FileResource fileResource = FileResource.newBuilder().setResourcePath(r.get("resourcePath").toString()).build();
+ builder = builder.setFile(fileResource);
+ break;
+ case "DIRECTORY":
+ DirectoryResource directoryResource = DirectoryResource.newBuilder().setResourcePath(r.get("resourcePath").toString()).build();
+ builder = builder.setDirectory(directoryResource);
+ break;
+ }
+ return builder.build();
+
+ }).collect(Collectors.toList());
+ return dropboxStorages.stream().filter(s -> request.getStorageId().equals(s.getStorageId())).findFirst();
+ }
+ }
+
+ @Override
+ public DropboxStorage createDropboxStorage(DropboxStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateDropboxStorage(DropboxStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteDropboxStorage(DropboxStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
@Override
public Optional<DropboxResource> getDropboxResource(DropboxResourceGetRequest request) throws Exception {
JSONParser jsonParser = new JSONParser();
@@ -498,6 +785,55 @@
}
@Override
+ public Optional<FTPStorage> getFTPStorage(FTPStorageGetRequest request) throws Exception {
+ InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(storageFile);
+
+ JSONParser jsonParser = new JSONParser();
+
+ if (inputStream == null) {
+ throw new IOException("resources file not found");
+ }
+
+ try (InputStreamReader reader = new InputStreamReader(inputStream)) {
+
+ Object obj = jsonParser.parse(reader);
+
+ JSONArray storageList = (JSONArray) obj;
+
+ List<FTPStorage> ftpStorages = (List<FTPStorage>) storageList.stream()
+ .filter(storage -> "FTP".equals(((JSONObject) storage).get("type").toString()))
+ .map(storage -> {
+ JSONObject s = (JSONObject) storage;
+
+ FTPStorage st = FTPStorage.newBuilder()
+ .setStorageId(s.get("storageId").toString())
+ .setHost(s.get("host").toString())
+ .setPort(Integer.parseInt(s.get("port").toString())).build();
+
+ return st;
+
+ }).collect(Collectors.toList());
+
+ return ftpStorages.stream().filter(s -> request.getStorageId().equals(s.getStorageId())).findFirst();
+ }
+ }
+
+ @Override
+ public FTPStorage createFTPStorage(FTPStorageCreateRequest request) {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateFTPStorage(FTPStorageUpdateRequest request) {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteFTPStorage(FTPStorageDeleteRequest request) {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<FTPResource> getFTPResource(FTPResourceGetRequest request) throws Exception {
InputStream inputStream = FileBasedResourceBackend.class.getClassLoader().getResourceAsStream(resourceFile);
@@ -559,24 +895,4 @@
public boolean deleteFTPResource(FTPResourceDeleteRequest request) {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
-
- @Override
- public Optional<FTPStorage> getFTPStorage(FTPStorageGetRequest request) {
- throw new UnsupportedOperationException("Operation is not supported in backend");
- }
-
- @Override
- public FTPStorage createFTPStorage(FTPStorageCreateRequest request) {
- throw new UnsupportedOperationException("Operation is not supported in backend");
- }
-
- @Override
- public boolean updateFTPStorage(FTPStorageUpdateRequest request) {
- throw new UnsupportedOperationException("Operation is not supported in backend");
- }
-
- @Override
- public boolean deleteFTPStorage(FTPStorageDeleteRequest request) {
- throw new UnsupportedOperationException("Operation is not supported in backend");
- }
}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/SQLResourceBackend.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/SQLResourceBackend.java
index a070b33..caf5057 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/SQLResourceBackend.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/backend/sql/SQLResourceBackend.java
@@ -17,18 +17,23 @@
package org.apache.airavata.mft.resource.server.backend.sql;
-
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.server.backend.sql.entity.*;
import org.apache.airavata.mft.resource.server.backend.sql.repository.*;
import org.apache.airavata.mft.resource.stubs.azure.resource.*;
+import org.apache.airavata.mft.resource.stubs.azure.storage.*;
import org.apache.airavata.mft.resource.stubs.box.resource.*;
+import org.apache.airavata.mft.resource.stubs.box.storage.*;
import org.apache.airavata.mft.resource.stubs.dropbox.resource.*;
+import org.apache.airavata.mft.resource.stubs.dropbox.storage.*;
import org.apache.airavata.mft.resource.stubs.ftp.resource.*;
import org.apache.airavata.mft.resource.stubs.ftp.storage.*;
import org.apache.airavata.mft.resource.stubs.gcs.resource.*;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.*;
import org.apache.airavata.mft.resource.stubs.local.resource.*;
+import org.apache.airavata.mft.resource.stubs.local.storage.*;
import org.apache.airavata.mft.resource.stubs.s3.resource.*;
+import org.apache.airavata.mft.resource.stubs.s3.storage.*;
import org.apache.airavata.mft.resource.stubs.scp.resource.*;
import org.apache.airavata.mft.resource.stubs.scp.storage.*;
import org.dozer.DozerBeanMapper;
@@ -121,6 +126,26 @@
}
@Override
+ public Optional<LocalStorage> getLocalStorage(LocalStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public LocalStorage createLocalStorage(LocalStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateLocalStorage(LocalStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteLocalStorage(LocalStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<LocalResource> getLocalResource(LocalResourceGetRequest request) {
Optional<LocalResourceEntity> resourceEntity = localResourceRepository.findByResourceId(request.getResourceId());
return resourceEntity.map(scpResourceEntity -> mapper.map(scpResourceEntity, LocalResource.newBuilder().getClass()).build());
@@ -145,6 +170,26 @@
}
@Override
+ public Optional<S3Storage> getS3Storage(S3StorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public S3Storage createS3Storage(S3StorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateS3Storage(S3StorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteS3Storage(S3StorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<S3Resource> getS3Resource(S3ResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
@@ -169,6 +214,26 @@
}
@Override
+ public Optional<BoxStorage> getBoxStorage(BoxStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public BoxStorage createBoxStorage(BoxStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateBoxStorage(BoxStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteBoxStorage(BoxStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<BoxResource> getBoxResource(BoxResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
@@ -189,6 +254,26 @@
}
@Override
+ public Optional<AzureStorage> getAzureStorage(AzureStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public AzureStorage createAzureStorage(AzureStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateAzureStorage(AzureStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteAzureStorage(AzureStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<AzureResource> getAzureResource(AzureResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
@@ -209,6 +294,26 @@
}
@Override
+ public Optional<GCSStorage> getGCSStorage(GCSStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public GCSStorage createGCSStorage(GCSStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateGCSStorage(GCSStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteGCSStorage(GCSStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<GCSResource> getGCSResource(GCSResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
@@ -229,6 +334,26 @@
}
@Override
+ public Optional<DropboxStorage> getDropboxStorage(DropboxStorageGetRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public DropboxStorage createDropboxStorage(DropboxStorageCreateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean updateDropboxStorage(DropboxStorageUpdateRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
+ public boolean deleteDropboxStorage(DropboxStorageDeleteRequest request) throws Exception {
+ throw new UnsupportedOperationException("Operation is not supported in backend");
+ }
+
+ @Override
public Optional<DropboxResource> getDropboxResource(DropboxResourceGetRequest request) throws Exception {
throw new UnsupportedOperationException("Operation is not supported in backend");
}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/AzureServiceHandler.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/AzureServiceHandler.java
index 1175ff2..6808096 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/AzureServiceHandler.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/AzureServiceHandler.java
@@ -23,6 +23,7 @@
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.service.azure.AzureResourceServiceGrpc;
import org.apache.airavata.mft.resource.stubs.azure.resource.*;
+import org.apache.airavata.mft.resource.stubs.azure.storage.*;
import org.lognet.springboot.grpc.GRpcService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -102,4 +103,69 @@
}
}
+ @Override
+ public void getAzureStorage(AzureStorageGetRequest request, StreamObserver<AzureStorage> responseObserver) {
+ try {
+ this.backend.getAzureStorage(request).ifPresentOrElse(resource -> {
+ responseObserver.onNext(resource);
+ responseObserver.onCompleted();
+ }, () -> {
+ responseObserver.onError(Status.INTERNAL
+ .withDescription("No Azure storage with id " + request.getStorageId())
+ .asRuntimeException());
+ });
+ } catch (Exception e) {
+ logger.error("Failed in retrieving Azure resource with id {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in retrieving Azure resource with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void createAzureStorage(AzureStorageCreateRequest request, StreamObserver<AzureStorage> responseObserver) {
+ try {
+ responseObserver.onNext(this.backend.createAzureStorage(request));
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in creating the Azure storage", e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in creating the Azure resource")
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void updateAzureStorage(AzureStorageUpdateRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ this.backend.updateAzureStorage(request);
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in updating the Azure storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in updating the Azure storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void deleteAzureStorage(AzureStorageDeleteRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ boolean res = this.backend.deleteAzureStorage(request);
+ if (res) {
+ responseObserver.onCompleted();
+ } else {
+ responseObserver.onError(new Exception("Failed to delete Azure storage with id " + request.getStorageId()));
+ }
+ } catch (Exception e) {
+ logger.error("Failed in deleting the Azure storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in deleting the Azure storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/BoxServiceHandler.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/BoxServiceHandler.java
index a17422c..5279d0a 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/BoxServiceHandler.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/BoxServiceHandler.java
@@ -23,6 +23,7 @@
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.service.box.BoxResourceServiceGrpc;
import org.apache.airavata.mft.resource.stubs.box.resource.*;
+import org.apache.airavata.mft.resource.stubs.box.storage.*;
import org.lognet.springboot.grpc.GRpcService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,7 +68,8 @@
responseObserver.onError(Status.INTERNAL.withCause(e)
.withDescription("Failed in creating the Box resource")
.asRuntimeException());
- } }
+ }
+ }
@Override
public void updateBoxResource(BoxResourceUpdateRequest request, StreamObserver<Empty> responseObserver) {
@@ -80,7 +82,8 @@
responseObserver.onError(Status.INTERNAL.withCause(e)
.withDescription("Failed in updating the Box resource with id " + request.getResourceId())
.asRuntimeException());
- } }
+ }
+ }
@Override
public void deleteBoxResource(BoxResourceDeleteRequest request, StreamObserver<Empty> responseObserver) {
@@ -99,4 +102,70 @@
.asRuntimeException());
}
}
+
+ @Override
+ public void getBoxStorage(BoxStorageGetRequest request, StreamObserver<BoxStorage> responseObserver) {
+ try {
+ this.backend.getBoxStorage(request).ifPresentOrElse(resource -> {
+ responseObserver.onNext(resource);
+ responseObserver.onCompleted();
+ }, () -> {
+ responseObserver.onError(Status.INTERNAL
+ .withDescription("No Box storage with id " + request.getStorageId())
+ .asRuntimeException());
+ });
+ } catch (Exception e) {
+ logger.error("Failed in retrieving Box resource with id {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in retrieving Box storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void createBoxStorage(BoxStorageCreateRequest request, StreamObserver<BoxStorage> responseObserver) {
+ try {
+ responseObserver.onNext(this.backend.createBoxStorage(request));
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in creating the Box storage", e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in creating the Box storage")
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void updateBoxStorage(BoxStorageUpdateRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ this.backend.updateBoxStorage(request);
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in updating the Box storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in updating the Box storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void deleteBoxStorage(BoxStorageDeleteRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ boolean res = this.backend.deleteBoxStorage(request);
+ if (res) {
+ responseObserver.onCompleted();
+ } else {
+ responseObserver.onError(new Exception("Failed to delete Box storage with id " + request.getStorageId()));
+ }
+ } catch (Exception e) {
+ logger.error("Failed in deleting the Box storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in deleting the Box storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/DropboxServiceHandler.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/DropboxServiceHandler.java
index acdd1cc..0435361 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/DropboxServiceHandler.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/DropboxServiceHandler.java
@@ -23,6 +23,7 @@
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.service.dropbox.DropboxServiceGrpc;
import org.apache.airavata.mft.resource.stubs.dropbox.resource.*;
+import org.apache.airavata.mft.resource.stubs.dropbox.storage.*;
import org.lognet.springboot.grpc.GRpcService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -101,4 +102,70 @@
.asRuntimeException());
}
}
+
+ @Override
+ public void getDropboxStorage(DropboxStorageGetRequest request, StreamObserver<DropboxStorage> responseObserver) {
+ try {
+ this.backend.getDropboxStorage(request).ifPresentOrElse(resource -> {
+ responseObserver.onNext(resource);
+ responseObserver.onCompleted();
+ }, () -> {
+ responseObserver.onError(Status.INTERNAL
+ .withDescription("No dropbox storage with id " + request.getStorageId())
+ .asRuntimeException());
+ });
+ } catch (Exception e) {
+ logger.error("Failed in retrieving dropbox storage with id {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in retrieving dropbox storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void createDropboxStorage(DropboxStorageCreateRequest request, StreamObserver<DropboxStorage> responseObserver) {
+ try {
+ responseObserver.onNext(this.backend.createDropboxStorage(request));
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in creating the dropbox storage", e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in creating the dropbox storage")
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void updateDropboxStorage(DropboxStorageUpdateRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ this.backend.updateDropboxStorage(request);
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in updating the dropbox storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in updating the dropbox storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void deleteDropboxStorage(DropboxStorageDeleteRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ boolean res = this.backend.deleteDropboxStorage(request);
+ if (res) {
+ responseObserver.onCompleted();
+ } else {
+ responseObserver.onError(new Exception("Failed to delete dropbox storage with id " + request.getStorageId()));
+ }
+ } catch (Exception e) {
+ logger.error("Failed in deleting the dropbox storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in deleting the dropbox storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/GCSServiceHandler.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/GCSServiceHandler.java
index 300e0e2..474ea62 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/GCSServiceHandler.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/GCSServiceHandler.java
@@ -23,6 +23,7 @@
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.service.gcs.GCSResourceServiceGrpc;
import org.apache.airavata.mft.resource.stubs.gcs.resource.*;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.*;
import org.lognet.springboot.grpc.GRpcService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -101,4 +102,70 @@
.asRuntimeException());
}
}
+
+ @Override
+ public void getGCSStorage(GCSStorageGetRequest request, StreamObserver<GCSStorage> responseObserver) {
+ try {
+ this.backend.getGCSStorage(request).ifPresentOrElse(resource -> {
+ responseObserver.onNext(resource);
+ responseObserver.onCompleted();
+ }, () -> {
+ responseObserver.onError(Status.INTERNAL
+ .withDescription("No GCS storage with id " + request.getStorageId())
+ .asRuntimeException());
+ });
+ } catch (Exception e) {
+ logger.error("Failed in retrieving GCS storage with id {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in retrieving GCS storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void createGCSStorage(GCSStorageCreateRequest request, StreamObserver<GCSStorage> responseObserver) {
+ try {
+ responseObserver.onNext(this.backend.createGCSStorage(request));
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in creating the GCS storage", e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in creating the GCS storage")
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void updateGCSStorage(GCSStorageUpdateRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ this.backend.updateGCSStorage(request);
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in updating the GCS storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in updating the GCS storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void deleteGCSStorage(GCSStorageDeleteRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ boolean res = this.backend.deleteGCSStorage(request);
+ if (res) {
+ responseObserver.onCompleted();
+ } else {
+ responseObserver.onError(new Exception("Failed to delete GCS storage with id " + request.getStorageId()));
+ }
+ } catch (Exception e) {
+ logger.error("Failed in deleting the GCS storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in deleting the GCS storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/LocalServiceHandler.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/LocalServiceHandler.java
index 212bd24..122ac03 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/LocalServiceHandler.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/LocalServiceHandler.java
@@ -23,6 +23,7 @@
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.service.local.LocalResourceServiceGrpc;
import org.apache.airavata.mft.resource.stubs.local.resource.*;
+import org.apache.airavata.mft.resource.stubs.local.storage.*;
import org.lognet.springboot.grpc.GRpcService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -103,4 +104,69 @@
}
}
+ @Override
+ public void getLocalStorage(LocalStorageGetRequest request, StreamObserver<LocalStorage> responseObserver) {
+ try {
+ this.backend.getLocalStorage(request).ifPresentOrElse(resource -> {
+ responseObserver.onNext(resource);
+ responseObserver.onCompleted();
+ }, () -> {
+ responseObserver.onError(Status.INTERNAL
+ .withDescription("No Local storage with id " + request.getStorageId())
+ .asRuntimeException());
+ });
+ } catch (Exception e) {
+ logger.error("Failed in retrieving storage with id {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in retrieving storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void createLocalStorage(LocalStorageCreateRequest request, StreamObserver<LocalStorage> responseObserver) {
+ try {
+ responseObserver.onNext(this.backend.createLocalStorage(request));
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in creating the local storage", e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in creating the local storage")
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void updateLocalStorage(LocalStorageUpdateRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ this.backend.updateLocalStorage(request);
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in updating the local storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in updating the local storge with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void deleteLocalStorage(LocalStorageDeleteRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ boolean res = this.backend.deleteLocalStorage(request);
+ if (res) {
+ responseObserver.onCompleted();
+ } else {
+ responseObserver.onError(new Exception("Failed to delete Local storage with id " + request.getStorageId()));
+ }
+ } catch (Exception e) {
+ logger.error("Failed in deleting the local storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in deleting the local storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
}
diff --git a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/S3ServiceHandler.java b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/S3ServiceHandler.java
index 753bd6a..53e0dcb 100644
--- a/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/S3ServiceHandler.java
+++ b/services/resource-service/server/src/main/java/org/apache/airavata/mft/resource/server/handler/S3ServiceHandler.java
@@ -23,6 +23,7 @@
import org.apache.airavata.mft.resource.server.backend.ResourceBackend;
import org.apache.airavata.mft.resource.service.s3.S3ResourceServiceGrpc;
import org.apache.airavata.mft.resource.stubs.s3.resource.*;
+import org.apache.airavata.mft.resource.stubs.s3.storage.*;
import org.lognet.springboot.grpc.GRpcService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -101,4 +102,70 @@
.asRuntimeException());
}
}
+
+ @Override
+ public void getS3Storage(S3StorageGetRequest request, StreamObserver<S3Storage> responseObserver) {
+ try {
+ this.backend.getS3Storage(request).ifPresentOrElse(resource -> {
+ responseObserver.onNext(resource);
+ responseObserver.onCompleted();
+ }, () -> {
+ responseObserver.onError(Status.INTERNAL
+ .withDescription("No S3 storage with id " + request.getStorageId())
+ .asRuntimeException());
+ });
+ } catch (Exception e) {
+ logger.error("Failed in retrieving S3 storage with id {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in retrieving S3 storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void createS3Storage(S3StorageCreateRequest request, StreamObserver<S3Storage> responseObserver) {
+ try {
+ responseObserver.onNext(this.backend.createS3Storage(request));
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in creating the S3 storage", e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in creating the S3 storage")
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void updateS3Storage(S3StorageUpdateRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ this.backend.updateS3Storage(request);
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ logger.error("Failed in updating the S3 storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in updating the S3 storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
+
+ @Override
+ public void deleteS3Storage(S3StorageDeleteRequest request, StreamObserver<Empty> responseObserver) {
+ try {
+ boolean res = this.backend.deleteS3Storage(request);
+ if (res) {
+ responseObserver.onCompleted();
+ } else {
+ responseObserver.onError(new Exception("Failed to delete S3 storage with id " + request.getStorageId()));
+ }
+ } catch (Exception e) {
+ logger.error("Failed in deleting the S3 storage {}", request.getStorageId(), e);
+
+ responseObserver.onError(Status.INTERNAL.withCause(e)
+ .withDescription("Failed in deleting the S3 storage with id " + request.getStorageId())
+ .asRuntimeException());
+ }
+ }
}
diff --git a/services/resource-service/server/src/main/resources/application.properties b/services/resource-service/server/src/main/resources/application.properties
index 51872df..3dbfa86 100644
--- a/services/resource-service/server/src/main/resources/application.properties
+++ b/services/resource-service/server/src/main/resources/application.properties
@@ -24,4 +24,5 @@
airavata.backend.registry.server.port=8970
# Configurations for File Backend
-file.backend.resource.file=resources.json
\ No newline at end of file
+file.backend.resource.file=resources.json
+file.backend.storage.file=storages.json
\ No newline at end of file
diff --git a/services/resource-service/server/src/main/resources/resources.json b/services/resource-service/server/src/main/resources/resources.json
index a389282..3d7b62f 100644
--- a/services/resource-service/server/src/main/resources/resources.json
+++ b/services/resource-service/server/src/main/resources/resources.json
@@ -4,114 +4,76 @@
"resourceId": "remote-ssh-dir-resource",
"resourceMode": "DIRECTORY",
"resourcePath": "/tmp",
- "scpStorage" : {
- "storageId": "remote-ssh-storage",
- "host": "149.165.156.124",
- "port": 22,
- "user": "root"
- }
+ "storageId" : "remote-ssh-storage"
},
{
"type": "SCP",
"resourceId": "remote-ssh-resource",
"resourceMode": "FILE",
"resourcePath": "/tmp/1mb.txt",
- "scpStorage" : {
- "storageId": "remote-ssh-storage",
- "host": "149.165.156.124",
- "port": 22,
- "user": "root"
- }
+ "storageId" : "remote-ssh-storage"
},
{
"type": "SCP",
"resourceId": "remote-ssh-resource2",
"resourceMode": "FILE",
"resourcePath": "/tmp/10mb.txt",
- "scpStorage" : {
- "storageId": "remote-ssh-storage",
- "host": "149.165.156.124",
- "port": 22,
- "user": "root"
- }
+ "storageId" : "remote-ssh-storage"
},
{
"type": "LOCAL",
"resourceId": "10mb-file",
"resourceMode": "FILE",
"resourcePath": "/tmp/10mb.txt",
- "localStorage": {
- "storageId": "local-storage-1",
- "agentId": "agent-0"
- }
+ "storageId": "local-storage-1"
},
{
"type": "S3",
"resourceId": "s3-file",
"resourceMode": "FILE",
"resourcePath": "10mb-s3.txt",
- "s3Storage": {
- "storageId": "s3-storage-1",
- "region": "us-east-2",
- "bucketName": "airavata-s3"
- }
+ "storageId": "s3-storage-1"
},
{
"type": "BOX",
"resourceId": "box-file-abcd",
"resourceMode": "FILE",
"resourcePath": "655108198452",
- "boxStorage" : {
- "storageId": "box-storage-1"
- }
+ "storageId" : "box-storage-1"
},
{
"type": "BOX",
"resourceId": "box-file-efgh",
"resourceMode": "FILE",
"resourcePath": "655450661536",
- "boxStorage" : {
- "storageId": "box-storage-1"
- }
+ "storageId" : "box-storage-1"
},
{
"type": "AZURE",
"resourceId": "azure-blob",
"resourceMode": "FILE",
"resourcePath": "sample.blob",
- "azureStorage" : {
- "storageId": "azure-storage-1",
- "container": "sample-container"
- }
+ "storageId" : "azure-storage-1"
},
{
"type": "GCS",
"resourceId": "gcs-bucket",
"resourceMode": "FILE",
"resourcePath": "PikaPikaTest.txt",
- "gcsStorage": {
- "storageId": "gcs-storage-1",
- "bucketName": "pika-pika-bucket"
- }
+ "storageId": "gcs-storage-1"
},
{
"type": "DROPBOX",
"resourceId": "dropbox-file",
"resourceMode": "FILE",
"resourcePath": "/test.txt",
- "dropboxStorage": {
- "storageId": "dropbox-storage-1"
- }
+ "storageId": "dropbox-storage-1"
},
{
"type": "FTP",
"resourceId": "ftp-resource",
"resourceMode": "FILE",
"resourcePath": "mft-1mb.txt",
- "ftpStorage": {
- "storageId": "ftp-resource",
- "host": "ftp.dlptest.com",
- "port": "21"
- }
+ "storageId": "ftp-resource"
}
]
\ No newline at end of file
diff --git a/services/resource-service/server/src/main/resources/storages.json b/services/resource-service/server/src/main/resources/storages.json
new file mode 100644
index 0000000..69d0b13
--- /dev/null
+++ b/services/resource-service/server/src/main/resources/storages.json
@@ -0,0 +1,44 @@
+[
+ {
+ "type": "SCP",
+ "storageId": "remote-ssh-storage",
+ "host": "149.165.156.124",
+ "port": 22,
+ "user": "root"
+ },
+ {
+ "type": "LOCAL",
+ "storageId": "local-storage-1",
+ "agentId": "agent-0"
+ },
+ {
+ "type": "S3",
+ "storageId": "s3-storage-1",
+ "region": "us-east-2",
+ "bucketName": "airavata-s3"
+ },
+ {
+ "type": "BOX",
+ "storageId": "box-storage-1"
+ },
+ {
+ "type": "AZURE",
+ "storageId": "azure-storage-1",
+ "container": "sample-container"
+ },
+ {
+ "type": "GCS",
+ "storageId": "gcs-storage-1",
+ "bucketName": "pika-pika-bucket"
+ },
+ {
+ "type": "DROPBOX",
+ "storageId": "dropbox-storage-1"
+ },
+ {
+ "type": "FTP",
+ "storageId": "ftp-resource",
+ "host": "ftp.dlptest.com",
+ "port": "21"
+ }
+]
\ No newline at end of file
diff --git a/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureMetadataCollector.java b/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureMetadataCollector.java
index 5a64d54..76a6f13 100644
--- a/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureMetadataCollector.java
+++ b/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureMetadataCollector.java
@@ -32,6 +32,9 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.azure.resource.AzureResource;
import org.apache.airavata.mft.resource.stubs.azure.resource.AzureResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.azure.storage.AzureStorage;
+import org.apache.airavata.mft.resource.stubs.azure.storage.AzureStorageGetRequest;
+import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
@@ -95,7 +98,7 @@
}
@Override
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -104,7 +107,7 @@
throw new UnsupportedOperationException("Method not implemented"); }
@Override
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -115,6 +118,20 @@
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
AzureResource azureResource = resourceClient.azure().getAzureResource(AzureResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ return isAvailable(azureResource, credentialToken);
+ }
+
+ @Override
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception {
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ AzureStorage azureStorage = resourceClient.azure().getAzureStorage(AzureStorageGetRequest.newBuilder().setStorageId(storageId).build());
+
+ AzureResource azureResource = AzureResource.newBuilder().setFile(FileResource.newBuilder().setResourcePath(resourcePath).build()).setAzureStorage(azureStorage).build();
+ return isAvailable(azureResource, credentialToken);
+ }
+
+ public Boolean isAvailable(AzureResource azureResource, String credentialToken) throws Exception {
+
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
AzureSecret azureSecret = secretClient.azure().getAzureSecret(AzureSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
@@ -132,4 +149,5 @@
}
return false;
}
+
}
diff --git a/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureReceiver.java b/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureReceiver.java
index a9bfdf7..dab9337 100644
--- a/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureReceiver.java
+++ b/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureReceiver.java
@@ -29,8 +29,8 @@
import org.apache.airavata.mft.credential.stubs.azure.AzureSecretGetRequest;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
-import org.apache.airavata.mft.resource.stubs.azure.resource.AzureResource;
-import org.apache.airavata.mft.resource.stubs.azure.resource.AzureResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.azure.storage.AzureStorage;
+import org.apache.airavata.mft.resource.stubs.azure.storage.AzureStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -43,21 +43,20 @@
private static final Logger logger = LoggerFactory.getLogger(AzureReceiver.class);
private boolean initialized = false;
- private AzureResource azureResource;
BlobContainerClient containerClient;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
this.initialized = true;
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.azureResource = resourceClient.azure().getAzureResource(AzureResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ AzureStorage azureStorage = resourceClient.azure().getAzureStorage(AzureStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
AzureSecret azureSecret = secretClient.azure().getAzureSecret(AzureSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
BlobServiceClient blobServiceClient = new BlobServiceClientBuilder().connectionString(azureSecret.getConnectionString()).buildClient();
- this.containerClient = blobServiceClient.getBlobContainerClient(azureResource.getAzureStorage().getContainer());
+ this.containerClient = blobServiceClient.getBlobContainerClient(azureStorage.getContainer());
}
@Override
@@ -72,49 +71,41 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting azure receive for remote server for transfer {}", context.getTransferId());
checkInitialized();
- if (ResourceTypes.FILE.equals(this.azureResource.getResourceCase().name())) {
- BlobClient blobClient = containerClient.getBlobClient(azureResource.getFile().getResourcePath());
- BlobInputStream blobInputStream = blobClient.openInputStream();
+ BlobClient blobClient = containerClient.getBlobClient(targetPath);
+ BlobInputStream blobInputStream = blobClient.openInputStream();
- OutputStream streamOs = context.getStreamBuffer().getOutputStream();
+ OutputStream streamOs = context.getStreamBuffer().getOutputStream();
- long fileSize = context.getMetadata().getResourceSize();
+ long fileSize = context.getMetadata().getResourceSize();
- byte[] buf = new byte[1024];
- while (true) {
- int bufSize = 0;
+ byte[] buf = new byte[1024];
+ while (true) {
+ int bufSize = 0;
- if (buf.length < fileSize) {
- bufSize = buf.length;
- } else {
- bufSize = (int) fileSize;
- }
- bufSize = blobInputStream.read(buf, 0, bufSize);
+ if (buf.length < fileSize) {
+ bufSize = buf.length;
+ } else {
+ bufSize = (int) fileSize;
+ }
+ bufSize = blobInputStream.read(buf, 0, bufSize);
- if (bufSize < 0) {
- break;
- }
-
- streamOs.write(buf, 0, bufSize);
- streamOs.flush();
-
- fileSize -= bufSize;
- if (fileSize == 0L)
- break;
+ if (bufSize < 0) {
+ break;
}
- streamOs.close();
- logger.info("Completed azure receive for remote server for transfer {}", context.getTransferId());
+ streamOs.write(buf, 0, bufSize);
+ streamOs.flush();
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.azureResource.getResourceId(), this.azureResource.getResourceCase().name());
- throw new Exception("Resource " + this.azureResource.getResourceId() + " should be a FILE type. Found a " +
- this.azureResource.getResourceCase().name());
+ fileSize -= bufSize;
+ if (fileSize == 0L)
+ break;
}
+
+ streamOs.close();
+ logger.info("Completed azure receive for remote server for transfer {}", context.getTransferId());
}
}
diff --git a/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureSender.java b/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureSender.java
index 7dcff4b..ef3249e 100644
--- a/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureSender.java
+++ b/transport/azure-transport/src/main/java/org/apache/airavata/mft/transport/azure/AzureSender.java
@@ -30,6 +30,8 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.azure.resource.AzureResource;
import org.apache.airavata.mft.resource.stubs.azure.resource.AzureResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.azure.storage.AzureStorage;
+import org.apache.airavata.mft.resource.stubs.azure.storage.AzureStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -40,21 +42,20 @@
private static final Logger logger = LoggerFactory.getLogger(AzureSender.class);
private boolean initialized = false;
- private AzureResource azureResource;
BlobContainerClient containerClient;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
this.initialized = true;
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.azureResource = resourceClient.azure().getAzureResource(AzureResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ AzureStorage azureStorage = resourceClient.azure().getAzureStorage(AzureStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
AzureSecret azureSecret = secretClient.azure().getAzureSecret(AzureSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
BlobServiceClient blobServiceClient = new BlobServiceClientBuilder().connectionString(azureSecret.getConnectionString()).buildClient();
- this.containerClient = blobServiceClient.getBlobContainerClient(azureResource.getAzureStorage().getContainer());
+ this.containerClient = blobServiceClient.getBlobContainerClient(azureStorage.getContainer());
}
@Override
@@ -69,20 +70,13 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting Azure send for remote server for transfer {}", context.getTransferId());
checkInitialized();
- if (ResourceTypes.FILE.equals(this.azureResource.getResourceCase().name())) {
- BlockBlobClient blockBlobClient = containerClient.getBlobClient(azureResource.getFile().getResourcePath()).getBlockBlobClient();
- blockBlobClient.upload(context.getStreamBuffer().getInputStream(), context.getMetadata().getResourceSize(), true);
- logger.info("Completed Azure send for remote server for transfer {}", context.getTransferId());
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.azureResource.getResourceId(), this.azureResource.getResourceCase().name());
- throw new Exception("Resource " + this.azureResource.getResourceId() + " should be a FILE type. Found a " +
- this.azureResource.getResourceCase().name());
- }
+ BlockBlobClient blockBlobClient = containerClient.getBlobClient(targetPath).getBlockBlobClient();
+ blockBlobClient.upload(context.getStreamBuffer().getInputStream(), context.getMetadata().getResourceSize(), true);
+ logger.info("Completed Azure send for remote server for transfer {}", context.getTransferId());
}
}
diff --git a/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxMetadataCollector.java b/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxMetadataCollector.java
index 8e21c9f..5e77e72 100644
--- a/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxMetadataCollector.java
+++ b/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxMetadataCollector.java
@@ -30,6 +30,9 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.box.resource.BoxResource;
import org.apache.airavata.mft.resource.stubs.box.resource.BoxResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.box.storage.BoxStorage;
+import org.apache.airavata.mft.resource.stubs.box.storage.BoxStorageGetRequest;
+import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
@@ -84,7 +87,7 @@
}
@Override
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -93,7 +96,7 @@
throw new UnsupportedOperationException("Method not implemented"); }
@Override
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -104,7 +107,22 @@
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
BoxResource boxResource = resourceClient.box().getBoxResource(BoxResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ return isAvailable(boxResource, credentialToken);
+ }
+ @Override
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception {
+ checkInitialized();
+
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ BoxStorage boxStorage = resourceClient.box().getBoxStorage(BoxStorageGetRequest.newBuilder().setStorageId(storageId).build());
+
+ BoxResource boxResource = BoxResource.newBuilder().setFile(FileResource.newBuilder().setResourcePath(resourcePath).build()).setBoxStorage(boxStorage).build();
+
+ return isAvailable(boxResource, credentialToken);
+ }
+
+ private Boolean isAvailable(BoxResource boxResource, String credentialToken) throws Exception {
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
BoxSecret boxSecret = secretClient.box().getBoxSecret(BoxSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
diff --git a/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxReceiver.java b/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxReceiver.java
index 907b485..e041e35 100644
--- a/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxReceiver.java
+++ b/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxReceiver.java
@@ -29,6 +29,8 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.box.resource.BoxResource;
import org.apache.airavata.mft.resource.stubs.box.resource.BoxResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.box.storage.BoxStorage;
+import org.apache.airavata.mft.resource.stubs.box.storage.BoxStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -40,20 +42,14 @@
public class BoxReceiver implements Connector {
private static final Logger logger = LoggerFactory.getLogger(BoxReceiver.class);
- private BoxSecret boxSecret;
- private BoxResource boxResource;
private BoxAPIConnection boxClient;
-
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort,
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort,
String secretServiceHost, int secretServicePort) throws Exception {
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- boxResource = resourceClient.box().getBoxResource(BoxResourceGetRequest.newBuilder().setResourceId(resourceId).build());
-
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- boxSecret = secretClient.box().getBoxSecret(BoxSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+ BoxSecret boxSecret = secretClient.box().getBoxSecret(BoxSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
boxClient = new BoxAPIConnection(boxSecret.getAccessToken());
}
@@ -64,25 +60,17 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
- if (ResourceTypes.FILE.equals(this.boxResource.getResourceCase().name())) {
- logger.info("Starting Box Receiver stream for transfer {}", context.getTransferId());
+ logger.info("Starting Box Receiver stream for transfer {}", context.getTransferId());
- BoxFile file = new BoxFile(this.boxClient, this.boxResource.getFile().getResourcePath());
+ BoxFile file = new BoxFile(this.boxClient, targetPath);
- OutputStream os = context.getStreamBuffer().getOutputStream();
- file.download(os);
- os.flush();
- os.close();
+ OutputStream os = context.getStreamBuffer().getOutputStream();
+ file.download(os);
+ os.flush();
+ os.close();
- logger.info("Completed Box Receiver stream for transfer {}", context.getTransferId());
-
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.boxResource.getResourceId(), this.boxResource.getResourceCase().name());
- throw new Exception("Resource " + this.boxResource.getResourceId() + " should be a FILE type. Found a " +
- this.boxResource.getResourceCase().name());
- }
+ logger.info("Completed Box Receiver stream for transfer {}", context.getTransferId());
}
}
diff --git a/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxSender.java b/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxSender.java
index c9b279f..6cd50c5 100644
--- a/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxSender.java
+++ b/transport/box-transport/src/main/java/org/apache/airavata/mft/transport/box/BoxSender.java
@@ -37,19 +37,13 @@
public class BoxSender implements Connector {
private static final Logger logger = LoggerFactory.getLogger(BoxSender.class);
- private BoxSecret boxSecret;
- private BoxResource boxResource;
private BoxAPIConnection boxClient;
-
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
-
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- boxResource = resourceClient.box().getBoxResource(BoxResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- boxSecret = secretClient.box().getBoxSecret(BoxSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+ BoxSecret boxSecret = secretClient.box().getBoxSecret(BoxSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
boxClient = new BoxAPIConnection(boxSecret.getAccessToken());
}
@@ -60,29 +54,22 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting Box Sender stream for transfer {}", context.getTransferId());
logger.debug("Content length for transfer {} {}", context.getTransferId(), context.getMetadata().getResourceSize());
- if (ResourceTypes.FILE.equals(this.boxResource.getResourceCase().name())) {
- BoxFile file = new BoxFile(this.boxClient, this.boxResource.getFile().getResourcePath());
+ BoxFile file = new BoxFile(this.boxClient, targetPath);
- // Upload chunks only if the file size is > 20mb
- // Ref: https://developer.box.com/guides/uploads/chunked/
- if (context.getMetadata().getResourceSize() > 20971520) {
- file.uploadLargeFile(context.getStreamBuffer().getInputStream(), context.getMetadata().getResourceSize());
- } else {
- file.uploadNewVersion(context.getStreamBuffer().getInputStream());
- }
-
- logger.info("Completed Box Sender stream for transfer {}", context.getTransferId());
-
+ // Upload chunks only if the file size is > 20mb
+ // Ref: https://developer.box.com/guides/uploads/chunked/
+ if (context.getMetadata().getResourceSize() > 20971520) {
+ file.uploadLargeFile(context.getStreamBuffer().getInputStream(), context.getMetadata().getResourceSize());
} else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.boxResource.getResourceId(), this.boxResource.getResourceCase().name());
- throw new Exception("Resource " + this.boxResource.getResourceId() + " should be a FILE type. Found a " +
- this.boxResource.getResourceCase().name());
+ file.uploadNewVersion(context.getStreamBuffer().getInputStream());
}
+
+ logger.info("Completed Box Sender stream for transfer {}", context.getTransferId());
+
}
}
diff --git a/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxMetadataCollector.java b/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxMetadataCollector.java
index 906fb0b..2d00e10 100644
--- a/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxMetadataCollector.java
+++ b/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxMetadataCollector.java
@@ -28,8 +28,11 @@
import org.apache.airavata.mft.credential.stubs.dropbox.DropboxSecretGetRequest;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
+import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.resource.stubs.dropbox.resource.DropboxResource;
import org.apache.airavata.mft.resource.stubs.dropbox.resource.DropboxResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.dropbox.storage.DropboxStorage;
+import org.apache.airavata.mft.resource.stubs.dropbox.storage.DropboxStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
@@ -78,7 +81,7 @@
}
@Override
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -87,7 +90,7 @@
throw new UnsupportedOperationException("Method not implemented"); }
@Override
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -97,6 +100,23 @@
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
DropboxResource dropboxResource = resourceClient.dropbox().getDropboxResource(DropboxResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ return isAvailable(dropboxResource, credentialToken);
+ }
+
+ @Override
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception {
+
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ DropboxStorage dropboxStorage = resourceClient.dropbox().getDropboxStorage(DropboxStorageGetRequest.newBuilder().setStorageId(storageId).build());
+
+ DropboxResource dropboxResource = DropboxResource.newBuilder()
+ .setFile(FileResource.newBuilder().setResourcePath(resourcePath).build())
+ .setDropboxStorage(dropboxStorage).build();
+
+ return isAvailable(dropboxResource, credentialToken);
+ }
+
+ private Boolean isAvailable(DropboxResource dropboxResource, String credentialToken) throws Exception {
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
DropboxSecret dropboxSecret = secretClient.dropbox().getDropboxSecret(DropboxSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
diff --git a/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxReceiver.java b/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxReceiver.java
index 5bdf5a9..d694f91 100644
--- a/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxReceiver.java
+++ b/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxReceiver.java
@@ -40,14 +40,10 @@
private static final Logger logger = LoggerFactory.getLogger(DropboxReceiver.class);
- private DropboxResource dropboxResource;
private DbxClientV2 dbxClientV2;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.dropboxResource = resourceClient.dropbox().getDropboxResource(DropboxResourceGetRequest.newBuilder().setResourceId(resourceId).build());
-
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
DropboxSecret dropboxSecret = secretClient.dropbox().getDropboxSecret(DropboxSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
@@ -61,48 +57,41 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
- if (ResourceTypes.FILE.equals(this.dropboxResource.getResourceCase().name())) {
- logger.info("Starting Dropbox Receiver stream for transfer {}", context.getTransferId());
+ logger.info("Starting Dropbox Receiver stream for transfer {}", context.getTransferId());
- InputStream inputStream = dbxClientV2.files().download(this.dropboxResource.getFile().getResourcePath()).getInputStream();
- OutputStream os = context.getStreamBuffer().getOutputStream();
- int read;
- long bytes = 0;
- long fileSize = context.getMetadata().getResourceSize();
- byte[] buf = new byte[1024];
- while (true) {
- int bufSize = 0;
+ InputStream inputStream = dbxClientV2.files().download(targetPath).getInputStream();
+ OutputStream os = context.getStreamBuffer().getOutputStream();
+ int read;
+ long bytes = 0;
+ long fileSize = context.getMetadata().getResourceSize();
+ byte[] buf = new byte[1024];
+ while (true) {
+ int bufSize = 0;
- if (buf.length < fileSize) {
- bufSize = buf.length;
- } else {
- bufSize = (int) fileSize;
- }
- bufSize = inputStream.read(buf, 0, bufSize);
+ if (buf.length < fileSize) {
+ bufSize = buf.length;
+ } else {
+ bufSize = (int) fileSize;
+ }
+ bufSize = inputStream.read(buf, 0, bufSize);
- if (bufSize < 0) {
- break;
- }
-
- os.write(buf, 0, bufSize);
- os.flush();
-
- fileSize -= bufSize;
- if (fileSize == 0L)
- break;
+ if (bufSize < 0) {
+ break;
}
- os.close();
+ os.write(buf, 0, bufSize);
+ os.flush();
- logger.info("Completed Dropbox Receiver stream for transfer {}", context.getTransferId());
-
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.dropboxResource.getResourceId(), this.dropboxResource.getResourceCase().name());
- throw new Exception("Resource " + this.dropboxResource.getResourceId() + " should be a FILE type. Found a " +
- this.dropboxResource.getResourceCase().name());
+ fileSize -= bufSize;
+ if (fileSize == 0L)
+ break;
}
+
+ os.close();
+
+ logger.info("Completed Dropbox Receiver stream for transfer {}", context.getTransferId());
+
}
}
diff --git a/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxSender.java b/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxSender.java
index ffab965..da8f8b7 100644
--- a/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxSender.java
+++ b/transport/dropbox-transport/src/main/java/org/apache/airavata/mft/transport/dropbox/DropboxSender.java
@@ -40,13 +40,10 @@
private static final Logger logger = LoggerFactory.getLogger(DropboxSender.class);
- private DropboxResource dropboxResource;
private DbxClientV2 dbxClientV2;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.dropboxResource = resourceClient.dropbox().getDropboxResource(DropboxResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
DropboxSecret dropboxSecret = secretClient.dropbox().getDropboxSecret(DropboxSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
@@ -61,24 +58,15 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting Dropbox Sender stream for transfer {}", context.getTransferId());
logger.info("Content length for transfer {} {}", context.getTransferId(), context.getMetadata().getResourceSize());
+ FileMetadata metadata = dbxClientV2.files().uploadBuilder(targetPath)
+ .withMode(WriteMode.OVERWRITE)
+ .uploadAndFinish(context.getStreamBuffer().getInputStream());
+ logger.info("Completed Dropbox Sender stream for transfer {}", context.getTransferId());
- if (ResourceTypes.FILE.equals(this.dropboxResource.getResourceCase().name())) {
- FileMetadata metadata = dbxClientV2.files().uploadBuilder(dropboxResource.getFile().getResourcePath())
- .withMode(WriteMode.OVERWRITE)
- .uploadAndFinish(context.getStreamBuffer().getInputStream());
- logger.info("Completed Dropbox Sender stream for transfer {}", context.getTransferId());
-
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.dropboxResource.getResourceId(), this.dropboxResource.getResourceCase().name());
- throw new Exception("Resource " + this.dropboxResource.getResourceId() + " should be a FILE type. Found a " +
- this.dropboxResource.getResourceCase().name());
- }
-
}
}
diff --git a/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPMetadataCollector.java b/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPMetadataCollector.java
index 66675f3..5f6f7af 100644
--- a/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPMetadataCollector.java
+++ b/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPMetadataCollector.java
@@ -25,8 +25,11 @@
import org.apache.airavata.mft.credential.stubs.ftp.FTPSecretGetRequest;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
+import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.resource.stubs.ftp.resource.FTPResource;
import org.apache.airavata.mft.resource.stubs.ftp.resource.FTPResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.ftp.storage.FTPStorage;
+import org.apache.airavata.mft.resource.stubs.ftp.storage.FTPStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.apache.commons.net.ftp.FTPClient;
@@ -74,7 +77,7 @@
FileResourceMetadata resourceMetadata = new FileResourceMetadata();
FTPClient ftpClient = null;
try {
- ftpClient = FTPTransportUtil.getFTPClient(ftpResource, ftpSecret);
+ ftpClient = FTPTransportUtil.getFTPClient(ftpResource.getFtpStorage(), ftpSecret);
logger.info("Fetching metadata for resource {} in {}", ftpResource.getFile().getResourcePath(), ftpResource.getFtpStorage().getHost());
FTPFile ftpFile = ftpClient.mlistFile(ftpResource.getFile().getResourcePath());
@@ -99,7 +102,7 @@
}
@Override
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -108,7 +111,7 @@
throw new UnsupportedOperationException("Method not implemented"); }
@Override
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -119,12 +122,32 @@
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
FTPResource ftpResource = resourceClient.ftp().getFTPResource(FTPResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+
+ return isAvailable(ftpResource, credentialToken);
+ }
+
+ @Override
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception {
+
+ checkInitialized();
+
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ FTPStorage ftpStorage = resourceClient.ftp().getFTPStorage(FTPStorageGetRequest.newBuilder().setStorageId(storageId).build());
+
+ FTPResource ftpResource = FTPResource.newBuilder()
+ .setFile(FileResource.newBuilder().setResourcePath(resourcePath).build())
+ .setFtpStorage(ftpStorage).build();
+ return isAvailable(ftpResource, credentialToken);
+ }
+
+ public Boolean isAvailable(FTPResource ftpResource, String credentialToken) {
+
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
FTPSecret ftpSecret = secretClient.ftp().getFTPSecret(FTPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
FTPClient ftpClient = null;
try {
- ftpClient = FTPTransportUtil.getFTPClient(ftpResource, ftpSecret);
+ ftpClient = FTPTransportUtil.getFTPClient(ftpResource.getFtpStorage(), ftpSecret);
InputStream inputStream = null;
switch (ftpResource.getResourceCase().name()){
diff --git a/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPReceiver.java b/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPReceiver.java
index 9225b0e..f3b4ac8 100644
--- a/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPReceiver.java
+++ b/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPReceiver.java
@@ -26,6 +26,8 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.ftp.resource.FTPResource;
import org.apache.airavata.mft.resource.stubs.ftp.resource.FTPResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.ftp.storage.FTPStorage;
+import org.apache.airavata.mft.resource.stubs.ftp.storage.FTPStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.apache.commons.net.ftp.FTPClient;
@@ -39,21 +41,20 @@
private static final Logger logger = LoggerFactory.getLogger(FTPReceiver.class);
- private FTPResource resource;
private boolean initialized;
private FTPClient ftpClient;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
this.initialized = true;
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.resource = resourceClient.ftp().getFTPResource(FTPResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ FTPStorage ftpStorage = resourceClient.ftp().getFTPStorage(FTPStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
FTPSecret ftpSecret = secretClient.ftp().getFTPSecret(FTPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
- this.ftpClient = FTPTransportUtil.getFTPClient(this.resource, ftpSecret);
+ this.ftpClient = FTPTransportUtil.getFTPClient(ftpStorage, ftpSecret);
}
@Override
@@ -62,50 +63,43 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
- if (ResourceTypes.FILE.equals(this.resource.getResourceCase().name())) {
- logger.info("Starting FTP receiver stream for transfer {}", context.getTransferId());
+ logger.info("Starting FTP receiver stream for transfer {}", context.getTransferId());
- checkInitialized();
- OutputStream streamOs = context.getStreamBuffer().getOutputStream();
- InputStream inputStream = ftpClient.retrieveFileStream(resource.getFile().getResourcePath());
+ checkInitialized();
+ OutputStream streamOs = context.getStreamBuffer().getOutputStream();
+ InputStream inputStream = ftpClient.retrieveFileStream(targetPath);
- long fileSize = context.getMetadata().getResourceSize();
+ long fileSize = context.getMetadata().getResourceSize();
- byte[] buf = new byte[1024];
- while (true) {
- int bufSize;
+ byte[] buf = new byte[1024];
+ while (true) {
+ int bufSize;
- if (buf.length < fileSize) {
- bufSize = buf.length;
- } else {
- bufSize = (int) fileSize;
- }
- bufSize = inputStream.read(buf, 0, bufSize);
+ if (buf.length < fileSize) {
+ bufSize = buf.length;
+ } else {
+ bufSize = (int) fileSize;
+ }
+ bufSize = inputStream.read(buf, 0, bufSize);
- if (bufSize < 0) {
- break;
- }
-
- streamOs.write(buf, 0, bufSize);
- streamOs.flush();
-
- fileSize -= bufSize;
- if (fileSize == 0L)
- break;
+ if (bufSize < 0) {
+ break;
}
- inputStream.close();
- streamOs.close();
- logger.info("Completed FTP receiver stream for transfer {}", context.getTransferId());
+ streamOs.write(buf, 0, bufSize);
+ streamOs.flush();
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.resource.getResourceId(), this.resource.getResourceCase().name());
- throw new Exception("Resource " + this.resource.getResourceId() + " should be a FILE type. Found a " +
- this.resource.getResourceCase().name());
+ fileSize -= bufSize;
+ if (fileSize == 0L)
+ break;
}
+
+ inputStream.close();
+ streamOs.close();
+ logger.info("Completed FTP receiver stream for transfer {}", context.getTransferId());
+
}
private void checkInitialized() {
diff --git a/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPSender.java b/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPSender.java
index f3a55f7..5579cb8 100644
--- a/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPSender.java
+++ b/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPSender.java
@@ -18,14 +18,13 @@
package org.apache.airavata.mft.transport.ftp;
import org.apache.airavata.mft.core.ConnectorContext;
-import org.apache.airavata.mft.core.ResourceTypes;
import org.apache.airavata.mft.core.api.Connector;
import org.apache.airavata.mft.credential.stubs.ftp.FTPSecret;
import org.apache.airavata.mft.credential.stubs.ftp.FTPSecretGetRequest;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
-import org.apache.airavata.mft.resource.stubs.ftp.resource.FTPResource;
-import org.apache.airavata.mft.resource.stubs.ftp.resource.FTPResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.ftp.storage.FTPStorage;
+import org.apache.airavata.mft.resource.stubs.ftp.storage.FTPStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.apache.commons.net.ftp.FTPClient;
@@ -39,21 +38,20 @@
private static final Logger logger = LoggerFactory.getLogger(FTPReceiver.class);
- private FTPResource resource;
private boolean initialized;
private FTPClient ftpClient;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
this.initialized = true;
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.resource = resourceClient.ftp().getFTPResource(FTPResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ FTPStorage ftpStorage = resourceClient.ftp().getFTPStorage(FTPStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
FTPSecret ftpSecret = secretClient.ftp().getFTPSecret(FTPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
- this.ftpClient = FTPTransportUtil.getFTPClient(this.resource, ftpSecret);
+ this.ftpClient = FTPTransportUtil.getFTPClient(ftpStorage, ftpSecret);
}
@Override
@@ -62,7 +60,7 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting FTP sender stream for transfer {}", context.getTransferId());
@@ -70,42 +68,35 @@
logger.info("Completed FTP sender stream for transfer {}", context.getTransferId());
- if (ResourceTypes.FILE.equals(this.resource.getResourceCase().name())) {
- InputStream in = context.getStreamBuffer().getInputStream();
- long fileSize = context.getMetadata().getResourceSize();
- OutputStream outputStream = ftpClient.storeFileStream(resource.getFile().getResourcePath());
+ InputStream in = context.getStreamBuffer().getInputStream();
+ long fileSize = context.getMetadata().getResourceSize();
+ OutputStream outputStream = ftpClient.storeFileStream(targetPath);
- byte[] buf = new byte[1024];
- while (true) {
- int bufSize;
+ byte[] buf = new byte[1024];
+ while (true) {
+ int bufSize;
- if (buf.length < fileSize) {
- bufSize = buf.length;
- } else {
- bufSize = (int) fileSize;
- }
- bufSize = in.read(buf, 0, bufSize);
+ if (buf.length < fileSize) {
+ bufSize = buf.length;
+ } else {
+ bufSize = (int) fileSize;
+ }
+ bufSize = in.read(buf, 0, bufSize);
- if (bufSize < 0) {
- break;
- }
-
- outputStream.write(buf, 0, bufSize);
- outputStream.flush();
-
- fileSize -= bufSize;
- if (fileSize == 0L)
- break;
+ if (bufSize < 0) {
+ break;
}
- in.close();
- outputStream.close();
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.resource.getResourceId(), this.resource.getResourceCase().name());
- throw new Exception("Resource " + this.resource.getResourceId() + " should be a FILE type. Found a " +
- this.resource.getResourceCase().name());
+ outputStream.write(buf, 0, bufSize);
+ outputStream.flush();
+
+ fileSize -= bufSize;
+ if (fileSize == 0L)
+ break;
}
+
+ in.close();
+ outputStream.close();
}
private void checkInitialized() {
diff --git a/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPTransportUtil.java b/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPTransportUtil.java
index 7a1e3c8..988a208 100644
--- a/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPTransportUtil.java
+++ b/transport/ftp-transport/src/main/java/org/apache/airavata/mft/transport/ftp/FTPTransportUtil.java
@@ -18,7 +18,7 @@
package org.apache.airavata.mft.transport.ftp;
import org.apache.airavata.mft.credential.stubs.ftp.FTPSecret;
-import org.apache.airavata.mft.resource.stubs.ftp.resource.FTPResource;
+import org.apache.airavata.mft.resource.stubs.ftp.storage.FTPStorage;
import org.apache.commons.net.ftp.FTPClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,10 +29,10 @@
private static final Logger logger = LoggerFactory.getLogger(FTPTransportUtil.class);
- static FTPClient getFTPClient(FTPResource ftpResource, FTPSecret ftpSecret) throws IOException {
+ static FTPClient getFTPClient(FTPStorage ftpStorage, FTPSecret ftpSecret) throws IOException {
FTPClient ftpClient = new FTPClient();
- ftpClient.connect(ftpResource.getFtpStorage().getHost(), ftpResource.getFtpStorage().getPort());
+ ftpClient.connect(ftpStorage.getHost(), ftpStorage.getPort());
ftpClient.enterLocalActiveMode();
ftpClient.login(ftpSecret.getUserId(), ftpSecret.getPassword());
diff --git a/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSMetadataCollector.java b/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSMetadataCollector.java
index aa97091..0d1a1a8 100644
--- a/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSMetadataCollector.java
+++ b/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSMetadataCollector.java
@@ -33,8 +33,11 @@
import org.apache.airavata.mft.credential.stubs.gcs.GCSSecretGetRequest;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
+import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.resource.stubs.gcs.resource.GCSResource;
import org.apache.airavata.mft.resource.stubs.gcs.resource.GCSResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.GCSStorage;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.GCSStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
@@ -99,7 +102,7 @@
}
@Override
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -108,7 +111,7 @@
throw new UnsupportedOperationException("Method not implemented"); }
@Override
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -118,6 +121,24 @@
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
GCSResource gcsResource = resourceClient.gcs().getGCSResource(GCSResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ return isAvailable(gcsResource, credentialToken);
+ }
+
+ @Override
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception {
+ checkInitialized();
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ GCSStorage gcsStorage = resourceClient.gcs().getGCSStorage(GCSStorageGetRequest.newBuilder().setStorageId(storageId).build());
+
+ GCSResource gcsResource = GCSResource.newBuilder()
+ .setFile(FileResource.newBuilder().setResourcePath(resourcePath).build())
+ .setGcsStorage(gcsStorage).build();
+
+ return isAvailable(gcsResource, credentialToken);
+ }
+
+ private Boolean isAvailable(GCSResource gcsResource, String credentialToken) throws Exception {
+
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
GCSSecret gcsSecret = secretClient.gcs().getGCSSecret(GCSSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
@@ -134,10 +155,10 @@
switch (gcsResource.getResourceCase().name()){
case ResourceTypes.FILE:
return !storage.objects().get(gcsResource.getGcsStorage().getBucketName(), gcsResource.getFile().getResourcePath())
- .execute().isEmpty();
+ .execute().isEmpty();
case ResourceTypes.DIRECTORY:
return !storage.objects().get(gcsResource.getGcsStorage().getBucketName(), gcsResource.getDirectory().getResourcePath())
- .execute().isEmpty();
+ .execute().isEmpty();
}
return false;
}
diff --git a/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSReceiver.java b/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSReceiver.java
index 03fd31c..d734f92 100644
--- a/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSReceiver.java
+++ b/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSReceiver.java
@@ -33,6 +33,8 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.gcs.resource.GCSResource;
import org.apache.airavata.mft.resource.stubs.gcs.resource.GCSResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.GCSStorage;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.GCSStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -49,16 +51,18 @@
private static final Logger logger = LoggerFactory.getLogger(GCSReceiver.class);
- private GCSResource gcsResource;
private Storage storage;
+ private GCSStorage gcsStorage;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
+
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.gcsResource = resourceClient.gcs().getGCSResource(GCSResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ this.gcsStorage = resourceClient.gcs().getGCSStorage(GCSStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
GCSSecret gcsSecret = secretClient.gcs().getGCSSecret(GCSSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+
HttpTransport transport = GoogleNetHttpTransport.newTrustedTransport();
JsonFactory jsonFactory = new JacksonFactory();
String jsonString = gcsSecret.getCredentialsJson();
@@ -76,48 +80,40 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting GCS Receiver stream for transfer {}", context.getTransferId());
- if (ResourceTypes.FILE.equals(this.gcsResource.getResourceCase().name())) {
- InputStream inputStream = storage.objects().get(this.gcsResource.getGcsStorage().getBucketName(),
- this.gcsResource.getFile().getResourcePath()).executeMediaAsInputStream();
- OutputStream os = context.getStreamBuffer().getOutputStream();
- int read;
- long bytes = 0;
- long fileSize = context.getMetadata().getResourceSize();
- byte[] buf = new byte[1024];
- while (true) {
- int bufSize = 0;
+ InputStream inputStream = storage.objects().get(this.gcsStorage.getBucketName(), targetPath).executeMediaAsInputStream();
+ OutputStream os = context.getStreamBuffer().getOutputStream();
+ int read;
+ long bytes = 0;
+ long fileSize = context.getMetadata().getResourceSize();
+ byte[] buf = new byte[1024];
+ while (true) {
+ int bufSize = 0;
- if (buf.length < fileSize) {
- bufSize = buf.length;
- } else {
- bufSize = (int) fileSize;
- }
- bufSize = inputStream.read(buf, 0, bufSize);
+ if (buf.length < fileSize) {
+ bufSize = buf.length;
+ } else {
+ bufSize = (int) fileSize;
+ }
+ bufSize = inputStream.read(buf, 0, bufSize);
- if (bufSize < 0) {
- break;
- }
-
- os.write(buf, 0, bufSize);
- os.flush();
-
- fileSize -= bufSize;
- if (fileSize == 0L)
- break;
+ if (bufSize < 0) {
+ break;
}
- os.close();
+ os.write(buf, 0, bufSize);
+ os.flush();
- logger.info("Completed GCS Receiver stream for transfer {}", context.getTransferId());
-
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.gcsResource.getResourceId(), this.gcsResource.getResourceCase().name());
- throw new Exception("Resource " + this.gcsResource.getResourceId() + " should be a FILE type. Found a " +
- this.gcsResource.getResourceCase().name());
+ fileSize -= bufSize;
+ if (fileSize == 0L)
+ break;
}
+
+ os.close();
+
+ logger.info("Completed GCS Receiver stream for transfer {}", context.getTransferId());
+
}
}
diff --git a/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSSender.java b/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSSender.java
index 368d24d..4d0ef29 100644
--- a/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSSender.java
+++ b/transport/gcp-transport/src/main/java/org/apache/airavata/mft/transport/gcp/GCSSender.java
@@ -31,14 +31,13 @@
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import org.apache.airavata.mft.core.ConnectorContext;
-import org.apache.airavata.mft.core.ResourceTypes;
import org.apache.airavata.mft.core.api.Connector;
import org.apache.airavata.mft.credential.stubs.gcs.GCSSecret;
import org.apache.airavata.mft.credential.stubs.gcs.GCSSecretGetRequest;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
-import org.apache.airavata.mft.resource.stubs.gcs.resource.GCSResource;
-import org.apache.airavata.mft.resource.stubs.gcs.resource.GCSResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.GCSStorage;
+import org.apache.airavata.mft.resource.stubs.gcs.storage.GCSStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -54,15 +53,15 @@
private static final Logger logger = LoggerFactory.getLogger(GCSSender.class);
- private GCSResource gcsResource;
private Storage storage;
+ private GCSStorage gcsStorage;
private JsonObject jsonObject;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.gcsResource = resourceClient.gcs().getGCSResource(GCSResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ this.gcsStorage = resourceClient.gcs().getGCSStorage(GCSStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
GCSSecret gcsSecret = secretClient.gcs().getGCSSecret(GCSSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
@@ -86,33 +85,23 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting GCS Sender stream for transfer {}", context.getTransferId());
logger.debug("Content length for transfer {} {}", context.getTransferId(), context.getMetadata().getResourceSize());
- if (ResourceTypes.FILE.equals(this.gcsResource.getResourceCase().name())) {
+ InputStreamContent contentStream = new InputStreamContent(
+ null, context.getStreamBuffer().getInputStream());
+ String entityUser = jsonObject.get("client_email").getAsString();
+ StorageObject objectMetadata = new StorageObject()
+ // Set the destination object name
+ .setName(targetPath)
+ // Set the access control list to publicly read-only
+ .setAcl(Arrays.asList(new ObjectAccessControl().setEntity("user-" + entityUser).setRole("OWNER")));
- InputStreamContent contentStream = new InputStreamContent(
- null, context.getStreamBuffer().getInputStream());
- String entityUser = jsonObject.get("client_email").getAsString();
- StorageObject objectMetadata = new StorageObject()
- // Set the destination object name
- .setName(this.gcsResource.getFile().getResourcePath())
- // Set the access control list to publicly read-only
- .setAcl(Arrays.asList(new ObjectAccessControl().setEntity("user-" + entityUser).setRole("OWNER")));
+ Insert insertRequest = storage.objects().insert(this.gcsStorage.getBucketName(), objectMetadata, contentStream);
- Insert insertRequest = storage.objects().insert(this.gcsResource.getGcsStorage().getBucketName(), objectMetadata, contentStream);
+ insertRequest.execute();
- insertRequest.execute();
-
- logger.info("Completed GCS Sender stream for transfer {}", context.getTransferId());
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.gcsResource.getResourceId(), this.gcsResource.getResourceCase().name());
- throw new Exception("Resource " + this.gcsResource.getResourceId() + " should be a FILE type. Found a " +
- this.gcsResource.getResourceCase().name());
- }
-
-
+ logger.info("Completed GCS Sender stream for transfer {}", context.getTransferId());
}
}
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalMetadataCollector.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalMetadataCollector.java
index f24c70f..9faa70a 100644
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalMetadataCollector.java
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalMetadataCollector.java
@@ -23,8 +23,11 @@
import org.apache.airavata.mft.core.api.MetadataCollector;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
+import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.resource.stubs.local.resource.LocalResource;
import org.apache.airavata.mft.resource.stubs.local.resource.LocalResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.local.storage.LocalStorage;
+import org.apache.airavata.mft.resource.stubs.local.storage.LocalStorageGetRequest;
import java.io.File;
import java.io.FileInputStream;
@@ -92,7 +95,7 @@
}
@Override
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -101,15 +104,29 @@
throw new UnsupportedOperationException("Method not implemented"); }
@Override
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@Override
public Boolean isAvailable(String resourceId, String credentialToken) throws Exception {
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
LocalResource localResource = resourceClient.local().getLocalResource(LocalResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ return isAvailable(localResource, credentialToken);
+ }
+ @Override
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception {
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ LocalStorage localStorage = resourceClient.local().getLocalStorage(LocalStorageGetRequest.newBuilder().setStorageId(storageId).build());
+
+ LocalResource localResource = LocalResource.newBuilder()
+ .setFile(FileResource.newBuilder().setResourcePath(resourcePath).build())
+ .setLocalStorage(localStorage).build();
+ return isAvailable(localResource, credentialToken);
+ }
+
+ public Boolean isAvailable(LocalResource localResource, String credentialToken) throws Exception {
switch (localResource.getResourceCase().name()){
case ResourceTypes.FILE:
return new File(localResource.getFile().getResourcePath()).exists();
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalReceiver.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalReceiver.java
index 1b61fb1..53776ff 100644
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalReceiver.java
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalReceiver.java
@@ -18,12 +18,7 @@
package org.apache.airavata.mft.transport.local;
import org.apache.airavata.mft.core.ConnectorContext;
-import org.apache.airavata.mft.core.ResourceTypes;
import org.apache.airavata.mft.core.api.Connector;
-import org.apache.airavata.mft.resource.client.ResourceServiceClient;
-import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
-import org.apache.airavata.mft.resource.stubs.local.resource.LocalResource;
-import org.apache.airavata.mft.resource.stubs.local.resource.LocalResourceGetRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,16 +28,12 @@
private static final Logger logger = LoggerFactory.getLogger(LocalReceiver.class);
- private LocalResource resource;
private boolean initialized;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort,
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort,
String secretServiceHost, int secretServicePort) throws Exception {
this.initialized = true;
-
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.resource = resourceClient.local().getLocalResource(LocalResourceGetRequest.newBuilder().setResourceId(resourceId).build());
}
@Override
@@ -57,50 +48,41 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting local receiver stream for transfer {}", context.getTransferId());
checkInitialized();
+ OutputStream streamOs = context.getStreamBuffer().getOutputStream();
+ FileInputStream fis = new FileInputStream(new File(targetPath));
- if (ResourceTypes.FILE.equals(this.resource.getResourceCase().name())) {
- OutputStream streamOs = context.getStreamBuffer().getOutputStream();
- FileInputStream fis = new FileInputStream(new File(resource.getFile().getResourcePath()));
+ long fileSize = context.getMetadata().getResourceSize();
- long fileSize = context.getMetadata().getResourceSize();
+ byte[] buf = new byte[1024];
+ while (true) {
+ int bufSize = 0;
- byte[] buf = new byte[1024];
- while (true) {
- int bufSize = 0;
+ if (buf.length < fileSize) {
+ bufSize = buf.length;
+ } else {
+ bufSize = (int) fileSize;
+ }
+ bufSize = fis.read(buf, 0, bufSize);
- if (buf.length < fileSize) {
- bufSize = buf.length;
- } else {
- bufSize = (int) fileSize;
- }
- bufSize = fis.read(buf, 0, bufSize);
-
- if (bufSize < 0) {
- break;
- }
-
- streamOs.write(buf, 0, bufSize);
- streamOs.flush();
-
- fileSize -= bufSize;
- if (fileSize == 0L)
- break;
+ if (bufSize < 0) {
+ break;
}
- fis.close();
- streamOs.close();
- logger.info("Completed local receiver stream for transfer {}", context.getTransferId());
+ streamOs.write(buf, 0, bufSize);
+ streamOs.flush();
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.resource.getResourceId(), this.resource.getResourceCase().name());
- throw new Exception("Resource " + this.resource.getResourceId() + " should be a FILE type. Found a " +
- this.resource.getResourceCase().name());
+ fileSize -= bufSize;
+ if (fileSize == 0L)
+ break;
}
+
+ fis.close();
+ streamOs.close();
+ logger.info("Completed local receiver stream for transfer {}", context.getTransferId());
}
}
diff --git a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalSender.java b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalSender.java
index 315fa25..e4d5e48 100644
--- a/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalSender.java
+++ b/transport/local-transport/src/main/java/org/apache/airavata/mft/transport/local/LocalSender.java
@@ -18,12 +18,7 @@
package org.apache.airavata.mft.transport.local;
import org.apache.airavata.mft.core.ConnectorContext;
-import org.apache.airavata.mft.core.ResourceTypes;
import org.apache.airavata.mft.core.api.Connector;
-import org.apache.airavata.mft.resource.client.ResourceServiceClient;
-import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
-import org.apache.airavata.mft.resource.stubs.local.resource.LocalResource;
-import org.apache.airavata.mft.resource.stubs.local.resource.LocalResourceGetRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,16 +28,12 @@
private static final Logger logger = LoggerFactory.getLogger(LocalSender.class);
- private LocalResource resource;
private boolean initialized;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort,
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort,
String secretServiceHost, int secretServicePort) throws Exception {
this.initialized = true;
-
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.resource = resourceClient.local().getLocalResource(LocalResourceGetRequest.newBuilder().setResourceId(resourceId).build());
}
@Override
@@ -57,49 +48,42 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting local sender stream for transfer {}", context.getTransferId());
checkInitialized();
- if (ResourceTypes.FILE.equals(this.resource.getResourceCase().name())) {
- InputStream in = context.getStreamBuffer().getInputStream();
- long fileSize = context.getMetadata().getResourceSize();
- OutputStream fos = new FileOutputStream(resource.getFile().getResourcePath());
+ InputStream in = context.getStreamBuffer().getInputStream();
+ long fileSize = context.getMetadata().getResourceSize();
+ OutputStream fos = new FileOutputStream(targetPath);
- byte[] buf = new byte[1024];
- while (true) {
- int bufSize = 0;
+ byte[] buf = new byte[1024];
+ while (true) {
+ int bufSize = 0;
- if (buf.length < fileSize) {
- bufSize = buf.length;
- } else {
- bufSize = (int) fileSize;
- }
- bufSize = in.read(buf, 0, bufSize);
+ if (buf.length < fileSize) {
+ bufSize = buf.length;
+ } else {
+ bufSize = (int) fileSize;
+ }
+ bufSize = in.read(buf, 0, bufSize);
- if (bufSize < 0) {
- break;
- }
-
- fos.write(buf, 0, bufSize);
- fos.flush();
-
- fileSize -= bufSize;
- if (fileSize == 0L)
- break;
+ if (bufSize < 0) {
+ break;
}
- in.close();
- fos.close();
+ fos.write(buf, 0, bufSize);
+ fos.flush();
- logger.info("Completed local sender stream for transfer {}", context.getTransferId());
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.resource.getResourceId(), this.resource.getResourceCase().name());
- throw new Exception("Resource " + this.resource.getResourceId() + " should be a FILE type. Found a " +
- this.resource.getResourceCase().name());
+ fileSize -= bufSize;
+ if (fileSize == 0L)
+ break;
}
+
+ in.close();
+ fos.close();
+
+ logger.info("Completed local sender stream for transfer {}", context.getTransferId());
}
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3MetadataCollector.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3MetadataCollector.java
index 572affd..58be5b6 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3MetadataCollector.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3MetadataCollector.java
@@ -30,8 +30,11 @@
import org.apache.airavata.mft.credential.stubs.s3.S3SecretGetRequest;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
+import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.resource.stubs.s3.resource.S3Resource;
import org.apache.airavata.mft.resource.stubs.s3.resource.S3ResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.s3.storage.S3Storage;
+import org.apache.airavata.mft.resource.stubs.s3.storage.S3StorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
@@ -85,7 +88,7 @@
}
@Override
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -94,7 +97,7 @@
throw new UnsupportedOperationException("Method not implemented"); }
@Override
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
throw new UnsupportedOperationException("Method not implemented");
}
@@ -105,6 +108,23 @@
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
S3Resource s3Resource = resourceClient.s3().getS3Resource(S3ResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ return isAvailable(s3Resource, credentialToken);
+ }
+
+ @Override
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception {
+ checkInitialized();
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ S3Storage s3Storage = resourceClient.s3().getS3Storage(S3StorageGetRequest.newBuilder().setStorageId(storageId).build());
+ S3Resource s3Resource = S3Resource.newBuilder()
+ .setFile(FileResource.newBuilder().setResourcePath(resourcePath).build())
+ .setS3Storage(s3Storage).build();
+
+ return isAvailable(s3Resource, credentialToken);
+ }
+
+ private Boolean isAvailable(S3Resource s3Resource, String credentialToken) throws Exception {
+
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
S3Secret s3Secret = secretClient.s3().getS3Secret(S3SecretGetRequest.newBuilder().setSecretId(credentialToken).build());
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
index ddfd425..2e7e816 100644
--- 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
@@ -32,6 +32,8 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.s3.resource.S3Resource;
import org.apache.airavata.mft.resource.stubs.s3.resource.S3ResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.s3.storage.S3Storage;
+import org.apache.airavata.mft.resource.stubs.s3.storage.S3StorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -43,15 +45,15 @@
private static final Logger logger = LoggerFactory.getLogger(S3Receiver.class);
- private S3Resource s3Resource;
private AmazonS3 s3Client;
+ private S3Storage s3Storage;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort,
- String secretServiceHost, int secretServicePort) throws Exception {
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort,
+ String secretServiceHost, int secretServicePort) throws Exception {
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.s3Resource = resourceClient.s3().getS3Resource(S3ResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ this.s3Storage = resourceClient.s3().getS3Storage(S3StorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
S3Secret s3Secret = secretClient.s3().getS3Secret(S3SecretGetRequest.newBuilder().setSecretId(credentialToken).build());
@@ -59,7 +61,7 @@
s3Client = AmazonS3ClientBuilder.standard()
.withCredentials(new AWSStaticCredentialsProvider(awsCreds))
- .withRegion(s3Resource.getS3Storage().getRegion())
+ .withRegion(s3Storage.getRegion())
.build();
}
@@ -69,31 +71,23 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
- if (ResourceTypes.FILE.equals(this.s3Resource.getResourceCase().name())) {
- logger.info("Starting S3 Receiver stream for transfer {}", context.getTransferId());
+ logger.info("Starting S3 Receiver stream for transfer {}", context.getTransferId());
- S3Object s3object = s3Client.getObject(s3Resource.getS3Storage().getBucketName(), s3Resource.getFile().getResourcePath());
- S3ObjectInputStream inputStream = s3object.getObjectContent();
+ S3Object s3object = s3Client.getObject(this.s3Storage.getBucketName(), targetPath);
+ S3ObjectInputStream inputStream = s3object.getObjectContent();
- OutputStream os = context.getStreamBuffer().getOutputStream();
- int read;
- long bytes = 0;
- while ((read = inputStream.read()) != -1) {
- bytes++;
- os.write(read);
- }
- os.flush();
- os.close();
-
- logger.info("Completed S3 Receiver stream for transfer {}", context.getTransferId());
-
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.s3Resource.getResourceId(), this.s3Resource.getResourceCase().name());
- throw new Exception("Resource " + this.s3Resource.getResourceId() + " should be a FILE type. Found a " +
- this.s3Resource.getResourceCase().name());
+ OutputStream os = context.getStreamBuffer().getOutputStream();
+ int read;
+ long bytes = 0;
+ while ((read = inputStream.read()) != -1) {
+ bytes++;
+ os.write(read);
}
+ os.flush();
+ os.close();
+
+ logger.info("Completed S3 Receiver stream for transfer {}", context.getTransferId());
}
}
diff --git a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
index 095ed3c..a7a640c 100644
--- a/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
+++ b/transport/s3-transport/src/main/java/org/apache/airavata/mft/transport/s3/S3Sender.java
@@ -31,6 +31,8 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.s3.resource.S3Resource;
import org.apache.airavata.mft.resource.stubs.s3.resource.S3ResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.s3.storage.S3Storage;
+import org.apache.airavata.mft.resource.stubs.s3.storage.S3StorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -41,13 +43,13 @@
private static final Logger logger = LoggerFactory.getLogger(S3Sender.class);
private AmazonS3 s3Client;
- private S3Resource s3Resource;
+ private S3Storage s3Storage;
@Override
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort, String secretServiceHost, int secretServicePort) throws Exception {
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.s3Resource = resourceClient.s3().getS3Resource(S3ResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ this.s3Storage = resourceClient.s3().getS3Storage(S3StorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
S3Secret s3Secret = secretClient.s3().getS3Secret(S3SecretGetRequest.newBuilder().setSecretId(credentialToken).build());
@@ -55,7 +57,7 @@
s3Client = AmazonS3ClientBuilder.standard()
.withCredentials(new AWSStaticCredentialsProvider(awsCreds))
- .withRegion(s3Resource.getS3Storage().getRegion())
+ .withRegion(s3Storage.getRegion())
.build();
}
@@ -65,23 +67,14 @@
}
@Override
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
logger.info("Starting S3 Sender stream for transfer {}", context.getTransferId());
logger.info("Content length for transfer {} {}", context.getTransferId(), context.getMetadata().getResourceSize());
ObjectMetadata metadata = new ObjectMetadata();
metadata.setContentLength(context.getMetadata().getResourceSize());
- if (ResourceTypes.FILE.equals(this.s3Resource.getResourceCase().name())) {
- s3Client.putObject(this.s3Resource.getS3Storage().getBucketName(), this.s3Resource.getFile().getResourcePath(),
- context.getStreamBuffer().getInputStream(), metadata);
- logger.info("Completed S3 Sender stream for transfer {}", context.getTransferId());
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.s3Resource.getResourceId(), this.s3Resource.getResourceCase().name());
- throw new Exception("Resource " + this.s3Resource.getResourceId() + " should be a FILE type. Found a " +
- this.s3Resource.getResourceCase().name());
- }
-
+ s3Client.putObject(this.s3Storage.getBucketName(), targetPath, context.getStreamBuffer().getInputStream(), metadata);
+ logger.info("Completed S3 Sender stream for transfer {}", context.getTransferId());
}
}
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
index d0086a5..ac09acb 100644
--- 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
@@ -40,6 +40,9 @@
import org.apache.airavata.mft.resource.stubs.common.FileResource;
import org.apache.airavata.mft.resource.stubs.scp.resource.SCPResource;
import org.apache.airavata.mft.resource.stubs.scp.resource.SCPResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.scp.storage.SCPStorage;
+import org.apache.airavata.mft.resource.stubs.scp.storage.SCPStorageCreateRequest;
+import org.apache.airavata.mft.resource.stubs.scp.storage.SCPStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.apache.commons.io.IOUtils;
@@ -76,7 +79,7 @@
}
}
- private FileResourceMetadata getFileResourceMetadata(SCPResource scpResource, SCPSecret scpSecret, String parentResourceId) throws Exception {
+ private FileResourceMetadata getFileResourceMetadata(SCPResource scpResource, SCPSecret scpSecret) throws Exception {
try (SSHClient sshClient = getSSHClient(scpResource, scpSecret)) {
logger.info("Fetching metadata for resource {} in {}", scpResource.getFile().getResourcePath(), scpResource.getScpStorage().getHost());
@@ -89,8 +92,6 @@
metadata.setResourceSize(lstat.getSize());
metadata.setCreatedTime(lstat.getAtime());
metadata.setUpdateTime(lstat.getMtime());
- metadata.setParentResourceId(parentResourceId);
- metadata.setParentResourceType("SCP");
metadata.setFriendlyName(new File(scpResource.getFile().getResourcePath()).getName());
metadata.setResourcePath(scpResource.getFile().getResourcePath());
@@ -127,28 +128,26 @@
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
- return getFileResourceMetadata(scpResource, scpSecret, resourceId);
+ return getFileResourceMetadata(scpResource, scpSecret);
}
@Override
- public FileResourceMetadata getFileResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public FileResourceMetadata getFileResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- SCPResource parentSCPResource = resourceClient.scp().getSCPResource(SCPResourceGetRequest.newBuilder().setResourceId(parentResourceId).build());
+ SCPStorage scpStorage = resourceClient.scp().getSCPStorage(SCPStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
- validateParent(parentSCPResource, resourcePath);
-
SCPResource scpResource = SCPResource.newBuilder()
.setFile(FileResource.newBuilder()
.setResourcePath(resourcePath).build())
- .setScpStorage(parentSCPResource.getScpStorage()).build();
+ .setScpStorage(scpStorage).build();
- return getFileResourceMetadata(scpResource, scpSecret, parentResourceId);
+ return getFileResourceMetadata(scpResource, scpSecret);
}
- private DirectoryResourceMetadata getDirectoryResourceMetadata(SCPResource scpResource, SCPSecret scpSecret, String parentResourceId) throws Exception {
+ private DirectoryResourceMetadata getDirectoryResourceMetadata(SCPResource scpResource, SCPSecret scpSecret) throws Exception {
try (SSHClient sshClient = getSSHClient(scpResource, scpSecret)) {
logger.info("Fetching metadata for resource {} in {}", scpResource.getFile().getResourcePath(), scpResource.getScpStorage().getHost());
@@ -167,9 +166,7 @@
.withFriendlyName(rri.getName())
.withResourcePath(rri.getPath())
.withCreatedTime(rri.getAttributes().getAtime())
- .withUpdateTime(rri.getAttributes().getMtime())
- .withParentResourceId(parentResourceId)
- .withParentResourceType("SCP");
+ .withUpdateTime(rri.getAttributes().getMtime());
dirMetadataBuilder = dirMetadataBuilder.withDirectory(childDirBuilder.build());
}
@@ -178,20 +175,16 @@
.withFriendlyName(rri.getName())
.withResourcePath(rri.getPath())
.withCreatedTime(rri.getAttributes().getAtime())
- .withUpdateTime(rri.getAttributes().getMtime())
- .withParentResourceId(parentResourceId)
- .withParentResourceType("SCP");
+ .withUpdateTime(rri.getAttributes().getMtime());
dirMetadataBuilder = dirMetadataBuilder.withFile(childFileBuilder.build());
}
}
dirMetadataBuilder = dirMetadataBuilder.withFriendlyName(new File(scpResource.getDirectory().getResourcePath()).getName())
- .withResourcePath(parentResourceId)
+ .withResourcePath(scpResource.getDirectory().getResourcePath())
.withCreatedTime(lsStat.getAtime())
- .withUpdateTime(lsStat.getMtime())
- .withParentResourceId(parentResourceId)
- .withParentResourceType("SCP");
+ .withUpdateTime(lsStat.getMtime());
return dirMetadataBuilder.build();
}
}
@@ -206,24 +199,23 @@
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
- return getDirectoryResourceMetadata(scpPResource, scpSecret, resourceId);
+ return getDirectoryResourceMetadata(scpPResource, scpSecret);
}
@Override
- public DirectoryResourceMetadata getDirectoryResourceMetadata(String parentResourceId, String resourcePath, String credentialToken) throws Exception {
+ public DirectoryResourceMetadata getDirectoryResourceMetadata(String storageId, String resourcePath, String credentialToken) throws Exception {
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- SCPResource parentSCPPResource = resourceClient.scp().getSCPResource(SCPResourceGetRequest.newBuilder().setResourceId(parentResourceId).build());
+ SCPStorage scpStorage = resourceClient.scp().getSCPStorage(SCPStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
- validateParent(parentSCPPResource, resourcePath);
- SCPResource scpResource = SCPResource.newBuilder().setScpStorage(parentSCPPResource.getScpStorage())
+ SCPResource scpResource = SCPResource.newBuilder().setScpStorage(scpStorage)
.setDirectory(DirectoryResource.newBuilder()
.setResourcePath(resourcePath).build()).build();
- return getDirectoryResourceMetadata(scpResource, scpSecret, parentResourceId);
+ return getDirectoryResourceMetadata(scpResource, scpSecret);
}
@Override
@@ -233,6 +225,23 @@
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
SCPResource scpResource = resourceClient.scp().getSCPResource(SCPResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ return isAvailable(scpResource, credentialToken);
+ }
+
+ @Override
+ public Boolean isAvailable(String storageId, String resourcePath, String credentialToken) throws Exception {
+ checkInitialized();
+ ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
+ SCPStorage scpStorage = resourceClient.scp().getSCPStorage(SCPStorageGetRequest.newBuilder().setStorageId(storageId).build());
+ SCPResource scpResource = SCPResource.newBuilder()
+ .setFile(FileResource.newBuilder().setResourcePath(resourcePath).build())
+ .setScpStorage(scpStorage).build();
+
+ return isAvailable(scpResource, credentialToken);
+ }
+
+ public Boolean isAvailable(SCPResource scpResource, String credentialToken) throws Exception {
+
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
index 127ea8c..516ff7d 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
@@ -21,14 +21,13 @@
import com.jcraft.jsch.Session;
import org.apache.airavata.mft.core.ConnectorContext;
import org.apache.airavata.mft.core.DoubleStreamingBuffer;
-import org.apache.airavata.mft.core.ResourceTypes;
import org.apache.airavata.mft.core.api.Connector;
import org.apache.airavata.mft.credential.stubs.scp.SCPSecret;
import org.apache.airavata.mft.credential.stubs.scp.SCPSecretGetRequest;
import org.apache.airavata.mft.resource.client.ResourceServiceClient;
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
-import org.apache.airavata.mft.resource.stubs.scp.resource.SCPResource;
-import org.apache.airavata.mft.resource.stubs.scp.resource.SCPResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.scp.storage.SCPStorage;
+import org.apache.airavata.mft.resource.stubs.scp.storage.SCPStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -43,30 +42,37 @@
boolean initialized = false;
private Session session;
- private SCPResource scpResource;
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort,
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort,
String secretServiceHost, int secretServicePort) throws Exception {
+ if (initialized) {
+ destroy();
+ }
+
this.initialized = true;
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.scpResource = resourceClient.scp().getSCPResource(SCPResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ SCPStorage scpStorage = resourceClient.scp().getSCPStorage(SCPStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
this.session = SCPTransportUtil.createSession(
- scpResource.getScpStorage().getUser(),
- scpResource.getScpStorage().getHost(),
- scpResource.getScpStorage().getPort(),
+ scpStorage.getUser(),
+ scpStorage.getHost(),
+ scpStorage.getPort(),
scpSecret.getPrivateKey().getBytes(),
scpSecret.getPublicKey().getBytes(),
scpSecret.getPassphrase().equals("")? null : scpSecret.getPassphrase().getBytes());
}
public void destroy() {
-
+ try {
+ this.session.disconnect();
+ } catch (Exception e) {
+ logger.error("Errored while disconnecting session", e);
+ }
}
private void checkInitialized() {
@@ -75,23 +81,16 @@
}
}
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
checkInitialized();
if (session == null) {
logger.error("Session can not be null. Make sure that SCP Receiver is properly initialized");
throw new Exception("Session can not be null. Make sure that SCP Receiver is properly initialized");
}
- if (ResourceTypes.FILE.equals(this.scpResource.getResourceCase().name())) {
- transferRemoteToStream(session, this.scpResource.getFile().getResourcePath(), context.getStreamBuffer());
- logger.info("SCP Receive completed. Transfer {}", context.getTransferId());
+ transferRemoteToStream(session, targetPath, context.getStreamBuffer());
+ logger.info("SCP Receive completed. Transfer {}", context.getTransferId());
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.scpResource.getResourceId(), this.scpResource.getResourceCase().name());
- throw new Exception("Resource " + this.scpResource.getResourceId() + " should be a FILE type. Found a " +
- this.scpResource.getResourceCase().name());
- }
}
private void transferRemoteToStream(Session session, String from, DoubleStreamingBuffer streamBuffer) throws Exception {
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
index b552f0e..01b8ed4 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
@@ -30,6 +30,8 @@
import org.apache.airavata.mft.resource.client.ResourceServiceClientBuilder;
import org.apache.airavata.mft.resource.stubs.scp.resource.SCPResource;
import org.apache.airavata.mft.resource.stubs.scp.resource.SCPResourceGetRequest;
+import org.apache.airavata.mft.resource.stubs.scp.storage.SCPStorage;
+import org.apache.airavata.mft.resource.stubs.scp.storage.SCPStorageGetRequest;
import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.client.SecretServiceClientBuilder;
import org.slf4j.Logger;
@@ -44,27 +46,28 @@
boolean initialized = false;
private Session session;
- private SCPResource scpResource;
- public void init(String resourceId, String credentialToken, String resourceServiceHost, int resourceServicePort,
+ public void init(String storageId, String credentialToken, String resourceServiceHost, int resourceServicePort,
String secretServiceHost, int secretServicePort) throws Exception {
+ if (initialized) {
+ destroy();
+ }
+
this.initialized = true;
ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- this.scpResource = resourceClient.scp().getSCPResource(SCPResourceGetRequest.newBuilder().setResourceId(resourceId).build());
+ SCPStorage scpStorage = resourceClient.scp().getSCPStorage(SCPStorageGetRequest.newBuilder().setStorageId(storageId).build());
SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
- logger.info("Creating a ssh session for {}@{}:{}",
- scpResource.getScpStorage().getUser(), scpResource.getScpStorage().getHost(),
- scpResource.getScpStorage().getPort());
+ logger.info("Creating a ssh session for {}@{}:{}", scpStorage.getUser(), scpStorage.getHost(), scpStorage.getPort());
this.session = SCPTransportUtil.createSession(
- scpResource.getScpStorage().getUser(),
- scpResource.getScpStorage().getHost(),
- scpResource.getScpStorage().getPort(),
+ scpStorage.getUser(),
+ scpStorage.getHost(),
+ scpStorage.getPort(),
scpSecret.getPrivateKey().getBytes(),
scpSecret.getPublicKey().getBytes(),
scpSecret.getPassphrase().equals("")? null : scpSecret.getPassphrase().getBytes());
@@ -72,11 +75,11 @@
public void destroy() {
- //try {
- // this.session.disconnect();
- //} catch (Exception e) {
- // logger.error("Errored while disconnecting session", e);
- //}
+ try {
+ this.session.disconnect();
+ } catch (Exception e) {
+ logger.error("Errored while disconnecting session", e);
+ }
}
private void checkInitialized() {
@@ -85,7 +88,7 @@
}
}
- public void startStream(ConnectorContext context) throws Exception {
+ public void startStream(String targetPath, ConnectorContext context) throws Exception {
checkInitialized();
if (session == null) {
@@ -93,16 +96,8 @@
throw new Exception("Session can not be null. Make sure that SCP Sender is properly initialized");
}
try {
- if (ResourceTypes.FILE.equals(this.scpResource.getResourceCase().name())) {
- copyLocalToRemote(this.session, this.scpResource.getFile().getResourcePath(), context.getStreamBuffer(), context.getMetadata().getResourceSize());
- logger.info("SCP send to transfer {} completed", context.getTransferId());
-
- } else {
- logger.error("Resource {} should be a FILE type. Found a {}",
- this.scpResource.getResourceId(), this.scpResource.getResourceCase().name());
- throw new Exception("Resource " + this.scpResource.getResourceId() + " should be a FILE type. Found a " +
- this.scpResource.getResourceCase().name());
- }
+ copyLocalToRemote(this.session, targetPath, context.getStreamBuffer(), context.getMetadata().getResourceSize());
+ logger.info("SCP send to transfer {} completed", context.getTransferId());
} catch (Exception e) {
logger.error("Errored while streaming to remote scp server. Transfer {}", context.getTransferId() , e);
