bpel-runtime/src/main/java/org/apache/ode/bpel/memdao/ProcessDaoImpl.java - ode - Git at Google

 /*
  * 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.ode.bpel.memdao;

 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;

 import javax.xml.namespace.QName;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.ode.bpel.common.CorrelationKey;
 import org.apache.ode.bpel.common.ProcessState;
 import org.apache.ode.bpel.dao.CorrelationSetDAO;
 import org.apache.ode.bpel.dao.CorrelatorDAO;
 import org.apache.ode.bpel.dao.PartnerLinkDAO;
 import org.apache.ode.bpel.dao.ProcessDAO;
 import org.apache.ode.bpel.dao.ProcessInstanceDAO;

 /**
  * A very simple, in-memory implementation of the {@link ProcessDAO} interface.
  */
 class ProcessDaoImpl extends DaoBaseImpl implements ProcessDAO {
     private static final Logger __log = LoggerFactory.getLogger(ProcessDaoImpl.class);

     private QName _processId;
     private QName _type;
     private long _version;
     final Map<String, CorrelatorDaoImpl> _correlators = new ConcurrentHashMap<String, CorrelatorDaoImpl>();
     protected final Map<Long, ProcessInstanceDAO> _instances = new ConcurrentHashMap<Long, ProcessInstanceDAO>();
     protected final Map<Long, Long> _instancesAge = new ConcurrentHashMap<Long, Long>();
     protected final Map<Integer, PartnerLinkDAO> _plinks = new ConcurrentHashMap<Integer, PartnerLinkDAO>();
     private Map<QName, ProcessDaoImpl> _store;
     private BpelDAOConnectionImpl _conn;
     private int _executionCount = 0;
     private Collection<Long> _instancesToRemove = new ConcurrentLinkedQueue<Long>();
     private volatile long _lastRemoval = 0;

     private String _guid;

     public ProcessDaoImpl(BpelDAOConnectionImpl conn, Map<QName, ProcessDaoImpl> store,
                           QName processId, QName type, String guid, long version) {
         if (__log.isDebugEnabled()) {
             __log.debug("Creating ProcessDao object for process \"" + processId + "\".");
         }

         _guid = guid;
         _conn = conn;
         _store = store;
         _processId = processId;
         _type = type;
         _version = version;
     }

     public Serializable getId() {
         return _guid;
     }

     public QName getProcessId() {
         return _processId;
     }

     public CorrelatorDAO getCorrelator(String cid) {
         CorrelatorDAO ret = _correlators.get(cid);
         if (ret == null) {
             throw new IllegalArgumentException("no such correlator: " + cid);
         }
         return ret;
     }

     public Collection<CorrelatorDAO> getCorrelators() {
         // Note: _correlators.values() is a Collection<CorrealatorDaoImpl>. We can't just return this object
         // since Collection<CorrelatorDAO> is /not/ assignment compatible with Collection<CorrelatorDaoImpl>.
         // However, a immutable Collection<CorrelationDAO> is assignment compatible with Collection<CorrelatorDaoImpl>,
         // but.... we need to introduce some ambiguity into the type hierarchy so that Java will infer the correct type.

         // Make an ambiguous collection.
         Collection<? extends CorrelatorDAO> foo =  _correlators.values();

         // In order to get a collection of the super-type from a sub-type we must make the collection read-only.
         return Collections.unmodifiableCollection(foo);
     }

     public void removeRoutes(String routeId, ProcessInstanceDAO target) {
         for (CorrelatorDAO correlatorDAO : _correlators.values()) {
             correlatorDAO.removeRoutes(routeId, target);
         }
     }

     public ProcessInstanceDAO createInstance(CorrelatorDAO correlator) {
         final ProcessInstanceDaoImpl newInstance = new ProcessInstanceDaoImpl(_conn, this, correlator);
         _conn.defer(new Runnable() {
             public void run() {
                 _instances.put(newInstance.getInstanceId(), newInstance);
                 _instancesAge.put(newInstance.getInstanceId(), System.currentTimeMillis());
             }
         });

         discardOldInstances();

         // Removing right away on rollback
         final Long iid = newInstance.getInstanceId();
         _conn.onRollback(new Runnable() {
             public void run() {
                 _instances.remove(iid);
                 _instancesAge.remove(iid);
             }
         });

         _executionCount++;
         return newInstance;
     }

     public ProcessInstanceDAO getInstance(Long instanceId) {
         return _instances.get(instanceId);
     }

     public Collection<ProcessInstanceDAO> findInstance(CorrelationKey key) {
         ArrayList<ProcessInstanceDAO> result = new ArrayList<ProcessInstanceDAO>();
         for (ProcessInstanceDAO instance : _instances.values()) {
             for (CorrelationSetDAO corrSet : instance.getCorrelationSets()) {
                 if (corrSet.getValue().equals(key)) result.add(instance);
             }
         }
         return result;
     }

     public void instanceCompleted(ProcessInstanceDAO instance) {
         // Cleaning up
         if (__log.isDebugEnabled())
           __log.debug("Removing completed process instance " + instance.getInstanceId() + " from in-memory store.");
         _instancesAge.remove(instance.getInstanceId());
         ProcessInstanceDAO removed = _instances.remove(instance.getInstanceId());
         if (removed == null) {
             // Checking for leftover instances that should be removed
             ArrayList<Long> removals = new ArrayList<Long>(_instancesToRemove);
             for (Long iid : removals) {
                 _instances.remove(iid);
             }
             _instancesToRemove.removeAll(removals);

             // The instance can't be found probably because the transaction isn't committed yet and
             // it doesn't exist. Saving its id for later cleanup.
             _instancesToRemove.add(instance.getInstanceId());
         }
     }

     public void deleteProcessAndRoutes() {
         _store.remove(_processId);
     }

     public long getVersion() {
         return _version;
     }

     public String getDeployer() {
         return "nobody";
     }

     public QName getType() {
         return _type;
     }

     public CorrelatorDAO addCorrelator(String correlator) {
         CorrelatorDaoImpl corr = new CorrelatorDaoImpl(correlator, _conn);
         _correlators.put(corr.getCorrelatorId(), corr);
         return corr;
     }

     /**
      * Nothing to do.
      */
     public void update() {
         //TODO Check requirement for persisting.
     }

     public int getNumInstances() {
         // Instances are removed after execution, using a counter instead
         return _executionCount;
     }

     public ProcessInstanceDAO getInstanceWithLock(Long iid) {
         return getInstance(iid);
     }

     public int getActivityFailureCount() {
         return 0;
     }

     public Date getActivityFailureDateTime() {
         return null;
     }

     public String getGuid() {
         return _guid;
     }

     public void setGuid(String guid) {
         _guid = guid;
     }

     public Collection<ProcessInstanceDAO> getActiveInstances() {
         ArrayList<ProcessInstanceDAO> pis = new ArrayList<ProcessInstanceDAO>();
         for (ProcessInstanceDAO processInstanceDAO : _instances.values()) {
             if (processInstanceDAO.getState() == ProcessState.STATE_ACTIVE)
                 pis.add(processInstanceDAO);
         }
         return pis;
     }

     /**
      * Discard in-memory instances that exceeded their time-to-live to prevent memory leaks
      */
     void discardOldInstances() {
         long now = System.currentTimeMillis();
         if (now > _lastRemoval + (_conn._mexTtl/10)) {
             _lastRemoval = now;
             Object[] oldInstances = _instancesAge.keySet().toArray();
             for (int i=oldInstances.length-1; i>=0; i--) {
                 Long id = (Long) oldInstances[i];
                 Long age = _instancesAge.get(id);
                 if (age != null && now-age > _conn._mexTtl) {
                     if (_instances.get(id) != null) {
                         __log.warn("Discarding in-memory instance "+id+" because it exceeded its time-to-live: "+_instances.get(id));
                     }
                     _instances.remove(id);
                     _instancesAge.remove(id);
                 }
             }
         }
     }
 }
	/*
	* 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.ode.bpel.memdao;

	import java.io.Serializable;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.Date;
	import java.util.Map;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.ConcurrentLinkedQueue;

	import javax.xml.namespace.QName;

	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.apache.ode.bpel.common.CorrelationKey;
	import org.apache.ode.bpel.common.ProcessState;
	import org.apache.ode.bpel.dao.CorrelationSetDAO;
	import org.apache.ode.bpel.dao.CorrelatorDAO;
	import org.apache.ode.bpel.dao.PartnerLinkDAO;
	import org.apache.ode.bpel.dao.ProcessDAO;
	import org.apache.ode.bpel.dao.ProcessInstanceDAO;

	/**
	* A very simple, in-memory implementation of the {@link ProcessDAO} interface.
	*/
	class ProcessDaoImpl extends DaoBaseImpl implements ProcessDAO {
	private static final Logger __log = LoggerFactory.getLogger(ProcessDaoImpl.class);

	private QName _processId;
	private QName _type;
	private long _version;
	final Map<String, CorrelatorDaoImpl> _correlators = new ConcurrentHashMap<String, CorrelatorDaoImpl>();
	protected final Map<Long, ProcessInstanceDAO> _instances = new ConcurrentHashMap<Long, ProcessInstanceDAO>();
	protected final Map<Long, Long> _instancesAge = new ConcurrentHashMap<Long, Long>();
	protected final Map<Integer, PartnerLinkDAO> _plinks = new ConcurrentHashMap<Integer, PartnerLinkDAO>();
	private Map<QName, ProcessDaoImpl> _store;
	private BpelDAOConnectionImpl _conn;
	private int _executionCount = 0;
	private Collection<Long> _instancesToRemove = new ConcurrentLinkedQueue<Long>();
	private volatile long _lastRemoval = 0;

	private String _guid;

	public ProcessDaoImpl(BpelDAOConnectionImpl conn, Map<QName, ProcessDaoImpl> store,
	QName processId, QName type, String guid, long version) {
	if (__log.isDebugEnabled()) {
	__log.debug("Creating ProcessDao object for process \"" + processId + "\".");
	}

	_guid = guid;
	_conn = conn;
	_store = store;
	_processId = processId;
	_type = type;
	_version = version;
	}

	public Serializable getId() {
	return _guid;
	}

	public QName getProcessId() {
	return _processId;
	}

	public CorrelatorDAO getCorrelator(String cid) {
	CorrelatorDAO ret = _correlators.get(cid);
	if (ret == null) {
	throw new IllegalArgumentException("no such correlator: " + cid);
	}
	return ret;
	}

	public Collection<CorrelatorDAO> getCorrelators() {
	// Note: _correlators.values() is a Collection<CorrealatorDaoImpl>. We can't just return this object
	// since Collection<CorrelatorDAO> is /not/ assignment compatible with Collection<CorrelatorDaoImpl>.
	// However, a immutable Collection<CorrelationDAO> is assignment compatible with Collection<CorrelatorDaoImpl>,
	// but.... we need to introduce some ambiguity into the type hierarchy so that Java will infer the correct type.

	// Make an ambiguous collection.
	Collection<? extends CorrelatorDAO> foo = _correlators.values();

	// In order to get a collection of the super-type from a sub-type we must make the collection read-only.
	return Collections.unmodifiableCollection(foo);
	}

	public void removeRoutes(String routeId, ProcessInstanceDAO target) {
	for (CorrelatorDAO correlatorDAO : _correlators.values()) {
	correlatorDAO.removeRoutes(routeId, target);
	}
	}

	public ProcessInstanceDAO createInstance(CorrelatorDAO correlator) {
	final ProcessInstanceDaoImpl newInstance = new ProcessInstanceDaoImpl(_conn, this, correlator);
	_conn.defer(new Runnable() {
	public void run() {
	_instances.put(newInstance.getInstanceId(), newInstance);
	_instancesAge.put(newInstance.getInstanceId(), System.currentTimeMillis());
	}
	});

	discardOldInstances();

	// Removing right away on rollback
	final Long iid = newInstance.getInstanceId();
	_conn.onRollback(new Runnable() {
	public void run() {
	_instances.remove(iid);
	_instancesAge.remove(iid);
	}
	});

	_executionCount++;
	return newInstance;
	}

	public ProcessInstanceDAO getInstance(Long instanceId) {
	return _instances.get(instanceId);
	}

	public Collection<ProcessInstanceDAO> findInstance(CorrelationKey key) {
	ArrayList<ProcessInstanceDAO> result = new ArrayList<ProcessInstanceDAO>();
	for (ProcessInstanceDAO instance : _instances.values()) {
	for (CorrelationSetDAO corrSet : instance.getCorrelationSets()) {
	if (corrSet.getValue().equals(key)) result.add(instance);
	}
	}
	return result;
	}

	public void instanceCompleted(ProcessInstanceDAO instance) {
	// Cleaning up
	if (__log.isDebugEnabled())
	__log.debug("Removing completed process instance " + instance.getInstanceId() + " from in-memory store.");
	_instancesAge.remove(instance.getInstanceId());
	ProcessInstanceDAO removed = _instances.remove(instance.getInstanceId());
	if (removed == null) {
	// Checking for leftover instances that should be removed
	ArrayList<Long> removals = new ArrayList<Long>(_instancesToRemove);
	for (Long iid : removals) {
	_instances.remove(iid);
	}
	_instancesToRemove.removeAll(removals);

	// The instance can't be found probably because the transaction isn't committed yet and
	// it doesn't exist. Saving its id for later cleanup.
	_instancesToRemove.add(instance.getInstanceId());
	}
	}

	public void deleteProcessAndRoutes() {
	_store.remove(_processId);
	}

	public long getVersion() {
	return _version;
	}

	public String getDeployer() {
	return "nobody";
	}

	public QName getType() {
	return _type;
	}

	public CorrelatorDAO addCorrelator(String correlator) {
	CorrelatorDaoImpl corr = new CorrelatorDaoImpl(correlator, _conn);
	_correlators.put(corr.getCorrelatorId(), corr);
	return corr;
	}

	/**
	* Nothing to do.
	*/
	public void update() {
	//TODO Check requirement for persisting.
	}

	public int getNumInstances() {
	// Instances are removed after execution, using a counter instead
	return _executionCount;
	}

	public ProcessInstanceDAO getInstanceWithLock(Long iid) {
	return getInstance(iid);
	}

	public int getActivityFailureCount() {
	return 0;
	}

	public Date getActivityFailureDateTime() {
	return null;
	}

	public String getGuid() {
	return _guid;
	}

	public void setGuid(String guid) {
	_guid = guid;
	}

	public Collection<ProcessInstanceDAO> getActiveInstances() {
	ArrayList<ProcessInstanceDAO> pis = new ArrayList<ProcessInstanceDAO>();
	for (ProcessInstanceDAO processInstanceDAO : _instances.values()) {
	if (processInstanceDAO.getState() == ProcessState.STATE_ACTIVE)
	pis.add(processInstanceDAO);
	}
	return pis;
	}

	/**
	* Discard in-memory instances that exceeded their time-to-live to prevent memory leaks
	*/
	void discardOldInstances() {
	long now = System.currentTimeMillis();
	if (now > _lastRemoval + (_conn._mexTtl/10)) {
	_lastRemoval = now;
	Object[] oldInstances = _instancesAge.keySet().toArray();
	for (int i=oldInstances.length-1; i>=0; i--) {
	Long id = (Long) oldInstances[i];
	Long age = _instancesAge.get(id);
	if (age != null && now-age > _conn._mexTtl) {
	if (_instances.get(id) != null) {
	__log.warn("Discarding in-memory instance "+id+" because it exceeded its time-to-live: "+_instances.get(id));
	}
	_instances.remove(id);
	_instancesAge.remove(id);
	}
	}
	}
	}
	}