lib/src/main/java/edu/utexas/tacc/tapis/files/lib/services/ChildTaskTransferService.java
private TransferTaskChild chevronTwo(TransferTaskChild taskChild) throws ServiceException, NotFoundException, IOException {

        //If we are cancelled/failed we can skip the transfer
        if (taskChild.isTerminal()) return taskChild;

        TapisSystem sourceSystem = null;
        TapisSystem destSystem = null;
        IRemoteDataClient sourceClient;
        IRemoteDataClient destClient;
        log.info("***** DOING chevronTwo **** {}", taskChild);

        //Step 1: Update task in DB to IN_PROGRESS and increment the retries on this particular task
        try {
            taskChild.setStatus(TransferTaskStatus.IN_PROGRESS);
            taskChild.setRetries(taskChild.getRetries() + 1);
            taskChild = dao.updateTransferTaskChild(taskChild);
        } catch (DAOException ex) {
            throw new ServiceException(Utils.getMsg("FILES_TXFR_SVC_ERR1", taskChild.getTenantId(), taskChild.getUsername(),
                "chevronTwoA", taskChild.getId(), taskChild.getUuid(), ex.getMessage()), ex);
        }

        String sourcePath;
        TransferURI destURL = taskChild.getDestinationURI();
        TransferURI sourceURL = taskChild.getSourceURI();

        if (taskChild.getSourceURI().toString().startsWith("https://") || taskChild.getSourceURI().toString().startsWith("http://")) {
            sourceClient = new HTTPClient(taskChild.getTenantId(), taskChild.getUsername(), sourceURL.toString(), destURL.toString());
            //This should be the full string URL such as http://google.com
            sourcePath = sourceURL.toString();
        } else {
            sourcePath = sourceURL.getPath();
            sourceSystem = systemsCache.getSystem(taskChild.getTenantId(), sourceURL.getSystemId(), taskChild.getUsername());
            if (sourceSystem.getEnabled() == null || !sourceSystem.getEnabled()) {
                String msg = Utils.getMsg("FILES_TXFR_SYS_NOTENABLED", taskChild.getTenantId(),
                    taskChild.getUsername(), taskChild.getId(), taskChild.getUuid(), sourceSystem.getId());
                throw new ServiceException(msg);
            }
            sourceClient = remoteDataClientFactory.getRemoteDataClient(taskChild.getTenantId(), taskChild.getUsername(),
                sourceSystem, taskChild.getUsername());

        }

        //Step 2: Get clients for source / dest
        try {
            destSystem = systemsCache.getSystem(taskChild.getTenantId(), destURL.getSystemId(), taskChild.getUsername());
            if (destSystem.getEnabled() == null || !destSystem.getEnabled()) {
                String msg = Utils.getMsg("FILES_TXFR_SYS_NOTENABLED", taskChild.getTenantId(),
                    taskChild.getUsername(), taskChild.getId(), taskChild.getUuid(), destSystem.getId());
                throw new ServiceException(msg);
            }
            destClient = remoteDataClientFactory.getRemoteDataClient(taskChild.getTenantId(), taskChild.getUsername(),
                destSystem, taskChild.getUsername());
        } catch (IOException | ServiceException ex) {
            throw new ServiceException(Utils.getMsg("FILES_TXFR_SVC_ERR1", taskChild.getTenantId(), taskChild.getUsername(),
                "chevronTwoB", taskChild.getId(), taskChild.getUuid(), ex.getMessage()), ex);
        }


        // If we have a directory to create, just do that and return the task child
        if (taskChild.isDir()) {
            destClient.mkdir(destURL.getPath());
            return taskChild;
        }

        //Step 4: Stream the file contents to dest. While the InputStream is open,
        // we put a tap on it and send events that get grouped into 100 ms intervals. Progress
        // on the child tasks are updated during the reading of the source input stream.
        try (InputStream sourceStream = sourceClient.getStream(sourcePath);
             ObservableInputStream observableInputStream = new ObservableInputStream(sourceStream)
        ) {
            // Observe the progress event stream, just get the last event from
            // the past 1 second.
            final TransferTaskChild finalTaskChild = taskChild;
            observableInputStream.getEventStream()
                .window(Duration.ofMillis(100))
                .flatMap(window -> window.takeLast(1))
                .flatMap((progress) -> this.updateProgress(progress, finalTaskChild))
                .subscribe();
            destClient.insert(destURL.getPath(), observableInputStream);
        }


        //If its an executable file on a posix system, chmod it to be +x. For HTTP inputs, there is no sourceSystem, so we have to check that.
        if (sourceSystem != null && Objects.equals(sourceSystem.getSystemType(), SystemTypeEnum.LINUX) && Objects.equals(destSystem.getSystemType(), SystemTypeEnum.LINUX)) {
            List<FileInfo> itemListing = sourceClient.ls(sourcePath);
            FileInfo item = itemListing.get(0);
            if (!item.isDir() && item.getNativePermissions().contains("x")) {
                try {
                    fileUtilsService.linuxOp(destClient, destURL.getPath(), FileUtilsService.NativeLinuxOperation.CHMOD, "700", false);
                } catch (TapisException ex) {
                    String msg = Utils.getMsg("FILES_TXFR_SVC_ERR1", taskChild.getTenantId(), taskChild.getUsername(),
                        "chmod", taskChild.getId(), taskChild.getUuid(), ex.getMessage());
                    log.error(msg);
                    throw new ServiceException(msg, ex);
                }
            }
        }


        //The ChildTransferTask gets updated in another thread so we look it up again here
        //before passing it on
        try {
            taskChild = dao.getTransferTaskChild(taskChild.getUuid());
            return taskChild;
        } catch (DAOException ex) {
            return null;
        }
    }
/**
     * The meat of the operation.
     *
     * @param taskChild the incoming child task
     * @return update child task
     * @throws ServiceException If the DAO updates failed or a transfer failed in flight
     */
The meat of the operation.
@param taskChild the incoming child task
@return update child task
@throws ServiceException If the DAO updates failed or a transfer failed in flight
The meat of the operation . @param taskChild the incoming child task @return update child task @throws ServiceException If the DAO updates failed or a transfer failed in flight
private TransferTaskChild chevronTwo(TransferTaskChild taskChild) throws ServiceException, NotFoundException, IOException {
        if (taskChild.isTerminal()) return taskChild;
        TapisSystem sourceSystem = null;
        TapisSystem destSystem = null;
        IRemoteDataClient sourceClient;
        IRemoteDataClient destClient;
        log.info("***** DOING chevronTwo **** {}", taskChild);
        try {
            taskChild.setStatus(TransferTaskStatus.IN_PROGRESS);
            taskChild.setRetries(taskChild.getRetries() + 1);
            taskChild = dao.updateTransferTaskChild(taskChild);
        } catch (DAOException ex) {
            throw new ServiceException(Utils.getMsg("FILES_TXFR_SVC_ERR1", taskChild.getTenantId(), taskChild.getUsername(),
                "chevronTwoA", taskChild.getId(), taskChild.getUuid(), ex.getMessage()), ex);
        }
        String sourcePath;
        TransferURI destURL = taskChild.getDestinationURI();
        TransferURI sourceURL = taskChild.getSourceURI();
        if (taskChild.getSourceURI().toString().startsWith("https://") || taskChild.getSourceURI().toString().startsWith("http://")) {
            sourceClient = new HTTPClient(taskChild.getTenantId(), taskChild.getUsername(), sourceURL.toString(), destURL.toString());
            sourcePath = sourceURL.toString();
        } else {
            sourcePath = sourceURL.getPath();
            sourceSystem = systemsCache.getSystem(taskChild.getTenantId(), sourceURL.getSystemId(), taskChild.getUsername());
            if (sourceSystem.getEnabled() == null || !sourceSystem.getEnabled()) {
                String msg = Utils.getMsg("FILES_TXFR_SYS_NOTENABLED", taskChild.getTenantId(),
                    taskChild.getUsername(), taskChild.getId(), taskChild.getUuid(), sourceSystem.getId());
                throw new ServiceException(msg);
            }
            sourceClient = remoteDataClientFactory.getRemoteDataClient(taskChild.getTenantId(), taskChild.getUsername(),
                sourceSystem, taskChild.getUsername());
        }
        try {
            destSystem = systemsCache.getSystem(taskChild.getTenantId(), destURL.getSystemId(), taskChild.getUsername());
            if (destSystem.getEnabled() == null || !destSystem.getEnabled()) {
                String msg = Utils.getMsg("FILES_TXFR_SYS_NOTENABLED", taskChild.getTenantId(),
                    taskChild.getUsername(), taskChild.getId(), taskChild.getUuid(), destSystem.getId());
                throw new ServiceException(msg);
            }
            destClient = remoteDataClientFactory.getRemoteDataClient(taskChild.getTenantId(), taskChild.getUsername(),
                destSystem, taskChild.getUsername());
        } catch (IOException | ServiceException ex) {
            throw new ServiceException(Utils.getMsg("FILES_TXFR_SVC_ERR1", taskChild.getTenantId(), taskChild.getUsername(),
                "chevronTwoB", taskChild.getId(), taskChild.getUuid(), ex.getMessage()), ex);
        }
        if (taskChild.isDir()) {
            destClient.mkdir(destURL.getPath());
            return taskChild;
        }
        try (InputStream sourceStream = sourceClient.getStream(sourcePath);
             ObservableInputStream observableInputStream = new ObservableInputStream(sourceStream)
        ) {
            final TransferTaskChild finalTaskChild = taskChild;
            observableInputStream.getEventStream()
                .window(Duration.ofMillis(100))
                .flatMap(window -> window.takeLast(1))
                .flatMap((progress) -> this.updateProgress(progress, finalTaskChild))
                .subscribe();
            destClient.insert(destURL.getPath(), observableInputStream);
        }
        if (sourceSystem != null && Objects.equals(sourceSystem.getSystemType(), SystemTypeEnum.LINUX) && Objects.equals(destSystem.getSystemType(), SystemTypeEnum.LINUX)) {
            List<FileInfo> itemListing = sourceClient.ls(sourcePath);
            FileInfo item = itemListing.get(0);
            if (!item.isDir() && item.getNativePermissions().contains("x")) {
                try {
                    fileUtilsService.linuxOp(destClient, destURL.getPath(), FileUtilsService.NativeLinuxOperation.CHMOD, "700", false);
                } catch (TapisException ex) {
                    String msg = Utils.getMsg("FILES_TXFR_SVC_ERR1", taskChild.getTenantId(), taskChild.getUsername(),
                        "chmod", taskChild.getId(), taskChild.getUuid(), ex.getMessage());
                    log.error(msg);
                    throw new ServiceException(msg, ex);
                }
            }
        }
        try {
            taskChild = dao.getTransferTaskChild(taskChild.getUuid());
            return taskChild;
        } catch (DAOException ex) {
            return null;
        }
    }
private TransferTaskChild chevronTwo ( TransferTaskChild taskChild ) throws ServiceException , NotFoundException , IOException { if ( taskChild . isTerminal ( ) ) return taskChild ; TapisSystem sourceSystem = null ; TapisSystem destSystem = null ; IRemoteDataClient sourceClient ; IRemoteDataClient destClient ; log . info ( "***** DOING chevronTwo **** {}" , taskChild ) ; try { taskChild . setStatus ( TransferTaskStatus . IN_PROGRESS ) ; taskChild . setRetries ( taskChild . getRetries ( ) + 1 ) ; taskChild = dao . updateTransferTaskChild ( taskChild ) ; } catch ( DAOException ex ) { throw new ServiceException ( Utils . getMsg ( "FILES_TXFR_SVC_ERR1" , taskChild . getTenantId ( ) , taskChild . getUsername ( ) , "chevronTwoA" , taskChild . getId ( ) , taskChild . getUuid ( ) , ex . getMessage ( ) ) , ex ) ; } String sourcePath ; TransferURI destURL = taskChild . getDestinationURI ( ) ; TransferURI sourceURL = taskChild . getSourceURI ( ) ; if ( taskChild . getSourceURI ( ) . toString ( ) . startsWith ( "https://" ) || taskChild . getSourceURI ( ) . toString ( ) . startsWith ( "http://" ) ) { sourceClient = new HTTPClient ( taskChild . getTenantId ( ) , taskChild . getUsername ( ) , sourceURL . toString ( ) , destURL . toString ( ) ) ; sourcePath = sourceURL . toString ( ) ; } else { sourcePath = sourceURL . getPath ( ) ; sourceSystem = systemsCache . getSystem ( taskChild . getTenantId ( ) , sourceURL . getSystemId ( ) , taskChild . getUsername ( ) ) ; if ( sourceSystem . getEnabled ( ) == null || ! sourceSystem . getEnabled ( ) ) { String msg = Utils . getMsg ( "FILES_TXFR_SYS_NOTENABLED" , taskChild . getTenantId ( ) , taskChild . getUsername ( ) , taskChild . getId ( ) , taskChild . getUuid ( ) , sourceSystem . getId ( ) ) ; throw new ServiceException ( msg ) ; } sourceClient = remoteDataClientFactory . getRemoteDataClient ( taskChild . getTenantId ( ) , taskChild . getUsername ( ) , sourceSystem , taskChild . getUsername ( ) ) ; } try { destSystem = systemsCache . getSystem ( taskChild . getTenantId ( ) , destURL . getSystemId ( ) , taskChild . getUsername ( ) ) ; if ( destSystem . getEnabled ( ) == null || ! destSystem . getEnabled ( ) ) { String msg = Utils . getMsg ( "FILES_TXFR_SYS_NOTENABLED" , taskChild . getTenantId ( ) , taskChild . getUsername ( ) , taskChild . getId ( ) , taskChild . getUuid ( ) , destSystem . getId ( ) ) ; throw new ServiceException ( msg ) ; } destClient = remoteDataClientFactory . getRemoteDataClient ( taskChild . getTenantId ( ) , taskChild . getUsername ( ) , destSystem , taskChild . getUsername ( ) ) ; } catch ( IOException | ServiceException ex ) { throw new ServiceException ( Utils . getMsg ( "FILES_TXFR_SVC_ERR1" , taskChild . getTenantId ( ) , taskChild . getUsername ( ) , "chevronTwoB" , taskChild . getId ( ) , taskChild . getUuid ( ) , ex . getMessage ( ) ) , ex ) ; } if ( taskChild . isDir ( ) ) { destClient . mkdir ( destURL . getPath ( ) ) ; return taskChild ; } try ( InputStream sourceStream = sourceClient . getStream ( sourcePath ) ; ObservableInputStream observableInputStream = new ObservableInputStream ( sourceStream ) ) { final TransferTaskChild finalTaskChild = taskChild ; observableInputStream . getEventStream ( ) . window ( Duration . ofMillis ( 100 ) ) . flatMap ( window -> window . takeLast ( 1 ) ) . flatMap ( ( progress ) -> this . updateProgress ( progress , finalTaskChild ) ) . subscribe ( ) ; destClient . insert ( destURL . getPath ( ) , observableInputStream ) ; } if ( sourceSystem != null && Objects . equals ( sourceSystem . getSystemType ( ) , SystemTypeEnum . LINUX ) && Objects . equals ( destSystem . getSystemType ( ) , SystemTypeEnum . LINUX ) ) { List < FileInfo > itemListing = sourceClient . ls ( sourcePath ) ; FileInfo item = itemListing . get ( 0 ) ; if ( ! item . isDir ( ) && item . getNativePermissions ( ) . contains ( "x" ) ) { try { fileUtilsService . linuxOp ( destClient , destURL . getPath ( ) , FileUtilsService . NativeLinuxOperation . CHMOD , "700" , false ) ; } catch ( TapisException ex ) { String msg = Utils . getMsg ( "FILES_TXFR_SVC_ERR1" , taskChild . getTenantId ( ) , taskChild . getUsername ( ) , "chmod" , taskChild . getId ( ) , taskChild . getUuid ( ) , ex . getMessage ( ) ) ; log . error ( msg ) ; throw new ServiceException ( msg , ex ) ; } } } try { taskChild = dao . getTransferTaskChild ( taskChild . getUuid ( ) ) ; return taskChild ; } catch ( DAOException ex ) { return null ; } }
The meat of the operation.
The meat of the operation .
//If we are cancelled/failed we can skip the transfer //Step 1: Update task in DB to IN_PROGRESS and increment the retries on this particular task //This should be the full string URL such as http://google.com //Step 2: Get clients for source / dest // If we have a directory to create, just do that and return the task child //Step 4: Stream the file contents to dest. While the InputStream is open, // we put a tap on it and send events that get grouped into 100 ms intervals. Progress // on the child tasks are updated during the reading of the source input stream. // Observe the progress event stream, just get the last event from // the past 1 second. //If its an executable file on a posix system, chmod it to be +x. For HTTP inputs, there is no sourceSystem, so we have to check that. //The ChildTransferTask gets updated in another thread so we look it up again here //before passing it on