[riftsaw-commits] riftsaw SVN: r1082 - in branches/ODE/RiftSaw-ODE-trunk: bpel-api/src/main/java/org/apache/ode/bpel/iapi and 7 other directories.

Author: jeff.yuchang Date: 2010-11-05 00:12:41 -0400 (Fri, 05 Nov 2010) New Revision: 1082 Added: branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/Cache.java branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/CacheProvider.java branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/ branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/CacheProviderFactory.java branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/DefaultCacheProvider.java branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/HashMapCache.java Modified: branches/ODE/RiftSaw-ODE-trunk/axis2/src/main/java/org/apache/ode/axis2/ODEServer.java branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/ProcessState.java branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/config/OdeConfigProperties.java branches/ODE/RiftSaw-ODE-trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/main/java/org/apache/ode/store/ProcessConfImpl.java branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/main/java/org/apache/ode/store/ProcessStoreImpl.java branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/test/java/org/apache/ode/store/ProcessStoreTest.java branches/ODE/RiftSaw-ODE-trunk/bpel-test/src/main/java/org/apache/ode/test/BPELTestAbstract.java Log: * RIFTSAW-282, abstract a cache API that can be implemented by ConcurrentHashMap and JBoss Cache. Modified: branches/ODE/RiftSaw-ODE-trunk/axis2/src/main/java/org/apache/ode/axis2/ODEServer.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/axis2/src/main/java/org/apache/ode/axis2/ODEServer.java 2010-11-04 04:28:29 UTC (rev 1081) +++ branches/ODE/RiftSaw-ODE-trunk/axis2/src/main/java/org/apache/ode/axis2/ODEServer.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -63,6 +63,7 @@ import org.apache.ode.bpel.engine.cron.CronScheduler; import org.apache.ode.bpel.extvar.jdbc.JdbcExternalVariableModule; import org.apache.ode.bpel.iapi.BpelEventListener; +import org.apache.ode.bpel.iapi.CacheProvider; import org.apache.ode.bpel.iapi.EndpointReferenceContext; import org.apache.ode.bpel.iapi.ProcessConf; import org.apache.ode.bpel.iapi.ProcessStoreEvent; @@ -75,6 +76,7 @@ import org.apache.ode.dao.bpel.BpelDAOConnectionFactory; import org.apache.ode.dao.scheduler.SchedulerDAOConnectionFactory; import org.apache.ode.dao.store.ConfStoreDAOConnectionFactory; +import org.apache.ode.il.cache.CacheProviderFactory; import org.apache.ode.il.dbutil.Database; import org.apache.ode.scheduler.simple.SimpleScheduler; import org.apache.ode.store.ProcessStoreImpl; @@ -114,6 +116,8 @@ protected ConfStoreDAOConnectionFactory _storeDaoCF; protected SchedulerDAOConnectionFactory _schedDaoCF; + + protected CacheProvider _cacheProvider; protected ExecutorService _executorService; @@ -179,7 +183,8 @@ initDataSource(); __log.debug("Starting DAO."); initDAO(); - EndpointReferenceContextImpl eprContext = new EndpointReferenceContextImpl(this); + EndpointReferenceContextImpl eprContext = new EndpointReferenceContextImpl(this); + initCacheProvider(); __log.debug("Initializing BPEL process store."); initProcessStore(eprContext); __log.debug("Initializing BPEL server."); @@ -228,7 +233,17 @@ __log.info(__msgs.msgOdeStarted()); } - @SuppressWarnings("unchecked") + private void initCacheProvider() { + _cacheProvider = CacheProviderFactory.getCacheProvider(_odeConfig); + try { + _cacheProvider.start(); + } catch (Exception e) { + __log.error("Error in starting cache provider"); + throw new RuntimeException("Error in initCacheProvider.", e); + } + } + + @SuppressWarnings("unchecked") private DeploymentPoller getDeploymentPollerExt() { DeploymentPoller poller = null; @@ -297,6 +312,14 @@ } catch (Throwable t) { __log.debug("Error stopping poller.", t); } + + if (_cacheProvider != null) { + try { + _cacheProvider.stop(); + } catch(Exception e) { + __log.debug("Error stopping cache provider.", e); + } + } if (_bpelServer != null) try { @@ -463,7 +486,7 @@ } protected ProcessStoreImpl createProcessStore(EndpointReferenceContext eprContext, DataSource ds) { - return new ProcessStoreImpl(eprContext,_txMgr,_storeDaoCF); + return new ProcessStoreImpl(eprContext,_txMgr,_storeDaoCF, _cacheProvider); } protected Scheduler createScheduler() { Added: branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/Cache.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/Cache.java (rev 0) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/Cache.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -0,0 +1,44 @@ +/* + * 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.iapi; + +import java.util.Collection; +import java.util.Set; + +/** + * This is the cache API that ODE will use for the consistency. + * + * @author jeffyu + * + */ +public interface Cache<K, V> { + + public V put(K key, V value); + + public V get(K key); + + public V remove(K key); + + public boolean containsKey(K key); + + public Set<K> keySet(); + + public Collection<V> values(); + +} Added: branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/CacheProvider.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/CacheProvider.java (rev 0) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/CacheProvider.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -0,0 +1,34 @@ +/* + * 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.iapi; + +/** + * + * @author jeffyu + * + */ +public interface CacheProvider{ + + public void start() throws Exception; + + public <K, V> Cache<K, V> createCache(); + + public void stop() throws Exception; + +} Modified: branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/ProcessState.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/ProcessState.java 2010-11-04 04:28:29 UTC (rev 1081) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-api/src/main/java/org/apache/ode/bpel/iapi/ProcessState.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -18,11 +18,13 @@ */ package org.apache.ode.bpel.iapi; +import java.io.Serializable; + /** * The user-selectable states of a process in the configuration store. * @author mszefler */ -public enum ProcessState { +public enum ProcessState implements Serializable{ /** Process can create new instances and execute old instances. */ ACTIVE, Added: branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/CacheProviderFactory.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/CacheProviderFactory.java (rev 0) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/CacheProviderFactory.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -0,0 +1,50 @@ +/* + * 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.il.cache; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.ode.bpel.iapi.CacheProvider; +import org.apache.ode.il.config.OdeConfigProperties; + +/** + * This is Cache Provider Factory, users should add following lines into ode configuration files. see {@CacheProvider} + * + * cache.provider = org.apache.ode.bpel.engine.DefaultCacheProvider (updating this value by using your own CacheProvider) + * + * @author jeffyu + * + */ +public class CacheProviderFactory { + + private static final Log __log = LogFactory.getLog(CacheProviderFactory.class); + + public static CacheProvider getCacheProvider(OdeConfigProperties props) { + Class<?> clz; + CacheProvider cp = null; + try { + clz = Thread.currentThread().getContextClassLoader().loadClass(props.getCacheProvider()); + cp = (CacheProvider)clz.newInstance(); + } catch (Exception e) { + __log.error("Error in creating cache provider by using " + props.getCacheProvider()); + throw new RuntimeException("Error in creating cache provider.", e); + } + return cp; + } +} Added: branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/DefaultCacheProvider.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/DefaultCacheProvider.java (rev 0) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/DefaultCacheProvider.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -0,0 +1,43 @@ +/* + * 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.il.cache; + +import org.apache.ode.bpel.iapi.Cache; +import org.apache.ode.bpel.iapi.CacheProvider; + +/** + * + * @author jeffyu + * + */ +public class DefaultCacheProvider implements CacheProvider{ + + public void start() throws Exception { + + } + + public <K, V> Cache<K, V> createCache() { + return new HashMapCache<K, V>(); + } + + public void stop() throws Exception { + + } + +} Added: branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/HashMapCache.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/HashMapCache.java (rev 0) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/cache/HashMapCache.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -0,0 +1,61 @@ +/* + * 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.il.cache; + +import java.util.Collection; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.ode.bpel.iapi.Cache; + +/** + * Cache implemented by using ConcurrentHashMap. + * + * @author jeffyu + * + */ +public class HashMapCache<K, V> implements Cache<K, V> { + + private ConcurrentHashMap<K, V> map = new ConcurrentHashMap<K, V>(); + + public V put(K key, V value) { + return map.put(key, value); + } + + public V get(K key) { + return map.get(key); + } + + public V remove(K key) { + return map.remove(key); + } + + public boolean containsKey(K key) { + return map.containsKey(key); + } + + public Set<K> keySet() { + return map.keySet(); + } + + public Collection<V> values() { + return map.values(); + } + +} Modified: branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/config/OdeConfigProperties.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/config/OdeConfigProperties.java 2010-11-04 04:28:29 UTC (rev 1081) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-epr/src/main/java/org/apache/ode/il/config/OdeConfigProperties.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -106,10 +106,14 @@ public static final String PROP_DAOCF_SCHEDULER = "dao.factory.scheduler"; + public static final String PROP_CACHE_PROVIDER = "cache.provider"; + public static final String PROP_MIGRATION_TRANSACTION_TIMEOUT = "migration.transaction.timeout"; public static final String DEFAULT_TX_FACTORY_CLASS_NAME = "org.apache.ode.il.EmbeddedGeronimoFactory"; + public static final String DEFAULT_CACHE_PROVIDER = "org.apache.ode.il.cache.DefaultCacheProvider"; + private File _cfgFile; private String _prefix; @@ -239,6 +243,10 @@ public String getDAOConfScheduleConnectionFactory() { return getProperty(PROP_DAOCF_SCHEDULER, DEFAULT_DAOCF_SCHEDULER_CLASS); } + + public String getCacheProvider() { + return getProperty(PROP_CACHE_PROVIDER, DEFAULT_CACHE_PROVIDER); + } public String getDbDataSource() { return getProperty(OdeConfigProperties.PROP_DB_EXTERNAL_DS, "java:comp/env/jdbc/ode-ds"); Modified: branches/ODE/RiftSaw-ODE-trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java 2010-11-04 04:28:29 UTC (rev 1081) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-runtime/src/test/java/org/apache/ode/bpel/runtime/MockBpelServer.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -37,6 +37,7 @@ import org.apache.ode.bpel.engine.BpelServerImpl; import org.apache.ode.bpel.iapi.BindingContext; +import org.apache.ode.bpel.iapi.CacheProvider; import org.apache.ode.bpel.iapi.ContextException; import org.apache.ode.bpel.iapi.Endpoint; import org.apache.ode.bpel.iapi.EndpointReference; @@ -52,6 +53,7 @@ import org.apache.ode.dao.store.ConfStoreDAOConnectionFactory; import org.apache.ode.il.EmbeddedGeronimoFactory; import org.apache.ode.il.MockScheduler; +import org.apache.ode.il.cache.CacheProviderFactory; import org.apache.ode.il.config.OdeConfigProperties; import org.apache.ode.il.dbutil.Database; import org.apache.ode.store.ProcessStoreImpl; @@ -71,6 +73,7 @@ SchedulerWrapper _scheduler; BpelDAOConnectionFactory _bpelDaoCF; ConfStoreDAOConnectionFactory _confDaoCF; + CacheProvider _cacheProvider; EndpointReferenceContext _eprContext; MessageExchangeContext _mexContext; BindingContext _bindContext; @@ -92,9 +95,10 @@ if (_scheduler == null) throw new RuntimeException("No scheduler"); createEndpointReferenceContext(); + createCacheProvider(); Properties storeProps = new Properties(); storeProps.setProperty("hibernate.hbm2ddl.auto", "update"); - _store = new ProcessStoreImpl(_eprContext,_txManager,_confDaoCF); + _store = new ProcessStoreImpl(_eprContext,_txManager,_confDaoCF, _cacheProvider); _server.setScheduler(_scheduler); _server.setEndpointReferenceContext(_eprContext); _server.setMessageExchangeContext(createMessageExchangeContext()); @@ -108,7 +112,16 @@ } } - public Collection<QName> deploy(File deploymentUnitDirectory) { + private void createCacheProvider() throws Exception{ + Properties props = new Properties(); + props.setProperty(OdeConfigProperties.PROP_CACHE_PROVIDER, System.getProperty(OdeConfigProperties.PROP_CACHE_PROVIDER, OdeConfigProperties.DEFAULT_CACHE_PROVIDER)); + OdeConfigProperties odeProps = new OdeConfigProperties(props,""); + + _cacheProvider = CacheProviderFactory.getCacheProvider(odeProps); + _cacheProvider.start(); + } + + public Collection<QName> deploy(File deploymentUnitDirectory) { Collection<QName> pids = _store.deploy(deploymentUnitDirectory); for (QName pid: pids) _server.register(_store.getProcessConfiguration(pid)); @@ -157,6 +170,7 @@ } public void shutdown() throws Exception { + _cacheProvider.stop(); _server.stop(); _scheduler.stop(); _scheduler.shutdown(); Modified: branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/main/java/org/apache/ode/store/ProcessConfImpl.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/main/java/org/apache/ode/store/ProcessConfImpl.java 2010-11-04 04:28:29 UTC (rev 1081) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/main/java/org/apache/ode/store/ProcessConfImpl.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -24,6 +24,7 @@ import java.io.FileNotFoundException; import java.io.IOException; import java.io.InputStream; +import java.io.Serializable; import java.net.URI; import java.text.ParseException; import java.util.ArrayList; @@ -53,6 +54,7 @@ import org.apache.ode.bpel.dd.TScopeEvents; import org.apache.ode.bpel.dd.TService; import org.apache.ode.bpel.evt.BpelEvent; +import org.apache.ode.bpel.iapi.Cache; import org.apache.ode.bpel.iapi.ContextException; import org.apache.ode.bpel.iapi.Endpoint; import org.apache.ode.bpel.iapi.EndpointReference; @@ -78,7 +80,7 @@ * @author mriou <mriou at apache dot org> */ public class ProcessConfImpl implements ProcessConf { - private static final Log __log = LogFactory.getLog(ProcessConfImpl.class); + private static final Log __log = LogFactory.getLog(ProcessConfImpl.class); private final Date _deployDate; private File _configDir; @@ -90,9 +92,8 @@ private final ArrayList<QName> _sharedServices = new ArrayList<QName>(); private final Map<String, Set<BpelEvent.TYPE>> _events = new HashMap<String, Set<BpelEvent.TYPE>>(); private final ArrayList<String> _mexi = new ArrayList<String>(); - ProcessState _state; - final TDeployment.Process _pinfo; - final DeploymentUnitDir _du; + private final TDeployment.Process _pinfo; + private final DeploymentUnitDir _du; private long _version = 0; private QName _pid; private QName _type; @@ -101,16 +102,18 @@ private volatile boolean _inMemory = false; // monitor the IL property file and reload it if necessary - private WatchDog<Map<File, Long>, PropertiesObserver> propertiesWatchDog; + private WatchDog<Map<File, Long>, PropertiesObserver> propertiesWatchDog; private EndpointReferenceContext eprContext; private final ProcessCleanupConfImpl processCleanupConfImpl; private final boolean generateProcessEventsAll; + + private Cache<QName, ProcessState> _statesCache; ProcessConfImpl(QName pid, QName type, long version, DeploymentUnitDir du, TDeployment.Process pinfo, Date deployDate, - Map<QName, Node> props, ProcessState pstate, EndpointReferenceContext eprContext, File configDir, boolean generateProcessEventsAll) { + Map<QName, Node> props, Cache<QName, ProcessState> states, EndpointReferenceContext eprContext, File configDir, boolean generateProcessEventsAll) { _pid = pid; _version = version; _du = du; @@ -118,7 +121,7 @@ _deployDate = deployDate; _configDir = configDir; _props = Collections.unmodifiableMap(props); - _state = pstate; + _statesCache = states; _type = type; _inMemory = _pinfo.isSetInMemory() && _pinfo.getInMemory(); this.generateProcessEventsAll = generateProcessEventsAll; @@ -292,11 +295,11 @@ } public ProcessState getState() { - return _state; + return _statesCache.get(_pid); } void setState(ProcessState state) { - _state = state; + _statesCache.put(_pid, state); } public List<String> getMexInterceptors(QName processId) { Modified: branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/main/java/org/apache/ode/store/ProcessStoreImpl.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/main/java/org/apache/ode/store/ProcessStoreImpl.java 2010-11-04 04:28:29 UTC (rev 1081) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/main/java/org/apache/ode/store/ProcessStoreImpl.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -31,6 +31,7 @@ import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -50,6 +51,8 @@ import org.apache.ode.bpel.compiler.api.CompilationException; import org.apache.ode.bpel.dd.DeployDocument; import org.apache.ode.bpel.dd.TDeployment; +import org.apache.ode.bpel.iapi.Cache; +import org.apache.ode.bpel.iapi.CacheProvider; import org.apache.ode.bpel.iapi.ContextException; import org.apache.ode.bpel.iapi.EndpointReferenceContext; import org.apache.ode.bpel.iapi.ProcessConf; @@ -93,9 +96,11 @@ private final CopyOnWriteArrayList<ProcessStoreListener> _listeners = new CopyOnWriteArrayList<ProcessStoreListener>(); - private Map<QName, ProcessConfImpl> _processes = new HashMap<QName, ProcessConfImpl>(); + private Map<QName, ProcessConfImpl> _processes = new ConcurrentHashMap<QName, ProcessConfImpl>(); + + private Cache<QName, ProcessState> _processStates; - protected Map<String, DeploymentUnitDir> _deploymentUnits = new HashMap<String, DeploymentUnitDir>(); + protected Map<String, DeploymentUnitDir> _deploymentUnits = new ConcurrentHashMap<String, DeploymentUnitDir>(); /** Guards access to the _processes and _deploymentUnits */ private final ReadWriteLock _rw = new ReentrantReadWriteLock(); @@ -111,6 +116,8 @@ protected File _deployDir; protected File _configDir; + + private CacheProvider _cacheProvider; /** * Executor used to process DB transactions. Allows us to isolate the TX context, and to ensure that only one TX gets executed a @@ -128,10 +135,13 @@ /** * Constructor that hardwires OpenJPA on a new in-memory database. Suitable for tests. */ - public <E> ProcessStoreImpl(EndpointReferenceContext eprContext, TransactionManager mgr, ConfStoreDAOConnectionFactory cf) { + public <E> ProcessStoreImpl(EndpointReferenceContext eprContext, TransactionManager mgr, + ConfStoreDAOConnectionFactory cf, CacheProvider cacheProvider) { _eprContext = eprContext; _txm = mgr; _cf=cf; + _cacheProvider = cacheProvider; + _processStates = _cacheProvider.createCache(); } public void shutdown() { @@ -201,7 +211,7 @@ du.scan(); final DeployDocument dd = du.getDeploymentDescriptor(); - final ArrayList<ProcessConfImpl> processes = new ArrayList<ProcessConfImpl>(); + final ArrayList<ProcessConfImpl> processes = new ArrayList<ProcessConfImpl>(); Collection<QName> deployed; _rw.writeLock().lock(); @@ -232,9 +242,11 @@ __log.error(errmsg); throw new ContextException(errmsg); } - + + ProcessState state = calcInitialState(processDD); + _processStates.put(pid, state); ProcessConfImpl pconf = new ProcessConfImpl(pid, processDD.getName(), version, du, processDD, deployDate, - calcInitialProperties(du.getProperties(), processDD), calcInitialState(processDD), _eprContext, _configDir, generateProcessEventsAll); + calcInitialProperties(du.getProperties(), processDD), _processStates, _eprContext, _configDir, generateProcessEventsAll); processes.add(pconf); } @@ -481,6 +493,7 @@ }); pconf.setState(state); + if (old != null && old != state) fireStateChange(pid, state, pconf.getDeploymentUnit().getName()); } @@ -765,9 +778,11 @@ Map<QName, Node> props = calcInitialProperties(dud.getProperties(), pinfo); // TODO: update the props based on the values in the DB. - + + _processStates.put(p.getPID(), p.getState()); + ProcessConfImpl pconf = new ProcessConfImpl(p.getPID(), p.getType(), p.getVersion(), dud, pinfo, dudao - .getDeployDate(), props, p.getState(), _eprContext, _configDir, generateProcessEventsAll); + .getDeployDate(), props, _processStates, _eprContext, _configDir, generateProcessEventsAll); version = p.getVersion(); _processes.put(pconf.getProcessId(), pconf); Modified: branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/test/java/org/apache/ode/store/ProcessStoreTest.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/test/java/org/apache/ode/store/ProcessStoreTest.java 2010-11-04 04:28:29 UTC (rev 1081) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-store/src/test/java/org/apache/ode/store/ProcessStoreTest.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -30,9 +30,11 @@ import junit.framework.TestCase; +import org.apache.ode.bpel.iapi.CacheProvider; import org.apache.ode.bpel.iapi.ProcessConf; import org.apache.ode.bpel.iapi.ProcessConf.CLEANUP_CATEGORY; import org.apache.ode.dao.store.ConfStoreDAOConnectionFactory; +import org.apache.ode.il.cache.CacheProviderFactory; import org.apache.ode.il.config.OdeConfigProperties; import org.apache.ode.il.dbutil.Database; import org.apache.ode.il.txutil.TxManager; @@ -43,10 +45,12 @@ Database _db; ConfStoreDAOConnectionFactory _cf; private File _testdd; + private CacheProvider _cp; public void setUp() throws Exception { Properties props = new Properties(); props.setProperty(OdeConfigProperties.PROP_DAOCF_STORE,System.getProperty(OdeConfigProperties.PROP_DAOCF_STORE,OdeConfigProperties.DEFAULT_DAOCF_STORE_CLASS)); + props.setProperty(OdeConfigProperties.PROP_CACHE_PROVIDER, OdeConfigProperties.DEFAULT_CACHE_PROVIDER); OdeConfigProperties odeProps = new OdeConfigProperties(props, ""); _db = new Database(odeProps); TxManager tx = new TxManager(odeProps); @@ -54,7 +58,9 @@ _db.setTransactionManager(txm); _db.start(); _cf = _db.createDaoStoreCF(); - _ps = new ProcessStoreImpl(null,txm, _cf); + _cp = CacheProviderFactory.getCacheProvider(odeProps) ; + _cp.start(); + _ps = new ProcessStoreImpl(null,txm, _cf, _cp); _ps.loadAll(); URI tdd = getClass().getResource("/testdd/deploy.xml").toURI(); _testdd = new File(tdd.getPath()).getParentFile(); @@ -62,6 +68,7 @@ } public void tearDown() throws Exception { + _cp.stop(); _ps.shutdown(); _cf.shutdown(); _db.shutdown(); Modified: branches/ODE/RiftSaw-ODE-trunk/bpel-test/src/main/java/org/apache/ode/test/BPELTestAbstract.java =================================================================== --- branches/ODE/RiftSaw-ODE-trunk/bpel-test/src/main/java/org/apache/ode/test/BPELTestAbstract.java 2010-11-04 04:28:29 UTC (rev 1081) +++ branches/ODE/RiftSaw-ODE-trunk/bpel-test/src/main/java/org/apache/ode/test/BPELTestAbstract.java 2010-11-05 04:12:41 UTC (rev 1082) @@ -41,6 +41,7 @@ import org.apache.ode.bpel.common.evt.DebugBpelEventListener; import org.apache.ode.bpel.engine.BpelServerImpl; +import org.apache.ode.bpel.iapi.CacheProvider; import org.apache.ode.bpel.iapi.EndpointReference; import org.apache.ode.bpel.iapi.EndpointReferenceContext; import org.apache.ode.bpel.iapi.Message; @@ -55,6 +56,7 @@ import org.apache.ode.dao.bpel.BpelDAOConnectionFactory; import org.apache.ode.dao.store.ConfStoreDAOConnectionFactory; import org.apache.ode.il.MockScheduler; +import org.apache.ode.il.cache.CacheProviderFactory; import org.apache.ode.il.config.OdeConfigProperties; import org.apache.ode.il.dbutil.Database; import org.apache.ode.store.ProcessConfImpl; @@ -102,6 +104,8 @@ private MockTransactionManager _txm; private Database _db; + + private CacheProvider _cp; @Before public void setUp() throws Exception { @@ -115,6 +119,7 @@ _txm = new MockTransactionManager(); Properties props = new Properties(); props.setProperty(OdeConfigProperties.PROP_DAOCF_STORE, System.getProperty(OdeConfigProperties.PROP_DAOCF_STORE,OdeConfigProperties.DEFAULT_DAOCF_STORE_CLASS)); + props.setProperty(OdeConfigProperties.PROP_CACHE_PROVIDER, System.getProperty(OdeConfigProperties.PROP_CACHE_PROVIDER, OdeConfigProperties.DEFAULT_CACHE_PROVIDER)); OdeConfigProperties odeProps = new OdeConfigProperties(props,""); _db = new Database(odeProps); _db.setTransactionManager(_txm); @@ -168,7 +173,11 @@ }; _scf = _db.createDaoStoreCF(); - store = new ProcessStoreImpl(eprContext, _txm, _scf); + + _cp = CacheProviderFactory.getCacheProvider(odeProps); + _cp.start(); + + store = new ProcessStoreImpl(eprContext, _txm, _scf, _cp); store.registerListener(new ProcessStoreListener() { public void onProcessStoreEvent(ProcessStoreEvent event) { // bounce the process @@ -200,7 +209,8 @@ if (em != null) em.close(); if (emf != null) emf.close(); - + + _cp.stop(); _server.stop(); _failures = null; _deployed = null;