Project

General

Profile

« Previous | Next » 

Revision 51980

added missing class

View differences:

modules/dnet-openaireplus-workflows/trunk/src/main/java/eu/dnetlib/msro/openaireplus/workflows/nodes/IncrementalTransformationJobNode.java
1
package eu.dnetlib.msro.openaireplus.workflows.nodes;
2

  
3
import java.util.Iterator;
4
import java.util.List;
5
import java.util.Map;
6

  
7
import com.googlecode.sarasvati.Arc;
8
import com.googlecode.sarasvati.NodeToken;
9
import eu.dnetlib.common.logging.DnetLogger;
10
import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
11
import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
12
import eu.dnetlib.enabling.locators.UniqueServiceLocator;
13
import eu.dnetlib.miscutils.datetime.DateUtils;
14
import eu.dnetlib.msro.rmi.MSROException;
15
import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;
16
import eu.dnetlib.msro.workflows.util.WorkflowsConstants;
17
import org.apache.commons.lang.math.NumberUtils;
18
import org.apache.commons.logging.Log;
19
import org.apache.commons.logging.LogFactory;
20
import org.springframework.beans.factory.annotation.Autowired;
21

  
22
public class IncrementalTransformationJobNode extends SimpleJobNode {
23

  
24
    private static final Log log = LogFactory.getLog(IncrementalTransformationJobNode.class);
25

  
26
    private String transformationType;
27

  
28
    @Autowired
29
    private DnetLogger dnetLogger;
30

  
31
    @Autowired
32
    private UniqueServiceLocator locator;
33

  
34
    private static final DateUtils dateUtils = new DateUtils();
35

  
36
    @Override
37
    protected String execute(NodeToken nodeToken) throws Exception {
38

  
39
        if ("incremental".equalsIgnoreCase(transformationType)) {
40
            final String currentWfProfileId = findCurrentWfProfileId(nodeToken);
41
            final Long lastSuccessEndDate = findLastSuccessEndDate(currentWfProfileId);
42
            if (lastSuccessEndDate<0) {
43
                nodeToken.getFullEnv().setAttribute("TransformationTypeInfo", "Last success date < 0, transformation forced to REFRESH");
44
                nodeToken.getFullEnv().setAttribute("transformationType", "REFRESH");
45
                return Arc.DEFAULT_ARC;
46
            }
47
            log.info("Last success date "+ lastSuccessEndDate);
48
            String trRuleId = getTransformationId(currentWfProfileId);
49
            final long lastUpdateDate = getLastTransformationRuleUpdate(trRuleId);
50
            log.info(String.format("Last update date of the transformation rule with id %s is %s", trRuleId, DateUtils.calculate_ISO8601(lastUpdateDate)));
51

  
52
            if (lastUpdateDate > lastSuccessEndDate) {
53
                nodeToken.getFullEnv().setAttribute("TransformationTypeInfo", "Transformation Rule has been updated, transformation forced to REFRESH");
54
                nodeToken.getFullEnv().setAttribute("transformationType", "REFRESH");
55
                return Arc.DEFAULT_ARC;
56
            }
57
            nodeToken.getFullEnv().setAttribute("TransformationTypeInfo", "Transformation type set to INCREMENTAL with date "+DateUtils.calculate_ISO8601(lastSuccessEndDate));
58
            nodeToken.getFullEnv().setAttribute("transformationType", "INCREMENTAL");
59
            nodeToken.getFullEnv().setAttribute("DateFromFilter", lastSuccessEndDate);            
60
            return Arc.DEFAULT_ARC;
61
        }
62
        nodeToken.getFullEnv().setAttribute("transformationType", "REFRESH");
63
        nodeToken.getFullEnv().setAttribute("TransformationTypeInfo", "Transformation type manually set to REFRESH");
64
        return Arc.DEFAULT_ARC;
65
    }
66

  
67
    private Long findLastSuccessEndDate(String profId) {
68
        long res = -1;
69

  
70
        final Iterator<Map<String, String>> iter = dnetLogger.find(WorkflowsConstants.SYSTEM_WF_PROFILE_ID, profId);
71
        while (iter.hasNext()) {
72
            final Map<String, String> map = iter.next();
73
            log.debug("Iterating on the logs");
74
            if ("true".equalsIgnoreCase(map.get(WorkflowsConstants.SYSTEM_COMPLETED_SUCCESSFULLY))) {
75
                final long curr = NumberUtils.toLong(map.get(WorkflowsConstants.SYSTEM_END_DATE), -1);
76
                if (curr > res) {
77
                    res = curr;
78
                }
79
            }
80
        }
81
        return res;
82
    }
83

  
84
    private String findCurrentWfProfileId(NodeToken token) throws MSROException {
85
        log.debug("Start to find the current profile Id");
86
        final String p1 = token.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);
87
        if (p1 != null && !p1.isEmpty()) {
88
            log.debug("The profile Id found is "+p1);
89
            return p1;
90
        }
91
        final String p2 = token.getFullEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);
92
        if (p2 != null && !p2.isEmpty()) {
93
            log.debug("The profile Id found is "+p2);
94
            return p2;
95
        }
96
        final String p3 = token.getProcess().getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID);
97
        if (p3 != null && !p3.isEmpty()) {
98
            log.debug("The profile Id found is "+p3);
99
            return p3;
100
        }
101
        throw new MSROException("Missing property in env: " + WorkflowsConstants.SYSTEM_WF_PROFILE_ID);
102
    }
103

  
104

  
105
    private String getTransformationId(final String workflowId) throws ISLookUpException {
106

  
107
        final String query="for $x in collection('/db/DRIVER/WorkflowDSResources/WorkflowDSResourceType') where $x//RESOURCE_IDENTIFIER/@value='%s' " +
108
                "return $x//PARAM[./@category='TRANSFORMATION_RULE_ID']/text()";
109
        final ISLookUpService lookUpService = locator.getService(ISLookUpService.class);
110
        final String queryInstance = String.format(query, workflowId);
111
        log.debug("Query to find the Transformation Rule");
112
        List<String> transformationId = lookUpService.quickSearchProfile(queryInstance);
113
        if(transformationId== null || transformationId.isEmpty())
114
            throw new RuntimeException("Error unable to find the Transformation rule ID on workflow profile "+workflowId);
115
        return transformationId.get(0);
116
    }
117

  
118

  
119
    private Long getLastTransformationRuleUpdate(final String trId) throws ISLookUpException {
120
        final String query = "for $x in collection('/db/DRIVER/TransformationRuleDSResources/TransformationRuleDSResourceType') where $x//RESOURCE_IDENTIFIER/@value='%s' return $x//DATE_OF_CREATION/@value/string()";
121
        log.debug("retrieve creation date from transformation ID "+trId);
122
        final ISLookUpService lookUpService = locator.getService(ISLookUpService.class);
123
        final String queryInstance = String.format(query, trId);
124
        log.debug("Query to find the Transformation Rule");
125
        List<String> currentDate = lookUpService.quickSearchProfile(queryInstance);
126
        if(currentDate== null || currentDate.isEmpty())
127
            throw new RuntimeException("Error unable to find the creation date of the  Transformation rule "+trId);
128
        return dateUtils.parse(currentDate.get(0)).getTime();
129
    }
130

  
131

  
132

  
133
    public void setTransformationType(String transformationType) {
134
        this.transformationType = transformationType;
135
    }
136

  
137
    public String getTransformationType() {
138
        return transformationType;
139
    }
140
}

Also available in: Unified diff