| /* |
| * 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.oodt.cas.workflow.instrepo; |
| |
| //OODT imports |
| import org.apache.oodt.cas.metadata.Metadata; |
| import org.apache.oodt.cas.workflow.instrepo.WorkflowInstanceRepository; |
| import org.apache.oodt.cas.workflow.structs.WorkflowInstance; |
| import org.apache.oodt.cas.workflow.structs.WorkflowTask; |
| import org.apache.oodt.cas.workflow.structs.exceptions.InstanceRepositoryException; |
| import org.apache.oodt.cas.workflow.util.DbStructFactory; |
| |
| //JDK imports |
| import java.net.URLDecoder; |
| import java.net.URLEncoder; |
| import java.sql.Connection; |
| import java.sql.ResultSet; |
| import java.sql.SQLException; |
| import java.sql.Statement; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Vector; |
| import java.util.logging.Level; |
| import java.util.logging.Logger; |
| import javax.sql.DataSource; |
| |
| /** |
| * @author mattmann |
| * @version $Revision$ |
| * |
| * <p> |
| * A {@link WorkflowInstanceRepository} that persists {@link WorkflowInstance}s |
| * to a JDBC-accessible DBMS. |
| * </p>. |
| */ |
| public class DataSourceWorkflowInstanceRepository extends |
| AbstractPaginatibleInstanceRepository { |
| |
| /* our data source */ |
| private DataSource dataSource = null; |
| |
| /* our log stream */ |
| private static final Logger LOG = Logger |
| .getLogger(DataSourceWorkflowInstanceRepository.class.getName()); |
| |
| /* should we quote fields or not */ |
| private boolean quoteFields = false; |
| |
| public DataSourceWorkflowInstanceRepository(DataSource ds, |
| boolean quoteFields, int pageSize) { |
| this.dataSource = ds; |
| this.quoteFields = quoteFields; |
| this.pageSize = pageSize; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.oodt.cas.workflow.engine.WorkflowInstanceRepository#addWorkflowInstance(org.apache.oodt.cas.workflow.structs.WorkflowInstance) |
| */ |
| public synchronized void addWorkflowInstance(WorkflowInstance wInst) |
| throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| ResultSet rs = null; |
| |
| try { |
| conn = dataSource.getConnection(); |
| conn.setAutoCommit(false); |
| statement = conn.createStatement(); |
| |
| String startWorkflowSql = null; |
| String taskIdField = null; |
| String workflowIdField = null; |
| |
| if (quoteFields) { |
| taskIdField = "'" |
| + ((WorkflowTask) wInst.getWorkflow().getTasks().get(0)) |
| .getTaskId() + "'"; |
| workflowIdField = "'" + wInst.getWorkflow().getId() + "'"; |
| } else { |
| taskIdField = ((WorkflowTask) wInst.getWorkflow().getTasks() |
| .get(0)).getTaskId(); |
| workflowIdField = wInst.getWorkflow().getId(); |
| } |
| |
| startWorkflowSql = "INSERT INTO workflow_instances " |
| + "(workflow_instance_status, workflow_id, current_task_id," |
| + "start_date_time, end_date_time, current_task_start_date_time," |
| + "current_task_end_date_time, priority, times_blocked) " + "VALUES ('" |
| + wInst.getStatus() + "', " + workflowIdField + "," |
| + taskIdField + ", '" + wInst.getStartDateTimeIsoStr() |
| + "','" + wInst.getEndDateTimeIsoStr() + "','" |
| + wInst.getCurrentTaskStartDateTimeIsoStr() + "','" |
| + wInst.getCurrentTaskEndDateTimeIsoStr() + "', "+wInst.getPriority().getValue()+", " |
| + wInst.getTimesBlocked() + ")"; |
| |
| LOG.log(Level.FINE, "sql: Executing: " + startWorkflowSql); |
| statement.execute(startWorkflowSql); |
| |
| String workflowInstId = new String(); |
| |
| synchronized (workflowInstId) { |
| String getWorkflowInstIdSql = "SELECT MAX(workflow_instance_id) " |
| + "AS max_id FROM workflow_instances"; |
| |
| rs = statement.executeQuery(getWorkflowInstIdSql); |
| |
| while (rs.next()) { |
| workflowInstId = String.valueOf(rs.getInt("max_id")); |
| } |
| } |
| |
| conn.commit(); |
| wInst.setId(workflowInstId); |
| |
| // now add its metadata |
| addWorkflowInstanceMetadata(wInst); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, "Exception starting workflow. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback startWorkflow transaction. Message: " |
| + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (rs != null) { |
| try { |
| rs.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| rs = null; |
| } |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.oodt.cas.workflow.engine.WorkflowInstanceRepository#updateWorkflowInstance(org.apache.oodt.cas.workflow.structs.WorkflowInstance) |
| */ |
| public synchronized void updateWorkflowInstance(WorkflowInstance wInst) |
| throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| String taskIdField = null, workflowIdField = null; |
| |
| try { |
| conn = dataSource.getConnection(); |
| conn.setAutoCommit(false); |
| statement = conn.createStatement(); |
| |
| if (quoteFields) { |
| taskIdField = "'" + wInst.getCurrentTaskId() + "'"; |
| workflowIdField = "'" + wInst.getWorkflow().getId() + "'"; |
| } else { |
| taskIdField = wInst.getCurrentTaskId(); |
| workflowIdField = wInst.getWorkflow().getId(); |
| } |
| |
| String updateStatusSql = "UPDATE workflow_instances SET " |
| + "workflow_instance_status='" + wInst.getStatus() |
| + "', current_task_id=" + taskIdField + ", workflow_id = " |
| + workflowIdField + ",start_date_time='" |
| + wInst.getStartDateTimeIsoStr() + "'," + "end_date_time='" |
| + wInst.getEndDateTimeIsoStr() |
| + "',current_task_start_date_time='" |
| + wInst.getCurrentTaskStartDateTimeIsoStr() |
| + "',current_task_end_date_time='" |
| + wInst.getCurrentTaskEndDateTimeIsoStr() |
| + "',priority=" |
| + wInst.getPriority().getValue() |
| + ",times_blocked=" |
| + wInst.getTimesBlocked() |
| +" WHERE workflow_instance_id = " + wInst.getId(); |
| |
| LOG.log(Level.FINE, "updateStatusSql: Executing: " |
| + updateStatusSql); |
| statement.execute(updateStatusSql); |
| conn.commit(); |
| |
| // now update its metadata |
| removeWorkflowInstanceMetadata(wInst.getId()); |
| addWorkflowInstanceMetadata(wInst); |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception updating workflow instance. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback updateWorkflowInstanceStatus " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.oodt.cas.workflow.engine.WorkflowInstanceRepository#removeWorkflowInstance(org.apache.oodt.cas.workflow.structs.WorkflowInstance) |
| */ |
| public synchronized void removeWorkflowInstance(WorkflowInstance wInst) |
| throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| |
| try { |
| conn = dataSource.getConnection(); |
| conn.setAutoCommit(false); |
| statement = conn.createStatement(); |
| |
| String deleteSql = "DELETE FROM workflow_instances " |
| + "WHERE workflow_instance_id = " + wInst.getId(); |
| |
| LOG.log(Level.FINE, "sql: Executing: " + deleteSql); |
| statement.execute(deleteSql); |
| conn.commit(); |
| |
| // now remove its metadata |
| removeWorkflowInstanceMetadata(wInst.getId()); |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception removing workflow instance. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback removeWorkflowInstance " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.oodt.cas.workflow.engine.WorkflowInstanceRepository#getWorkflowInstanceById(java.lang.String) |
| */ |
| public WorkflowInstance getWorkflowInstanceById(String workflowInstId) |
| throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| ResultSet rs = null; |
| |
| WorkflowInstance workflowInst = null; |
| |
| try { |
| conn = dataSource.getConnection(); |
| statement = conn.createStatement(); |
| |
| String getWorkflowSql = "SELECT * from workflow_instances " |
| + "WHERE workflow_instance_id = " + workflowInstId; |
| |
| LOG.log(Level.FINE, "getWorkflowInstanceById: Executing: " |
| + getWorkflowSql); |
| rs = statement.executeQuery(getWorkflowSql); |
| |
| while (rs.next()) { |
| workflowInst = DbStructFactory.getWorkflowInstance(rs); |
| // add its metadata |
| workflowInst |
| .setSharedContext(getWorkflowInstanceMetadata(workflowInst |
| .getId())); |
| } |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception getting workflow instance. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback getWorkflowInstanceById " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (rs != null) { |
| try { |
| rs.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| rs = null; |
| } |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| return workflowInst; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.oodt.cas.workflow.engine.WorkflowInstanceRepository#getWorkflowInstances() |
| */ |
| public List getWorkflowInstances() throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| ResultSet rs = null; |
| List workflowInsts = null; |
| |
| try { |
| conn = dataSource.getConnection(); |
| statement = conn.createStatement(); |
| |
| String getWorkflowSql = "SELECT * from workflow_instances " |
| + "ORDER BY workflow_instance_id DESC"; |
| |
| LOG.log(Level.FINE, "getWorkflowInstances: Executing: " |
| + getWorkflowSql); |
| rs = statement.executeQuery(getWorkflowSql); |
| |
| workflowInsts = new Vector(); |
| while (rs.next()) { |
| WorkflowInstance workflowInst = DbStructFactory |
| .getWorkflowInstance(rs); |
| // add its metadata |
| workflowInst |
| .setSharedContext(getWorkflowInstanceMetadata(workflowInst |
| .getId())); |
| workflowInsts.add(workflowInst); |
| } |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception getting workflow instance. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback getWorkflowInstances " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (rs != null) { |
| try { |
| rs.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| rs = null; |
| } |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| return workflowInsts; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.oodt.cas.workflow.engine.WorkflowInstanceRepository#getWorkflowInstancesByStatus(java.lang.String) |
| */ |
| public List getWorkflowInstancesByStatus(String status) |
| throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| ResultSet rs = null; |
| |
| List workflowInsts = null; |
| |
| try { |
| conn = dataSource.getConnection(); |
| statement = conn.createStatement(); |
| |
| String getWorkflowSql = "SELECT * from workflow_instances " |
| + "WHERE workflow_instance_status = '" + status |
| + "' ORDER BY workflow_instance_id DESC"; |
| |
| LOG.log(Level.FINE, "getWorkflowInstancesByStatus: Executing: " |
| + getWorkflowSql); |
| rs = statement.executeQuery(getWorkflowSql); |
| |
| workflowInsts = new Vector(); |
| while (rs.next()) { |
| WorkflowInstance workflowInst = DbStructFactory |
| .getWorkflowInstance(rs); |
| // add its metadata |
| workflowInst |
| .setSharedContext(getWorkflowInstanceMetadata(workflowInst |
| .getId())); |
| workflowInsts.add(workflowInst); |
| } |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception getting workflow instance. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback getWorkflowInstancesByStatus " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (rs != null) { |
| try { |
| rs.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| rs = null; |
| } |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| return workflowInsts; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.oodt.cas.workflow.instrepo.WorkflowInstanceRepository#getNumWorkflowInstances() |
| */ |
| public int getNumWorkflowInstances() throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| ResultSet rs = null; |
| int numInsts = -1; |
| |
| try { |
| conn = dataSource.getConnection(); |
| statement = conn.createStatement(); |
| |
| String getWorkflowSql = "SELECT COUNT(workflow_instance_id) AS num_insts from workflow_instances"; |
| |
| LOG.log(Level.FINE, "getNumWorkflowInstances: Executing: " |
| + getWorkflowSql); |
| rs = statement.executeQuery(getWorkflowSql); |
| |
| while (rs.next()) { |
| numInsts = rs.getInt("num_insts"); |
| } |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception getting num workflow instances. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback getNumWorkflowInstances " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (rs != null) { |
| try { |
| rs.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| rs = null; |
| } |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| return numInsts; |
| } |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see org.apache.oodt.cas.workflow.instrepo.WorkflowInstanceRepository#getNumWorkflowInstancesByStatus(java.lang.String) |
| */ |
| public int getNumWorkflowInstancesByStatus(String status) |
| throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| ResultSet rs = null; |
| int numInsts = -1; |
| |
| try { |
| conn = dataSource.getConnection(); |
| statement = conn.createStatement(); |
| |
| String getWorkflowSql = "SELECT COUNT(workflow_instance_id) AS num_insts from workflow_instances " |
| + "WHERE workflow_instance_status = '" + status + "'"; |
| |
| LOG.log(Level.FINE, "getNumWorkflowInstancesByStatus: Executing: " |
| + getWorkflowSql); |
| rs = statement.executeQuery(getWorkflowSql); |
| |
| while (rs.next()) { |
| numInsts = rs.getInt("num_insts"); |
| } |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception getting num workflow instances by status. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback getNumWorkflowInstancesByStatus " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (rs != null) { |
| try { |
| rs.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| rs = null; |
| } |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| return numInsts; |
| } |
| |
| protected List paginateWorkflows(int pageNum, String status) |
| throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| ResultSet rs = null; |
| |
| List wInstIds = null; |
| int numResults = -1; |
| |
| if (status == null || (status != null && status.equals(""))) { |
| numResults = getNumWorkflowInstances(); |
| } else { |
| numResults = getNumWorkflowInstancesByStatus(status); |
| } |
| |
| try { |
| conn = dataSource.getConnection(); |
| statement = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, |
| ResultSet.CONCUR_READ_ONLY); |
| |
| String getWorkflowSql = "SELECT workflow_instance_id FROM workflow_instances "; |
| if (status != null && !status.equals("")) { |
| getWorkflowSql += "WHERE workflow_instance_status = '" + status |
| + "'"; |
| |
| } |
| |
| getWorkflowSql += "ORDER BY workflow_instance_id DESC "; |
| |
| LOG.log(Level.FINE, "workflow instance paged query: executing: " |
| + getWorkflowSql); |
| |
| rs = statement.executeQuery(getWorkflowSql); |
| wInstIds = new Vector(); |
| |
| int startNum = (pageNum - 1) * pageSize; |
| |
| if (startNum > numResults) { |
| startNum = 0; |
| } |
| |
| // must call next first, or else no relative cursor |
| if (rs.next()) { |
| // grab the first one |
| int numGrabbed = -1; |
| if(pageNum == 1){ |
| numGrabbed = 1; |
| wInstIds.add(rs.getString("workflow_instance_id")); |
| } |
| else{ |
| numGrabbed = 0; |
| } |
| |
| if(pageNum != 1){ |
| // now move the cursor to the correct position |
| rs.relative(startNum); |
| } |
| |
| // grab the rest |
| while (rs.next() && numGrabbed < pageSize) { |
| String wInstId = rs.getString("workflow_instance_id"); |
| wInstIds.add(wInstId); |
| numGrabbed++; |
| } |
| } |
| |
| if (wInstIds.size() == 0) { |
| wInstIds = null; |
| } |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, "Exception performing query. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback query transaction. Message: " |
| + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (rs != null) { |
| try { |
| rs.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| rs = null; |
| } |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| return wInstIds; |
| } |
| |
| private Metadata getWorkflowInstanceMetadata(String workflowInstId) |
| throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| ResultSet rs = null; |
| |
| Metadata met = new Metadata(); |
| |
| try { |
| conn = dataSource.getConnection(); |
| statement = conn.createStatement(); |
| |
| String getWorkflowSql = "SELECT * from workflow_instance_metadata " |
| + "WHERE workflow_instance_id = " + workflowInstId; |
| |
| LOG.log(Level.FINE, "Executing: " + getWorkflowSql); |
| rs = statement.executeQuery(getWorkflowSql); |
| |
| while (rs.next()) { |
| met.addMetadata(rs.getString("workflow_met_key"), URLDecoder.decode(rs |
| .getString("workflow_met_val"), "UTF-8")); |
| } |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception getting workflow instance metadata. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback getWorkflowInstancesMetadata " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (rs != null) { |
| try { |
| rs.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| rs = null; |
| } |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| return met; |
| } |
| |
| private synchronized void addWorkflowInstanceMetadata(WorkflowInstance inst) |
| throws InstanceRepositoryException { |
| |
| if (inst.getSharedContext() != null |
| && inst.getSharedContext().getHashtable().keySet().size() > 0) { |
| for (Iterator i = inst.getSharedContext().getHashtable().keySet() |
| .iterator(); i.hasNext();) { |
| String key = (String) i.next(); |
| List vals = inst.getSharedContext().getAllMetadata(key); |
| if (vals != null && vals.size() > 0) { |
| for (Iterator j = vals.iterator(); j.hasNext();) { |
| String val = (String) j.next(); |
| if (val != null && !val.equals("")) { |
| addMetadataValue(inst.getId(), key, val); |
| } |
| } |
| } |
| |
| } |
| } |
| |
| } |
| |
| private synchronized void addMetadataValue(String wInstId, String key, |
| String val) throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| |
| try { |
| conn = dataSource.getConnection(); |
| conn.setAutoCommit(false); |
| statement = conn.createStatement(); |
| String addMetSql = "INSERT INTO workflow_instance_metadata" |
| + " (workflow_instance_id,workflow_met_key,workflow_met_val) VALUES (" |
| + wInstId + ",'" + key + "','" + URLEncoder.encode(val, "UTF-8") + "')"; |
| |
| LOG.log(Level.FINE, "sql: Executing: " + addMetSql); |
| statement.execute(addMetSql); |
| |
| conn.commit(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, "Exception adding metadata [" + key + "=>" |
| + val + "] to workflow inst: [" + wInstId + "]. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback addMetadataValue transaction. Message: " |
| + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| |
| } |
| |
| private synchronized void removeWorkflowInstanceMetadata( |
| String workflowInstId) throws InstanceRepositoryException { |
| Connection conn = null; |
| Statement statement = null; |
| |
| try { |
| conn = dataSource.getConnection(); |
| conn.setAutoCommit(false); |
| statement = conn.createStatement(); |
| |
| String deleteSql = "DELETE FROM workflow_instance_metadata " |
| + "WHERE workflow_instance_id = " + workflowInstId; |
| |
| LOG.log(Level.FINE, "sql: Executing: " + deleteSql); |
| statement.execute(deleteSql); |
| conn.commit(); |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| LOG.log(Level.WARNING, |
| "Exception removing workflow instance metadata. Message: " |
| + e.getMessage()); |
| try { |
| conn.rollback(); |
| } catch (SQLException e2) { |
| LOG.log(Level.SEVERE, |
| "Unable to rollback removeWorkflowInstanceMetadata " |
| + "transaction. Message: " + e2.getMessage()); |
| } |
| throw new InstanceRepositoryException(e.getMessage()); |
| } finally { |
| if (statement != null) { |
| try { |
| statement.close(); |
| } catch (SQLException ignore) { |
| } |
| |
| statement = null; |
| } |
| |
| if (conn != null) { |
| try { |
| conn.close(); |
| |
| } catch (SQLException ignore) { |
| } |
| |
| conn = null; |
| } |
| } |
| } |
| |
| } |
