JobService.java

package it.inaf.oats.vospace;

import it.inaf.oats.vospace.exception.InternalFaultException;
import it.inaf.oats.vospace.persistence.JobDAO;
import net.ivoa.xml.uws.v1.ExecutionPhase;
import net.ivoa.xml.uws.v1.JobSummary;
import net.ivoa.xml.vospace.v2.Protocol;
import net.ivoa.xml.vospace.v2.Transfer;
import net.ivoa.xml.uws.v1.ErrorSummaryFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import it.inaf.oats.vospace.exception.VoSpaceErrorSummarizableException;

@Service
public class JobService {

    @Autowired
    private JobDAO jobDAO;

    @Autowired
    private UriService uriService;

    @Autowired
    private AsyncTransferService asyncTransfService;

    public enum JobType {
        pullToVoSpace,
        pullFromVoSpace,
        pushToVoSpace,
        moveNode,
        copyNode
    }

    public void setJobPhase(JobSummary job, String phase) {

        // TODO: check allowed job phase transitions
        switch (phase) {
            case "RUN":
                startJob(job);
                break;
            case "ABORT":
                throw new UnsupportedOperationException("Not implemented yet");
            default:
                throw new IllegalArgumentException("Invalid phase parameter: " + phase);
        }
    }

    private void startJob(JobSummary job) {

        Transfer transfer = uriService.getTransfer(job);
        
        switch (getJobType(transfer)) {
            case pullToVoSpace:
                handlePullToVoSpace(job, transfer);
                break;
            case pullFromVoSpace:
            case pushToVoSpace:
                handleVoSpaceUrlsListResult(job, transfer);
                break;
            default:
                throw new UnsupportedOperationException("Not implemented yet");
        }
    }

    private void handlePullToVoSpace(JobSummary job, Transfer transfer) {
        try {
            for (Protocol protocol : transfer.getProtocols()) {
                switch (protocol.getUri()) {
                    case "ia2:async-recall":
                        asyncTransfService.startJob(job);
                        // ASK IF IT's OK neglect phase update.
                        return;
                    case "ivo://ivoa.net/vospace/core#httpget":
                        String nodeUri = transfer.getTarget();
                        String contentUri = protocol.getEndpoint();
                        uriService.setNodeRemoteLocation(nodeUri, contentUri);
                        uriService.setTransferJobResult(job, transfer);
                        job.setPhase(ExecutionPhase.COMPLETED);
                        jobDAO.updateJob(job);
                        return;
                    default:
                        throw new InternalFaultException("Unsupported pullToVoSpace protocol: " + protocol.getUri());
                }
            }
        } catch (VoSpaceErrorSummarizableException e) {
            job.setPhase(ExecutionPhase.ERROR);
            job.setErrorSummary(ErrorSummaryFactory.newErrorSummary(e.getFault()));
            jobDAO.updateJob(job);
        }
    }

    private void handleVoSpaceUrlsListResult(JobSummary job, Transfer transfer) {
        try {
            job.setPhase(ExecutionPhase.EXECUTING);
            uriService.setTransferJobResult(job, transfer);
            job.setPhase(ExecutionPhase.COMPLETED);
            // Need to catch other exceptions too to avoid inconsistent job status
        } catch (VoSpaceErrorSummarizableException e) {
            job.setPhase(ExecutionPhase.ERROR);
            job.setErrorSummary(ErrorSummaryFactory.newErrorSummary(e.getFault()));
        } finally {
            jobDAO.updateJob(job);
        }
    }

    private JobType getJobType(Transfer transfer) {
        return JobType.valueOf(transfer.getDirection());
    }

    /**
     * Synchronous transfer endpoint creates a job that is immediately set to
     * COMPLETED or to ERROR in case some fault occurred.
     *
     * In case of ERROR, Protocols are stripped from job representation in
     * compliance with specifications
     *
     */
    public void createSyncJobResult(JobSummary job) {
        try {
            uriService.setSyncTransferEndpoints(job);
            job.setPhase(ExecutionPhase.COMPLETED);
            // Need to catch other exceptions too to avoid inconsistent job status
        } catch (VoSpaceErrorSummarizableException e) {
            job.setPhase(ExecutionPhase.ERROR);
            uriService.getTransfer(job).getProtocols().clear();
            job.setErrorSummary(ErrorSummaryFactory.newErrorSummary(e.getFault()));
        } finally {
            jobDAO.createJob(job);
        }
    }
}