From a1f0ecbb6a2bea6aa214322c412af2f3c5ce124b Mon Sep 17 00:00:00 2001 From: Andrew Branson Date: Wed, 7 May 2014 17:33:13 +0200 Subject: Agent now extends Item, so they can have workflows. All traces of the old 'Entity' superclasses should be removed, including proxies and paths. Very large change, breaks API compatibility with CRISTAL 2.x. Fixes #135 --- .../com/c2kernel/entity/AgentImplementation.java | 78 +++++ src/main/java/com/c2kernel/entity/CorbaServer.java | 14 +- .../com/c2kernel/entity/ItemImplementation.java | 281 +++++++++++++++++ .../java/com/c2kernel/entity/TraceableEntity.java | 207 +------------ .../com/c2kernel/entity/agent/ActiveEntity.java | 263 ++++------------ src/main/java/com/c2kernel/entity/agent/Job.java | 8 +- .../java/com/c2kernel/entity/proxy/AgentProxy.java | 63 ++-- .../com/c2kernel/entity/proxy/EntityProxy.java | 247 --------------- .../c2kernel/entity/proxy/EntityProxyManager.java | 339 --------------------- .../c2kernel/entity/proxy/EntityProxyObserver.java | 27 -- .../java/com/c2kernel/entity/proxy/ItemProxy.java | 285 +++++++++++++---- .../c2kernel/entity/proxy/MemberSubscription.java | 18 +- .../entity/proxy/ProxyClientConnection.java | 2 +- .../com/c2kernel/entity/proxy/ProxyManager.java | 337 ++++++++++++++++++++ .../com/c2kernel/entity/proxy/ProxyObserver.java | 27 ++ .../entity/proxy/ProxyServerConnection.java | 4 +- .../com/c2kernel/entity/transfer/TransferItem.java | 9 +- .../com/c2kernel/entity/transfer/TransferSet.java | 4 +- 18 files changed, 1077 insertions(+), 1136 deletions(-) create mode 100644 src/main/java/com/c2kernel/entity/AgentImplementation.java create mode 100644 src/main/java/com/c2kernel/entity/ItemImplementation.java delete mode 100644 src/main/java/com/c2kernel/entity/proxy/EntityProxy.java delete mode 100644 src/main/java/com/c2kernel/entity/proxy/EntityProxyManager.java delete mode 100644 src/main/java/com/c2kernel/entity/proxy/EntityProxyObserver.java create mode 100644 src/main/java/com/c2kernel/entity/proxy/ProxyManager.java create mode 100644 src/main/java/com/c2kernel/entity/proxy/ProxyObserver.java (limited to 'src/main/java/com/c2kernel/entity') diff --git a/src/main/java/com/c2kernel/entity/AgentImplementation.java b/src/main/java/com/c2kernel/entity/AgentImplementation.java new file mode 100644 index 0000000..d31b94a --- /dev/null +++ b/src/main/java/com/c2kernel/entity/AgentImplementation.java @@ -0,0 +1,78 @@ +package com.c2kernel.entity; + +import com.c2kernel.common.CannotManageException; +import com.c2kernel.common.ObjectCannotBeUpdated; +import com.c2kernel.common.ObjectNotFoundException; +import com.c2kernel.entity.agent.Job; +import com.c2kernel.entity.agent.JobArrayList; +import com.c2kernel.entity.agent.JobList; +import com.c2kernel.lookup.AgentPath; +import com.c2kernel.lookup.InvalidItemPathException; +import com.c2kernel.lookup.RolePath; +import com.c2kernel.process.Gateway; +import com.c2kernel.utils.Logger; + +public class AgentImplementation extends ItemImplementation implements + AgentOperations { + + private JobList currentJobs; + + public AgentImplementation(int systemKey) { + super(systemKey); + } + + /** + * Called by an activity when it reckons we need to update our joblist for it + */ + + @Override + public synchronized void refreshJobList(int sysKey, String stepPath, String newJobs) { + try { + JobArrayList newJobList = (JobArrayList)Gateway.getMarshaller().unmarshall(newJobs); + + // get our joblist + if (currentJobs == null) + currentJobs = new JobList( mSystemKey, null); + + // remove old jobs for this item + currentJobs.removeJobsForStep( sysKey, stepPath ); + + // merge new jobs in + for (Object name : newJobList.list) { + Job newJob = (Job)name; + Logger.msg(6, "Adding job for "+newJob.getItemSysKey()+"/"+newJob.getStepPath()+":"+newJob.getTransition().getId()); + currentJobs.addJob(newJob); + } + + } catch (Throwable ex) { + Logger.error("Could not refresh job list."); + Logger.error(ex); + } + + } + + @Override + public void addRole(String roleName) throws CannotManageException, ObjectNotFoundException { + RolePath newRole = Gateway.getLDAPLookup().getRoleManager().getRolePath(roleName); + try { + newRole.addAgent(new AgentPath(mSystemKey)); + } catch (InvalidItemPathException ex) { + throw new CannotManageException("Invalid syskey for agent: "+mSystemKey, ""); + } catch (ObjectCannotBeUpdated ex) { + throw new CannotManageException("Could not update role"); + } + } + + @Override + public void removeRole(String roleName) throws CannotManageException, ObjectNotFoundException { + RolePath rolePath = Gateway.getLDAPLookup().getRoleManager().getRolePath(roleName); + try { + rolePath.removeAgent(new AgentPath(mSystemKey)); + } catch (InvalidItemPathException e) { + throw new CannotManageException("Invalid syskey for agent: "+mSystemKey, ""); + } catch (ObjectCannotBeUpdated ex) { + throw new CannotManageException("Could not update role"); + } + } + +} diff --git a/src/main/java/com/c2kernel/entity/CorbaServer.java b/src/main/java/com/c2kernel/entity/CorbaServer.java index 3d7c79b..4a129ae 100644 --- a/src/main/java/com/c2kernel/entity/CorbaServer.java +++ b/src/main/java/com/c2kernel/entity/CorbaServer.java @@ -14,8 +14,8 @@ import com.c2kernel.common.ObjectNotFoundException; import com.c2kernel.entity.agent.ActiveEntity; import com.c2kernel.entity.agent.ActiveLocator; import com.c2kernel.lookup.AgentPath; -import com.c2kernel.lookup.EntityPath; -import com.c2kernel.lookup.InvalidEntityPathException; +import com.c2kernel.lookup.ItemPath; +import com.c2kernel.lookup.InvalidItemPathException; import com.c2kernel.process.Gateway; import com.c2kernel.utils.Logger; import com.c2kernel.utils.SoftCache; @@ -31,14 +31,14 @@ import com.c2kernel.utils.SoftCache; public class CorbaServer { - private final Map mEntityCache; + private final Map mEntityCache; private POA mRootPOA; private POA mItemPOA; private POA mAgentPOA; private POAManager mPOAManager; public CorbaServer() throws InvalidDataException { - mEntityCache = new SoftCache(50); + mEntityCache = new SoftCache(50); // init POA try { @@ -119,7 +119,7 @@ public class CorbaServer { **************************************************************************/ private Servant getEntity(int sysKey, org.omg.PortableServer.POA poa) throws ObjectNotFoundException { try { - EntityPath entityPath = new EntityPath(sysKey); + ItemPath entityPath = new ItemPath(sysKey); Servant entity = null; synchronized (mEntityCache) { entity = mEntityCache.get(entityPath); @@ -141,7 +141,7 @@ public class CorbaServer { } return entity; - } catch (InvalidEntityPathException ex) { + } catch (InvalidItemPathException ex) { throw new ObjectNotFoundException("Invalid Entity Key", ""); } } @@ -164,7 +164,7 @@ public class CorbaServer { * @param entityPath * @return */ - public Servant createEntity(EntityPath entityPath) throws CannotManageException, ObjectAlreadyExistsException { + public Servant createEntity(ItemPath entityPath) throws CannotManageException, ObjectAlreadyExistsException { try { if (entityPath == null) entityPath = Gateway.getLDAPLookup().getNextKeyManager().generateNextEntityKey(); diff --git a/src/main/java/com/c2kernel/entity/ItemImplementation.java b/src/main/java/com/c2kernel/entity/ItemImplementation.java new file mode 100644 index 0000000..e0d107a --- /dev/null +++ b/src/main/java/com/c2kernel/entity/ItemImplementation.java @@ -0,0 +1,281 @@ +package com.c2kernel.entity; + +import com.c2kernel.collection.Collection; +import com.c2kernel.collection.CollectionArrayList; +import com.c2kernel.common.AccessRightsException; +import com.c2kernel.common.InvalidDataException; +import com.c2kernel.common.InvalidTransitionException; +import com.c2kernel.common.ObjectAlreadyExistsException; +import com.c2kernel.common.ObjectNotFoundException; +import com.c2kernel.common.PersistencyException; +import com.c2kernel.entity.agent.JobArrayList; +import com.c2kernel.lifecycle.instance.CompositeActivity; +import com.c2kernel.lifecycle.instance.Workflow; +import com.c2kernel.lookup.AgentPath; +import com.c2kernel.lookup.InvalidItemPathException; +import com.c2kernel.persistency.ClusterStorage; +import com.c2kernel.persistency.ClusterStorageException; +import com.c2kernel.persistency.TransactionManager; +import com.c2kernel.process.Gateway; +import com.c2kernel.property.Property; +import com.c2kernel.property.PropertyArrayList; +import com.c2kernel.utils.Logger; + +public class ItemImplementation implements ItemOperations { + + protected final TransactionManager mStorage; + protected final int mSystemKey; + + protected ItemImplementation(int systemKey) { + this.mStorage = Gateway.getStorage(); + this.mSystemKey = systemKey; + } + + @Override + public int getSystemKey() { + // TODO Auto-generated method stub + return 0; + } + + @Override + public void initialise(int agentId, String propString, String initWfString, + String initCollsString) throws AccessRightsException, + InvalidDataException, PersistencyException + { + Logger.msg(5, "Item::initialise("+mSystemKey+") - agent:"+agentId); + Object locker = new Object(); + + AgentPath agentPath; + try { + agentPath = new AgentPath(agentId); + } catch (InvalidItemPathException e) { + throw new AccessRightsException("Invalid Agent Id:" + agentId); + } + + // must supply properties + if (propString == null || propString.length() == 0) { + throw new InvalidDataException("No properties supplied", ""); + } + + // store properties + try { + PropertyArrayList props = (PropertyArrayList) Gateway + .getMarshaller().unmarshall(propString); + for (Property thisProp : props.list) + mStorage.put(mSystemKey, thisProp, locker); + } catch (Throwable ex) { + Logger.msg(8, "TraceableEntity::initialise(" + mSystemKey + + ") - Properties were invalid: " + propString); + Logger.error(ex); + mStorage.abort(locker); + throw new InvalidDataException("Properties were invalid", ""); + } + + // create wf + try { + Workflow lc = null; + if (initWfString == null || initWfString.length() == 0) + lc = new Workflow(new CompositeActivity()); + else + lc = new Workflow((CompositeActivity) Gateway + .getMarshaller().unmarshall(initWfString)); + lc.initialise(mSystemKey, agentPath); + mStorage.put(mSystemKey, lc, locker); + } catch (Throwable ex) { + Logger.msg(8, "TraceableEntity::initialise(" + mSystemKey + + ") - Workflow was invalid: " + initWfString); + Logger.error(ex); + mStorage.abort(locker); + throw new InvalidDataException("Workflow was invalid", ""); + } + + // init collections + if (initCollsString != null && initCollsString.length() > 0) { + try { + CollectionArrayList colls = (CollectionArrayList) Gateway + .getMarshaller().unmarshall(initCollsString); + for (Collection thisColl : colls.list) { + mStorage.put(mSystemKey, thisColl, locker); + } + } catch (Throwable ex) { + Logger.msg(8, "TraceableEntity::initialise(" + mSystemKey + + ") - Collections were invalid: " + + initCollsString); + Logger.error(ex); + mStorage.abort(locker); + throw new InvalidDataException("Collections were invalid"); + } + } + mStorage.commit(locker); + Logger.msg(3, "Initialisation of item " + mSystemKey + + " was successful"); + } + + + @Override + public void requestAction(int agentId, String stepPath, int transitionID, + String requestData) throws AccessRightsException, + InvalidTransitionException, ObjectNotFoundException, + InvalidDataException, PersistencyException, + ObjectAlreadyExistsException { + + try { + + Logger.msg(1, "TraceableEntity::request(" + mSystemKey + ") - " + + transitionID + " " + stepPath + " by " + agentId); + + AgentPath agent = new AgentPath(agentId); + Workflow lifeCycle = (Workflow) mStorage.get(mSystemKey, + ClusterStorage.LIFECYCLE + "/workflow", null); + + lifeCycle.requestAction(agent, stepPath, mSystemKey, + transitionID, requestData); + + // store the workflow if we've changed the state of the domain + // wf + if (!(stepPath.startsWith("workflow/predefined"))) + mStorage.put(mSystemKey, lifeCycle, null); + + // Normal operation exceptions + } catch (AccessRightsException ex) { + Logger.msg("Propagating AccessRightsException back to the calling agent"); + throw ex; + } catch (InvalidTransitionException ex) { + Logger.msg("Propagating InvalidTransitionException back to the calling agent"); + throw ex; + } catch (ObjectNotFoundException ex) { + Logger.msg("Propagating ObjectNotFoundException back to the calling agent"); + throw ex; + // errors + } catch (ClusterStorageException ex) { + Logger.error(ex); + throw new PersistencyException("Error on storage: " + + ex.getMessage(), ""); + } catch (InvalidItemPathException ex) { + Logger.error(ex); + throw new AccessRightsException("Invalid Agent Id: " + agentId, + ""); + } catch (InvalidDataException ex) { + Logger.error(ex); + Logger.msg("Propagating InvalidDataException back to the calling agent"); + throw ex; + } catch (ObjectAlreadyExistsException ex) { + Logger.error(ex); + Logger.msg("Propagating ObjectAlreadyExistsException back to the calling agent"); + throw ex; + // non-CORBA exception hasn't been caught! + } catch (Throwable ex) { + Logger.error("Unknown Error: requestAction on " + mSystemKey + + " by " + agentId + " executing " + stepPath); + Logger.error(ex); + throw new InvalidDataException( + "Extraordinary Exception during execution:" + + ex.getClass().getName() + " - " + + ex.getMessage(), ""); + } + } + + @Override + public String queryLifeCycle(int agentId, boolean filter) + throws AccessRightsException, ObjectNotFoundException, + PersistencyException { + Logger.msg(1, "TraceableEntity::queryLifeCycle(" + mSystemKey + + ") - agent: " + agentId); + try { + AgentPath agent; + try { + agent = new AgentPath(agentId); + } catch (InvalidItemPathException e) { + throw new AccessRightsException("Agent " + agentId + + " doesn't exist"); + } + Workflow wf; + try { + wf = (Workflow) mStorage.get(mSystemKey, + ClusterStorage.LIFECYCLE + "/workflow", null); + } catch (ClusterStorageException e) { + Logger.error("TraceableEntity::queryLifeCycle(" + + mSystemKey + ") - Error loading workflow"); + Logger.error(e); + throw new PersistencyException("Error loading workflow"); + } + JobArrayList jobBag = new JobArrayList(); + CompositeActivity domainWf = (CompositeActivity) wf + .search("workflow/domain"); + jobBag.list = filter ? domainWf.calculateJobs(agent, + mSystemKey, true) : domainWf.calculateAllJobs(agent, + mSystemKey, true); + Logger.msg(1, "TraceableEntity::queryLifeCycle(" + mSystemKey + + ") - Returning " + jobBag.list.size() + " jobs."); + try { + return Gateway.getMarshaller().marshall(jobBag); + } catch (Exception e) { + Logger.error(e); + throw new PersistencyException("Error marshalling job bag"); + } + } catch (Throwable ex) { + Logger.error("TraceableEntity::queryLifeCycle(" + mSystemKey + + ") - Unknown error"); + Logger.error(ex); + throw new PersistencyException( + "Unknown error querying jobs. Please see server log."); + } + } + + @Override + public String queryData(String path) throws AccessRightsException, + ObjectNotFoundException, PersistencyException { + + String result = ""; + + Logger.msg(1, "TraceableEntity::queryData(" + mSystemKey + ") - " + + path); + + try { // check for cluster contents query + + if (path.endsWith("/all")) { + int allPos = path.lastIndexOf("all"); + String query = path.substring(0, allPos); + String[] ids = mStorage.getClusterContents(mSystemKey, + query); + + for (int i = 0; i < ids.length; i++) { + result += ids[i]; + + if (i != ids.length - 1) + result += ","; + } + } + // **************************************************************** + else { // retrieve the object instead + C2KLocalObject obj = mStorage.get(mSystemKey, path, null); + + // marshall it, or in the case of an outcome get the data. + result = Gateway.getMarshaller().marshall(obj); + } + } catch (ObjectNotFoundException ex) { + throw ex; + } catch (Throwable ex) { + Logger.warning("TraceableEntity::queryData(" + mSystemKey + + ") - " + path + " Failed: " + ex.getClass().getName()); + throw new PersistencyException("Server exception: " + + ex.getClass().getName(), ""); + } + + if (Logger.doLog(9)) + Logger.msg(9, "TraceableEntity::queryData(" + mSystemKey + + ") - result:" + result); + + return result; + } + + /** + * + */ + @Override + protected void finalize() throws Throwable { + Logger.msg(7, "Item "+mSystemKey+" reaped"); + Gateway.getStorage().clearCache(mSystemKey, null); + super.finalize(); + } +} diff --git a/src/main/java/com/c2kernel/entity/TraceableEntity.java b/src/main/java/com/c2kernel/entity/TraceableEntity.java index b6ccd8c..ffd5859 100644 --- a/src/main/java/com/c2kernel/entity/TraceableEntity.java +++ b/src/main/java/com/c2kernel/entity/TraceableEntity.java @@ -18,17 +18,6 @@ import com.c2kernel.common.InvalidTransitionException; import com.c2kernel.common.ObjectAlreadyExistsException; import com.c2kernel.common.ObjectNotFoundException; import com.c2kernel.common.PersistencyException; -import com.c2kernel.entity.agent.JobArrayList; -import com.c2kernel.lifecycle.instance.CompositeActivity; -import com.c2kernel.lifecycle.instance.Workflow; -import com.c2kernel.lookup.AgentPath; -import com.c2kernel.lookup.InvalidEntityPathException; -import com.c2kernel.persistency.ClusterStorage; -import com.c2kernel.persistency.ClusterStorageException; -import com.c2kernel.persistency.TransactionManager; -import com.c2kernel.process.Gateway; -import com.c2kernel.property.Property; -import com.c2kernel.property.PropertyArrayList; import com.c2kernel.utils.Logger; /************************************************************************** @@ -62,10 +51,8 @@ import com.c2kernel.utils.Logger; public class TraceableEntity extends ItemPOA { - private final int mSystemKey; private final org.omg.PortableServer.POA mPoa; - private final TransactionManager mStorage; - + private final ItemImplementation mItemImpl; /************************************************************************** * Constructor used by the Locator only @@ -74,10 +61,8 @@ public class TraceableEntity extends ItemPOA org.omg.PortableServer.POA poa ) { Logger.msg(5,"TraceableEntity::constructor() - SystemKey:" + key ); - - mSystemKey = key; - mPoa = poa; - mStorage = Gateway.getStorage(); + mPoa = poa; + mItemImpl = new ItemImplementation(key); } @@ -100,8 +85,7 @@ public class TraceableEntity extends ItemPOA @Override public int getSystemKey() { - Logger.msg(8, "TraceableEntity::getSystemKey() - " + mSystemKey); - return mSystemKey; + return mItemImpl.getSystemKey(); } /************************************************************************** @@ -110,54 +94,15 @@ public class TraceableEntity extends ItemPOA @Override public void initialise( int agentId, String propString, - String initWfString + String initWfString, + String initCollsString ) throws AccessRightsException, InvalidDataException, PersistencyException { - Logger.msg(5, "TraceableEntity::initialise("+mSystemKey+") - agent:"+agentId); synchronized (this) { - Workflow lc = null; - PropertyArrayList props = null; - - AgentPath agentPath; - try { - agentPath = new AgentPath(agentId); - } catch (InvalidEntityPathException e) { - throw new AccessRightsException("Invalid Agent Id:" + agentId); - } - - //unmarshalling checks the validity of the received strings - - // create properties - if (!propString.equals("")) { - try { - props = (PropertyArrayList)Gateway.getMarshaller().unmarshall(propString); - for (Object name : props.list) { - Property thisProp = (Property)name; - mStorage.put(mSystemKey, thisProp, props); - } - } catch (Throwable ex) { - Logger.msg(8, "TraceableEntity::initialise("+mSystemKey+ ") - Properties were invalid: "+propString); - Logger.error(ex); - mStorage.abort(props); - } - mStorage.commit(props); - } - - // create wf - try { - if (initWfString == null || initWfString.equals("")) - lc = new Workflow(new CompositeActivity()); - else - lc = new Workflow((CompositeActivity)Gateway.getMarshaller().unmarshall(initWfString)); - lc.initialise(mSystemKey, agentPath); - mStorage.put(mSystemKey, lc, null); - } catch (Throwable ex) { - Logger.msg(8, "TraceableEntity::initialise("+mSystemKey+") - Workflow was invalid: "+initWfString); - Logger.error(ex); - } + mItemImpl.initialise(agentId, propString, initWfString, initCollsString); } } @@ -179,55 +124,7 @@ public class TraceableEntity extends ItemPOA ObjectAlreadyExistsException { synchronized (this) { - try { - - Logger.msg(1, "TraceableEntity::request("+mSystemKey+") - " + - transitionID + " "+stepPath + " by " +agentId ); - - AgentPath agent = new AgentPath(agentId); - Workflow lifeCycle = (Workflow)mStorage.get(mSystemKey, ClusterStorage.LIFECYCLE+"/workflow", null); - - lifeCycle.requestAction( agent, - stepPath, - mSystemKey, - transitionID, - requestData ); - - // store the workflow if we've changed the state of the domain wf - if (!(stepPath.startsWith("workflow/predefined"))) - mStorage.put(mSystemKey, lifeCycle, null); - - // Normal operation exceptions - } catch (AccessRightsException ex) { - Logger.msg("Propagating AccessRightsException back to the calling agent"); - throw ex; - } catch (InvalidTransitionException ex) { - Logger.msg("Propagating InvalidTransitionException back to the calling agent"); - throw ex; - } catch (ObjectNotFoundException ex) { - Logger.msg("Propagating ObjectNotFoundException back to the calling agent"); - throw ex; - // errors - } catch (ClusterStorageException ex) { - Logger.error(ex); - throw new PersistencyException("Error on storage: "+ex.getMessage(), ""); - } catch (InvalidEntityPathException ex) { - Logger.error(ex); - throw new AccessRightsException("Invalid Agent Id: "+agentId, ""); - } catch (InvalidDataException ex) { - Logger.error(ex); - Logger.msg("Propagating InvalidDataException back to the calling agent"); - throw ex; - } catch (ObjectAlreadyExistsException ex) { - Logger.error(ex); - Logger.msg("Propagating ObjectAlreadyExistsException back to the calling agent"); - throw ex; - // non-CORBA exception hasn't been caught! - } catch (Throwable ex) { - Logger.error("Unknown Error: requestAction on "+mSystemKey+" by "+agentId+" executing "+stepPath); - Logger.error(ex); - throw new InvalidDataException("Extraordinary Exception during execution:"+ex.getClass().getName()+" - "+ex.getMessage(), ""); - } + mItemImpl.requestAction(agentId, stepPath, transitionID, requestData); } } @@ -243,38 +140,7 @@ public class TraceableEntity extends ItemPOA PersistencyException { synchronized (this) { - Logger.msg(1, "TraceableEntity::queryLifeCycle("+mSystemKey+") - agent: " + agentId); - try { - AgentPath agent; - try { - agent = new AgentPath(agentId); - } catch (InvalidEntityPathException e) { - throw new AccessRightsException("Agent "+agentId+" doesn't exist"); - } - Workflow wf; - try { - wf = (Workflow)mStorage.get(mSystemKey, ClusterStorage.LIFECYCLE+"/workflow", null); - } catch (ClusterStorageException e) { - Logger.error("TraceableEntity::queryLifeCycle("+mSystemKey+") - Error loading workflow"); - Logger.error(e); - throw new PersistencyException("Error loading workflow"); - } - JobArrayList jobBag = new JobArrayList(); - CompositeActivity domainWf = (CompositeActivity)wf.search("workflow/domain"); - jobBag.list = filter?domainWf.calculateJobs(agent, mSystemKey, true):domainWf.calculateAllJobs(agent, mSystemKey, true); - Logger.msg(1, "TraceableEntity::queryLifeCycle("+mSystemKey+") - Returning "+jobBag.list.size()+" jobs."); - try { - return Gateway.getMarshaller().marshall( jobBag ); - } catch (Exception e) { - Logger.error(e); - throw new PersistencyException("Error marshalling job bag"); - } - } - catch ( Throwable ex ) { - Logger.error("TraceableEntity::queryLifeCycle("+mSystemKey+") - Unknown error"); - Logger.error(ex); - throw new PersistencyException("Unknown error querying jobs. Please see server log."); - } + return mItemImpl.queryLifeCycle(agentId, filter); } } @@ -296,60 +162,7 @@ public class TraceableEntity extends ItemPOA PersistencyException { synchronized (this) { - String result = ""; - - Logger.msg(1, "TraceableEntity::queryData("+mSystemKey+") - " + path ); - - try - { // check for cluster contents query - - if (path.endsWith("/all")) - { - int allPos = path.lastIndexOf("all"); - String query = path.substring(0,allPos); - String[] ids = mStorage.getClusterContents( mSystemKey, query ); - - for( int i=0; i transaction aborted!"); - } - - Logger.msg(5, "ActiveEntity::init() - completed."); - } - - /** - * - * @param propsString - * @return Properties - * @throws InvalidDataException Properties cannot be unmarshalled - * @throws ClusterStorageException - */ - private PropertyArrayList initProps( String propsString ) - throws InvalidDataException, - ClusterStorageException - { - PropertyArrayList props = null; - - // create properties - if( propsString != null && !propsString.equals("") ) - { - try - { - props = (PropertyArrayList)Gateway.getMarshaller().unmarshall(propsString); - } - catch( Exception ex ) - { - //any exception during unmarshall indicates that data was - //incorrect or the castor mapping was not set up - Logger.error(ex); - throw new InvalidDataException(ex.toString(), null); - } - - Iterator iter = props.list.iterator(); - - while( iter.hasNext() ) - mDatabase.put( mSystemKey, iter.next(), props ); - } - else - { - Logger.warning("ActiveEntity::initProps() - NO Properties!"); - } - - return props; - } /************************************************************************** * @@ -152,8 +49,8 @@ public class ActiveEntity extends AgentPOA @Override public org.omg.PortableServer.POA _default_POA() { - if(mPOA != null) - return mPOA; + if(mPoa != null) + return mPoa; else return super._default_POA(); } @@ -166,7 +63,7 @@ public class ActiveEntity extends AgentPOA @Override public int getSystemKey() { - return mSystemKey; + return mAgentImpl.getSystemKey(); } @@ -175,55 +72,14 @@ public class ActiveEntity extends AgentPOA * **************************************************************************/ @Override - public String queryData(String xpath) + public String queryData(String path) throws AccessRightsException, ObjectNotFoundException, PersistencyException { - String result = ""; - int allPos = -1; - - Logger.msg(1, "ActiveEntity::queryData("+mSystemKey+") - " + xpath ); - - try - { - if( (allPos=xpath.indexOf("all")) != -1 ) - { - String query = xpath.substring(0,allPos); - String[] ids = mDatabase.getClusterContents( mSystemKey, query ); - - for( int i=0; i, EntityProxyObserver> mSubscriptions; - - /************************************************************************** - * - **************************************************************************/ - protected EntityProxy( org.omg.CORBA.Object ior, - int systemKey) - throws ObjectNotFoundException - { - Logger.msg(8,"EntityProxy::EntityProxy() - Initialising '" +systemKey+ "' entity"); - - initialise( ior, systemKey); - } - - /************************************************************************** - * - **************************************************************************/ - private void initialise( org.omg.CORBA.Object ior, - int systemKey) - throws ObjectNotFoundException - { - Logger.msg(8, "EntityProxy::initialise() - Initialising '" +systemKey+ "' entity"); - - mIOR = ior; - mSystemKey = systemKey; - mSubscriptions = new HashMap, EntityProxyObserver>(); - } - - - /************************************************************************** - * - **************************************************************************/ - public ManageableEntity getEntity() throws ObjectNotFoundException - { - if (mEntity == null) { - mEntity = narrow(); - } - return mEntity; - } - - abstract public ManageableEntity narrow() throws ObjectNotFoundException; - - /************************************************************************** - * - **************************************************************************/ - //check who is using.. and if toString() is sufficient - @Override - public int getSystemKey() - { - return mSystemKey; - } - - - /************************************************************************** - * - **************************************************************************/ - @Override - public String queryData( String path ) - throws ObjectNotFoundException - { - - try { - Logger.msg(7, "EntityProxy.queryData() - "+mSystemKey+"/"+path); - if (path.endsWith("all")) { - Logger.msg(7, "EntityProxy.queryData() - listing contents"); - String[] result = Gateway.getStorage().getClusterContents(mSystemKey, path.substring(0, path.length()-3)); - StringBuffer retString = new StringBuffer(); - for (int i = 0; i < result.length; i++) { - retString.append(result[i]); - if (i"+e.getMessage()+""; - } - } - - public String[] getContents( String path ) throws ObjectNotFoundException { - try { - return Gateway.getStorage().getClusterContents(mSystemKey, path.substring(0, path.length())); - } catch (ClusterStorageException e) { - throw new ObjectNotFoundException(e.toString()); - } - } - - - /************************************************************************** - * - **************************************************************************/ - public C2KLocalObject getObject( String xpath ) - throws ObjectNotFoundException - { - // load from storage, falling back to proxy loader if not found in others - try - { - return Gateway.getStorage().get( mSystemKey, xpath , null); - } - catch( ClusterStorageException ex ) - { - Logger.msg(4, "Exception loading object :"+mSystemKey+"/"+xpath); - throw new ObjectNotFoundException( ex.toString() ); - } - } - - - - public String getProperty( String name ) - throws ObjectNotFoundException - { - Logger.msg(5, "Get property "+name+" from syskey/"+mSystemKey); - Property prop = (Property)getObject("Property/"+name); - try - { - return prop.getValue(); - } - catch (NullPointerException ex) - { - throw new ObjectNotFoundException(); - } - } - - public String getName() - { - try { - return getProperty("Name"); - } catch (ObjectNotFoundException ex) { - return null; - } - } - - - /************************************************************************** - * Subscription methods - **************************************************************************/ - - public void subscribe (MemberSubscription newSub) { - - newSub.setSubject(this); - synchronized (this){ - mSubscriptions.put( newSub, newSub.getObserver() ); - } - new Thread(newSub).start(); - Logger.msg(7, "Subscribed "+newSub.getObserver().getClass().getName()+" for "+newSub.interest); - } - - public void unsubscribe(EntityProxyObserver observer) - { - synchronized (this){ - for (Iterator> e = mSubscriptions.keySet().iterator(); e.hasNext();) { - MemberSubscription thisSub = e.next(); - if (mSubscriptions.get( thisSub ) == observer) { - e.remove(); - Logger.msg(7, "Unsubscribed "+observer.getClass().getName()); - } - } - } - } - - public void dumpSubscriptions(int logLevel) { - if (mSubscriptions.size() == 0) return; - Logger.msg(logLevel, "Subscriptions to proxy "+mSystemKey+":"); - synchronized(this) { - for (MemberSubscription element : mSubscriptions.keySet()) { - EntityProxyObserver obs = element.getObserver(); - if (obs != null) - Logger.msg(logLevel, " "+element.getObserver().getClass().getName()+" subscribed to "+element.interest); - else - Logger.msg(logLevel, " Phantom subscription to "+element.interest); - } - } - } - - public void notify(ProxyMessage message) { - Logger.msg(4, "EntityProxy.notify() - Received change notification for "+message.getPath()+" on "+mSystemKey); - synchronized (this){ - if (!message.getServer().equals(EntityProxyManager.serverName)) - Gateway.getStorage().clearCache(mSystemKey, message.getPath()); - for (Iterator> e = mSubscriptions.keySet().iterator(); e.hasNext();) { - MemberSubscription newSub = e.next(); - if (newSub.getObserver() == null) { // phantom - Logger.msg(4, "Removing phantom subscription to "+newSub.interest); - e.remove(); - } - else - newSub.update(message.getPath(), message.getState()); - } - } - } - - /** - * If this is reaped, clear out the cache for it too. - */ - @Override - protected void finalize() throws Throwable { - Logger.msg(7, "Proxy "+mSystemKey+" reaped"); - Gateway.getStorage().clearCache(mSystemKey, null); - Gateway.getProxyManager().removeProxy(mSystemKey); - super.finalize(); - } - -} diff --git a/src/main/java/com/c2kernel/entity/proxy/EntityProxyManager.java b/src/main/java/com/c2kernel/entity/proxy/EntityProxyManager.java deleted file mode 100644 index c49e7f5..0000000 --- a/src/main/java/com/c2kernel/entity/proxy/EntityProxyManager.java +++ /dev/null @@ -1,339 +0,0 @@ -/************************************************************************** - * EntityProxyFactory.java - * - * $Revision: 1.45 $ - * $Date: 2005/05/10 11:40:09 $ - * - * Copyright (C) 2001 CERN - European Organization for Nuclear Research - * All rights reserved. - **************************************************************************/ - -package com.c2kernel.entity.proxy; - -import java.util.ArrayList; -import java.util.ConcurrentModificationException; -import java.util.Enumeration; -import java.util.HashMap; -import java.util.Iterator; - -import com.c2kernel.common.InvalidDataException; -import com.c2kernel.common.ObjectNotFoundException; -import com.c2kernel.lookup.AgentPath; -import com.c2kernel.lookup.DomainPath; -import com.c2kernel.lookup.Path; -import com.c2kernel.persistency.ClusterStorage; -import com.c2kernel.process.Gateway; -import com.c2kernel.property.Property; -import com.c2kernel.utils.Logger; -import com.c2kernel.utils.SoftCache; -import com.c2kernel.utils.server.SimpleTCPIPServer; - - -public class EntityProxyManager -{ - SoftCache proxyPool = new SoftCache(50); - HashMap treeSubscribers = new HashMap(); - HashMap connections = new HashMap(); - - // server objects - static ArrayList proxyClients = new ArrayList(); - static SimpleTCPIPServer proxyServer = null; - static String serverName = null; - - /** - * Create an entity proxy manager to listen for proxy events and reap unused proxies - */ - public EntityProxyManager() - { - Logger.msg(5, "EntityProxyManager - Starting....."); - - Enumeration servers = Gateway.getLDAPLookup().searchEntities(new DomainPath("/servers")); - while(servers.hasMoreElements()) { - Path thisServerPath = servers.nextElement(); - try { - int syskey = thisServerPath.getSysKey(); - String remoteServer = ((Property)Gateway.getStorage().get(syskey, ClusterStorage.PROPERTY+"/Name", null)).getValue(); - String portStr = ((Property)Gateway.getStorage().get(syskey, ClusterStorage.PROPERTY+"/ProxyPort", null)).getValue(); - int remotePort = Integer.parseInt(portStr); - connectToProxyServer(remoteServer, remotePort); - - } catch (Exception ex) { - Logger.error("Exception retrieving proxy server connection data for "+thisServerPath); - Logger.error(ex); - } - } - } - - public void connectToProxyServer(String name, int port) { - ProxyServerConnection oldConn = connections.get(name); - if (oldConn != null) - oldConn.shutdown(); - connections.put(name, new ProxyServerConnection(name, port, this)); - } - - - protected void resubscribe(ProxyServerConnection conn) { - synchronized (proxyPool) { - for (Integer key : proxyPool.keySet()) { - ProxyMessage sub = new ProxyMessage(key.intValue(), ProxyMessage.ADDPATH, false); - Logger.msg(5, "Subscribing to entity "+key); - conn.sendMessage(sub); - } - } - } - - /** - * @param sub - */ - private void sendMessage(ProxyMessage sub) { - for (ProxyServerConnection element : connections.values()) { - element.sendMessage(sub); - } - - } - - public void shutdown() { - Logger.msg("EntityProxyManager.shutdown() - flagging shutdown of server connections"); - for (ProxyServerConnection element : connections.values()) { - element.shutdown(); - } - } - - protected void processMessage(ProxyMessage thisMessage) throws InvalidDataException { - if (Logger.doLog(9)) Logger.msg(9, thisMessage.toString()); - - if (thisMessage.getPath().equals(ProxyMessage.PINGPATH)) // ping response - return; - - if (thisMessage.getSysKey() == ProxyMessage.NA) // must be domain path info - informTreeSubscribers(thisMessage.getState(), thisMessage.getPath()); - else { - // proper proxy message - Logger.msg(5, "Received proxy message: "+thisMessage.toString()); - Integer key = new Integer(thisMessage.getSysKey()); - EntityProxy relevant = proxyPool.get(key); - if (relevant == null) - Logger.warning("Received proxy message for sysKey "+thisMessage.getSysKey()+" which we don't have a proxy for."); - else - try { - relevant.notify(thisMessage); - } catch (Throwable ex) { - Logger.error("Error caught notifying proxy listener "+relevant.toString()+" of "+thisMessage.toString()); - Logger.error(ex); - } - } - } - - private void informTreeSubscribers(boolean state, String path) { - DomainPath last = new DomainPath(path); - DomainPath parent; boolean first = true; - synchronized(treeSubscribers) { - while((parent = last.getParent()) != null) { - ArrayList currentKeys = new ArrayList(); - currentKeys.addAll(treeSubscribers.keySet()); - for (DomainPathSubscriber sub : currentKeys) { - DomainPath interest = treeSubscribers.get(sub); - if (interest!= null && interest.equals(parent)) { - if (state == ProxyMessage.ADDED) - sub.pathAdded(last); - else if (first) - sub.pathRemoved(last); - } - } - last = parent; - first = false; - } - } - } - - public void subscribeTree(DomainPathSubscriber sub, DomainPath interest) { - synchronized(treeSubscribers) { - treeSubscribers.put(sub, interest); - } - } - - public void unsubscribeTree(DomainPathSubscriber sub) { - synchronized(treeSubscribers) { - treeSubscribers.remove(sub); - } - } - - /************************************************************************** - * - **************************************************************************/ - private EntityProxy createProxy( org.omg.CORBA.Object ior, - int systemKey, - boolean isItem ) - throws ObjectNotFoundException - { - - EntityProxy newProxy = null; - - Logger.msg(5, "EntityProxyFactory::creating proxy on entity " + systemKey); - - if( isItem ) - { - newProxy = new ItemProxy(ior, systemKey); - } - else - { - newProxy = new AgentProxy(ior, systemKey); - } - - // subscribe to changes from server - ProxyMessage sub = new ProxyMessage(systemKey, ProxyMessage.ADDPATH, false); - sendMessage(sub); - reportCurrentProxies(9); - return ( newProxy ); - } - - protected void removeProxy( int systemKey ) - { - ProxyMessage sub = new ProxyMessage(systemKey, ProxyMessage.DELPATH, true); - Logger.msg(5,"EntityProxyManager.removeProxy() - Unsubscribing to proxy informer for "+systemKey); - sendMessage(sub); - } - - - /************************************************************************** - * EntityProxy getProxy( ManageableEntity, SystemKey) - * - * Called by the other GetProxy methods. Fills in either the ior or the - * SystemKey - **************************************************************************/ - private EntityProxy getProxy( org.omg.CORBA.Object ior, - int systemKey, - boolean isItem ) - throws ObjectNotFoundException - { - Integer key = new Integer(systemKey); - - synchronized(proxyPool) { - EntityProxy newProxy; - // return it if it exists - newProxy = proxyPool.get(key); - if (newProxy == null) { - // create a new one - newProxy = createProxy(ior, systemKey, isItem ); - proxyPool.put(key, newProxy); - } - return newProxy; - - } - } - - /************************************************************************** - * EntityProxy getProxy( String ) - * - * Proxy from Alias - **************************************************************************/ - public EntityProxy getProxy( Path path ) - throws ObjectNotFoundException - { - - //convert namePath to dn format - Logger.msg(8,"EntityProxyFactory::getProxy(" + path.toString() + ")"); - boolean isItem = !(path.getEntity() instanceof AgentPath); - return getProxy( Gateway.getLDAPLookup().getIOR(path), - path.getSysKey(), - isItem ); - - } - - /************************************************************************** - * void reportCurrentProxies() - * - * A utility to Dump the current proxies loaded - **************************************************************************/ - public void reportCurrentProxies(int logLevel) - { - if (!Logger.doLog(logLevel)) return; - Logger.msg(logLevel, "Current proxies: "); - try { - synchronized(proxyPool) { - Iterator i = proxyPool.keySet().iterator(); - - for( int count=0; i.hasNext(); count++ ) - { - Integer nextProxy = i.next(); - EntityProxy thisProxy = proxyPool.get(nextProxy); - if (thisProxy != null) - Logger.msg(logLevel, - "" + count + ": " - + proxyPool.get(nextProxy).getClass().getName() - + ": " + nextProxy); - } - } - } catch (ConcurrentModificationException ex) { - Logger.msg(logLevel, "Proxy cache modified. Aborting."); - } - } - - - /************************************************************************** - * Static Proxy Server methods - **************************************************************************/ - - /** - * Initialises the Proxy event UDP server listening on 'Host.Proxy.port' from c2kprops - * @param c2kProps - */ - public static void initServer() - { - Logger.msg(5, "EntityProxyFactory::initServer - Starting....."); - int port = Gateway.getProperties().getInt("ItemServer.Proxy.port", 0); - serverName = Gateway.getProperties().getProperty("ItemServer.name"); - if (port == 0) { - Logger.error("ItemServer.Proxy.port not defined in connect file. Remote proxies will not be informed of entity changes."); - return; - } - - // set up the proxy server - try { - Logger.msg(5, "EntityProxyFactory::initServer - Initialising proxy informer on port "+port); - proxyServer = new SimpleTCPIPServer(port, ProxyClientConnection.class, 200); - proxyServer.startListening(); - } catch (Exception ex) { - Logger.error("Error setting up Proxy Server. Remote proxies will not be informed of entity changes."); - Logger.error(ex); - } - } - - public static void sendProxyEvent(ProxyMessage message) { - if (proxyServer != null && message.getPath() != null) - synchronized(proxyClients) { - for (ProxyClientConnection client : proxyClients) { - client.sendMessage(message); - } - } - } - - public static void reportConnections(int logLevel) { - synchronized(proxyClients) { - Logger.msg(logLevel, "Currently connected proxy clients:"); - for (ProxyClientConnection client : proxyClients) { - Logger.msg(logLevel, " "+client); - } - } - } - - public static void shutdownServer() { - if (proxyServer != null) { - Logger.msg(1, "EntityProxyManager: Closing Server."); - proxyServer.stopListening(); - } - } - - public static void registerProxyClient(ProxyClientConnection client) { - synchronized(proxyClients) { - proxyClients.add(client); - } - } - - public static void unRegisterProxyClient(ProxyClientConnection client) { - synchronized(proxyClients) { - proxyClients.remove(client); - } - } -} - diff --git a/src/main/java/com/c2kernel/entity/proxy/EntityProxyObserver.java b/src/main/java/com/c2kernel/entity/proxy/EntityProxyObserver.java deleted file mode 100644 index 3ddb99c..0000000 --- a/src/main/java/com/c2kernel/entity/proxy/EntityProxyObserver.java +++ /dev/null @@ -1,27 +0,0 @@ -package com.c2kernel.entity.proxy; - -import com.c2kernel.entity.C2KLocalObject; - - - -public interface EntityProxyObserver -{ - /************************************************************************** - * Subscribed items are broken apart and fed one by one to these methods. - * Replacement after an event is done by feeding the new memberbase with the same id. - * ID could be an XPath? - **************************************************************************/ - public void add(V contents); - - /************************************************************************** - * the 'type' parameter should be an indication of the type of object - * supplied so that the subscriber can associate the call back with - * one of its subscriptions. If we go with an Xpath subscription form, - * then the id will probably be sufficient. - * Should be comparable (substring whatever) with the parameter given to - * the subscribe method of ItemProxy. - **************************************************************************/ - public void remove(String id); - - public void control(String control, String msg); -} diff --git a/src/main/java/com/c2kernel/entity/proxy/ItemProxy.java b/src/main/java/com/c2kernel/entity/proxy/ItemProxy.java index 0e6859d..355acd8 100644 --- a/src/main/java/com/c2kernel/entity/proxy/ItemProxy.java +++ b/src/main/java/com/c2kernel/entity/proxy/ItemProxy.java @@ -10,25 +10,40 @@ package com.c2kernel.entity.proxy; +import java.io.IOException; import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; + +import org.exolab.castor.mapping.MappingException; +import org.exolab.castor.xml.MarshalException; +import org.exolab.castor.xml.ValidationException; import com.c2kernel.collection.Collection; -import com.c2kernel.collection.CollectionMember; +import com.c2kernel.collection.CollectionArrayList; import com.c2kernel.common.AccessRightsException; import com.c2kernel.common.InvalidDataException; import com.c2kernel.common.InvalidTransitionException; import com.c2kernel.common.ObjectAlreadyExistsException; import com.c2kernel.common.ObjectNotFoundException; import com.c2kernel.common.PersistencyException; +import com.c2kernel.entity.C2KLocalObject; import com.c2kernel.entity.Item; import com.c2kernel.entity.ItemHelper; -import com.c2kernel.entity.ManageableEntity; import com.c2kernel.entity.agent.Job; import com.c2kernel.entity.agent.JobArrayList; +import com.c2kernel.lifecycle.instance.CompositeActivity; import com.c2kernel.lifecycle.instance.Workflow; +import com.c2kernel.lookup.InvalidItemPathException; +import com.c2kernel.lookup.ItemPath; +import com.c2kernel.lookup.Path; import com.c2kernel.persistency.ClusterStorage; +import com.c2kernel.persistency.ClusterStorageException; import com.c2kernel.persistency.outcome.Viewpoint; import com.c2kernel.process.Gateway; +import com.c2kernel.property.Property; +import com.c2kernel.property.PropertyArrayList; +import com.c2kernel.utils.CastorXMLUtility; import com.c2kernel.utils.Logger; /****************************************************************************** @@ -38,43 +53,86 @@ import com.c2kernel.utils.Logger; * @version $Revision: 1.25 $ $Date: 2005/05/10 11:40:09 $ * @author $Author: abranson $ ******************************************************************************/ -public class ItemProxy extends EntityProxy +public class ItemProxy { + protected Item mItem = null; + protected org.omg.CORBA.Object mIOR; + protected int mSystemKey; + protected Path mPath; + private final HashMap, ProxyObserver> + mSubscriptions; + /************************************************************************** - * + * **************************************************************************/ protected ItemProxy( org.omg.CORBA.Object ior, int systemKey) throws ObjectNotFoundException { - super(ior, systemKey); + Logger.msg(8, "ItemProxy::initialise() - Initialising entity " +systemKey); + + mIOR = ior; + mSystemKey = systemKey; + mSubscriptions = new HashMap, ProxyObserver>(); + try { + mPath = new ItemPath(systemKey); + } catch (InvalidItemPathException e) { + throw new ObjectNotFoundException(); + } } + + public int getSystemKey() + { + return mSystemKey; + } + + public Path getPath() { + return mPath; + } + + protected Item getItem() throws ObjectNotFoundException { + if (mItem == null) + mItem = narrow(); + return mItem; + } - @Override - public ManageableEntity narrow() throws ObjectNotFoundException + public Item narrow() throws ObjectNotFoundException { try { return ItemHelper.narrow(mIOR); } catch (org.omg.CORBA.BAD_PARAM ex) { } throw new ObjectNotFoundException("CORBA Object was not an Item, or the server is down."); } - /************************************************************************** + /** + * @throws MappingException + * @throws IOException + * @throws ValidationException + * @throws MarshalException ************************************************************************ * * **************************************************************************/ - public void initialise( int agentId, - String itemProps, - String workflow ) + public void initialise( int agentId, + PropertyArrayList itemProps, + CompositeActivity workflow, + CollectionArrayList colls + ) throws AccessRightsException, InvalidDataException, PersistencyException, - ObjectNotFoundException + ObjectNotFoundException, MarshalException, ValidationException, IOException, MappingException { Logger.msg(7, "ItemProxy::initialise - started"); - - ((Item)getEntity()).initialise( agentId, itemProps, workflow ); + CastorXMLUtility xml = Gateway.getMarshaller(); + if (itemProps == null) throw new InvalidDataException("No initial properties supplied"); + String propString = xml.marshall(itemProps); + String wfString = ""; + if (workflow != null) wfString = xml.marshall(workflow); + String collString = ""; + if (colls != null) collString = xml.marshall(colls); + + getItem().initialise( agentId, propString, wfString, collString); } public void setProperty(AgentProxy agent, String name, String value) @@ -100,7 +158,7 @@ public class ItemProxy extends EntityProxy /************************************************************************** * **************************************************************************/ - protected void requestAction( Job thisJob ) + public void requestAction( Job thisJob ) throws AccessRightsException, InvalidTransitionException, ObjectNotFoundException, @@ -120,44 +178,10 @@ public class ItemProxy extends EntityProxy throw new InvalidDataException("No Agent specified.", ""); Logger.msg(7, "ItemProxy - executing "+thisJob.getStepPath()+" for "+thisJob.getAgentName()); - requestAction (thisJob.getAgentId(), thisJob.getStepPath(), + getItem().requestAction (thisJob.getAgentId(), thisJob.getStepPath(), thisJob.getTransition().getId(), outcome); } - //requestData is xmlString - public void requestAction( int agentId, - String stepPath, - int transitionID, - String requestData - ) - throws AccessRightsException, - InvalidTransitionException, - ObjectNotFoundException, - InvalidDataException, - PersistencyException, - ObjectAlreadyExistsException - { - ((Item)getEntity()).requestAction( agentId, - stepPath, - transitionID, - requestData ); - } - - /************************************************************************** - * - **************************************************************************/ - public String queryLifeCycle( int agentId, - boolean filter - ) - throws AccessRightsException, - ObjectNotFoundException, - PersistencyException - { - return ((Item)getEntity()).queryLifeCycle( agentId, - filter ); - } - - /************************************************************************** * **************************************************************************/ @@ -168,7 +192,7 @@ public class ItemProxy extends EntityProxy { JobArrayList thisJobList; try { - String jobs = queryLifeCycle(agentId, filter); + String jobs = getItem().queryLifeCycle(agentId, filter); thisJobList = (JobArrayList)Gateway.getMarshaller().unmarshall(jobs); } catch (Exception e) { @@ -208,8 +232,8 @@ public class ItemProxy extends EntityProxy } - public Collection getCollection(String collName) throws ObjectNotFoundException { - return (Collection)getObject(ClusterStorage.COLLECTION+"/"+collName); + public Collection getCollection(String collName) throws ObjectNotFoundException { + return (Collection)getObject(ClusterStorage.COLLECTION+"/"+collName); } public Workflow getWorkflow() throws ObjectNotFoundException { @@ -226,4 +250,159 @@ public class ItemProxy extends EntityProxy PersistencyException { return getJobByName(actName, agent.getSystemKey()); } + + /** + * If this is reaped, clear out the cache for it too. + */ + @Override + protected void finalize() throws Throwable { + Logger.msg(7, "Proxy "+mSystemKey+" reaped"); + Gateway.getStorage().clearCache(mSystemKey, null); + Gateway.getProxyManager().removeProxy(mSystemKey); + super.finalize(); + } + + /************************************************************************** + * + **************************************************************************/ + public String queryData( String path ) + throws ObjectNotFoundException + { + + try { + Logger.msg(7, "EntityProxy.queryData() - "+mSystemKey+"/"+path); + if (path.endsWith("all")) { + Logger.msg(7, "EntityProxy.queryData() - listing contents"); + String[] result = Gateway.getStorage().getClusterContents(mSystemKey, path.substring(0, path.length()-3)); + StringBuffer retString = new StringBuffer(); + for (int i = 0; i < result.length; i++) { + retString.append(result[i]); + if (i"+e.getMessage()+""; + } + } + + public String[] getContents( String path ) throws ObjectNotFoundException { + try { + return Gateway.getStorage().getClusterContents(mSystemKey, path.substring(0, path.length())); + } catch (ClusterStorageException e) { + throw new ObjectNotFoundException(e.toString()); + } + } + + + /************************************************************************** + * + **************************************************************************/ + public C2KLocalObject getObject( String xpath ) + throws ObjectNotFoundException + { + // load from storage, falling back to proxy loader if not found in others + try + { + return Gateway.getStorage().get( mSystemKey, xpath , null); + } + catch( ClusterStorageException ex ) + { + Logger.msg(4, "Exception loading object :"+mSystemKey+"/"+xpath); + throw new ObjectNotFoundException( ex.toString() ); + } + } + + + + public String getProperty( String name ) + throws ObjectNotFoundException + { + Logger.msg(5, "Get property "+name+" from syskey/"+mSystemKey); + Property prop = (Property)getObject("Property/"+name); + try + { + return prop.getValue(); + } + catch (NullPointerException ex) + { + throw new ObjectNotFoundException(); + } + } + + public String getName() + { + try { + return getProperty("Name"); + } catch (ObjectNotFoundException ex) { + return null; + } + } + + + + + /************************************************************************** + * Subscription methods + **************************************************************************/ + + public void subscribe(MemberSubscription newSub) { + + newSub.setSubject(this); + synchronized (this){ + mSubscriptions.put( newSub, newSub.getObserver() ); + } + new Thread(newSub).start(); + Logger.msg(7, "Subscribed "+newSub.getObserver().getClass().getName()+" for "+newSub.interest); + } + + public void unsubscribe(ProxyObserver observer) + { + synchronized (this){ + for (Iterator> e = mSubscriptions.keySet().iterator(); e.hasNext();) { + MemberSubscription thisSub = e.next(); + if (mSubscriptions.get( thisSub ) == observer) { + e.remove(); + Logger.msg(7, "Unsubscribed "+observer.getClass().getName()); + } + } + } + } + + public void dumpSubscriptions(int logLevel) { + if (mSubscriptions.size() == 0) return; + Logger.msg(logLevel, "Subscriptions to proxy "+mSystemKey+":"); + synchronized(this) { + for (MemberSubscription element : mSubscriptions.keySet()) { + ProxyObserver obs = element.getObserver(); + if (obs != null) + Logger.msg(logLevel, " "+element.getObserver().getClass().getName()+" subscribed to "+element.interest); + else + Logger.msg(logLevel, " Phantom subscription to "+element.interest); + } + } + } + + public void notify(ProxyMessage message) { + Logger.msg(4, "EntityProxy.notify() - Received change notification for "+message.getPath()+" on "+mSystemKey); + synchronized (this){ + if (!message.getServer().equals(ProxyManager.serverName)) + Gateway.getStorage().clearCache(mSystemKey, message.getPath()); + for (Iterator> e = mSubscriptions.keySet().iterator(); e.hasNext();) { + MemberSubscription newSub = e.next(); + if (newSub.getObserver() == null) { // phantom + Logger.msg(4, "Removing phantom subscription to "+newSub.interest); + e.remove(); + } + else + newSub.update(message.getPath(), message.getState()); + } + } + } } diff --git a/src/main/java/com/c2kernel/entity/proxy/MemberSubscription.java b/src/main/java/com/c2kernel/entity/proxy/MemberSubscription.java index 1de18f8..01994e4 100644 --- a/src/main/java/com/c2kernel/entity/proxy/MemberSubscription.java +++ b/src/main/java/com/c2kernel/entity/proxy/MemberSubscription.java @@ -12,14 +12,14 @@ public class MemberSubscription implements Runnable { public static final String ERROR = "Error"; public static final String END = "theEND"; - EntityProxy subject; + ItemProxy subject; String interest; // keep the subscriber by weak reference, so it is not kept from the garbage collector if no longer used - WeakReference> observerReference; + WeakReference> observerReference; ArrayList contents = new ArrayList(); boolean preLoad; - public MemberSubscription(EntityProxyObserver observer, String interest, boolean preLoad) { + public MemberSubscription(ProxyObserver observer, String interest, boolean preLoad) { setObserver(observer); this.interest = interest; this.preLoad = preLoad; @@ -33,7 +33,7 @@ public class MemberSubscription implements Runnable { private void loadChildren() { C newMember; - EntityProxyObserver observer = getObserver(); + ProxyObserver observer = getObserver(); if (observer == null) return; //reaped try { // fetch contents of path @@ -77,7 +77,7 @@ public class MemberSubscription implements Runnable { } public void update(String path, boolean deleted) { - EntityProxyObserver observer = getObserver(); + ProxyObserver observer = getObserver(); if (observer == null) return; //reaped Logger.msg(7, "Processing proxy message path "+path +" for "+observer+". Interest: "+interest+" Was Deleted:"+deleted); if (!path.startsWith(interest)) // doesn't concern us @@ -106,15 +106,15 @@ public class MemberSubscription implements Runnable { } } - public void setObserver(EntityProxyObserver observer) { - observerReference = new WeakReference>(observer); + public void setObserver(ProxyObserver observer) { + observerReference = new WeakReference>(observer); } - public void setSubject(EntityProxy subject) { + public void setSubject(ItemProxy subject) { this.subject = subject; } - public EntityProxyObserver getObserver() { + public ProxyObserver getObserver() { return observerReference.get(); } } diff --git a/src/main/java/com/c2kernel/entity/proxy/ProxyClientConnection.java b/src/main/java/com/c2kernel/entity/proxy/ProxyClientConnection.java index 9687f22..5abdb16 100644 --- a/src/main/java/com/c2kernel/entity/proxy/ProxyClientConnection.java +++ b/src/main/java/com/c2kernel/entity/proxy/ProxyClientConnection.java @@ -36,7 +36,7 @@ public class ProxyClientConnection implements SocketHandler { public ProxyClientConnection() { super(); thisClientId = ++clientId; - EntityProxyManager.registerProxyClient(this); + ProxyManager.registerProxyClient(this); Logger.msg(1, "Proxy Client Connection Handler "+thisClientId+" ready."); } diff --git a/src/main/java/com/c2kernel/entity/proxy/ProxyManager.java b/src/main/java/com/c2kernel/entity/proxy/ProxyManager.java new file mode 100644 index 0000000..d19e38f --- /dev/null +++ b/src/main/java/com/c2kernel/entity/proxy/ProxyManager.java @@ -0,0 +1,337 @@ +/************************************************************************** + * ProxyManager.java + * + * $Revision: 1.45 $ + * $Date: 2005/05/10 11:40:09 $ + * + * Copyright (C) 2001 CERN - European Organization for Nuclear Research + * All rights reserved. + **************************************************************************/ + +package com.c2kernel.entity.proxy; + +import java.util.ArrayList; +import java.util.ConcurrentModificationException; +import java.util.Enumeration; +import java.util.HashMap; +import java.util.Iterator; + +import com.c2kernel.common.InvalidDataException; +import com.c2kernel.common.ObjectNotFoundException; +import com.c2kernel.lookup.AgentPath; +import com.c2kernel.lookup.DomainPath; +import com.c2kernel.lookup.Path; +import com.c2kernel.persistency.ClusterStorage; +import com.c2kernel.process.Gateway; +import com.c2kernel.property.Property; +import com.c2kernel.utils.Logger; +import com.c2kernel.utils.SoftCache; +import com.c2kernel.utils.server.SimpleTCPIPServer; + + +public class ProxyManager +{ + SoftCache proxyPool = new SoftCache(50); + HashMap treeSubscribers = new HashMap(); + HashMap connections = new HashMap(); + + // server objects + static ArrayList proxyClients = new ArrayList(); + static SimpleTCPIPServer proxyServer = null; + static String serverName = null; + + /** + * Create a proxy manager to listen for proxy events and reap unused proxies + */ + public ProxyManager() + { + Logger.msg(5, "ProxyManager - Starting....."); + + Enumeration servers = Gateway.getLDAPLookup().searchEntities(new DomainPath("/servers")); + while(servers.hasMoreElements()) { + Path thisServerPath = servers.nextElement(); + try { + int syskey = thisServerPath.getSysKey(); + String remoteServer = ((Property)Gateway.getStorage().get(syskey, ClusterStorage.PROPERTY+"/Name", null)).getValue(); + String portStr = ((Property)Gateway.getStorage().get(syskey, ClusterStorage.PROPERTY+"/ProxyPort", null)).getValue(); + int remotePort = Integer.parseInt(portStr); + connectToProxyServer(remoteServer, remotePort); + + } catch (Exception ex) { + Logger.error("Exception retrieving proxy server connection data for "+thisServerPath); + Logger.error(ex); + } + } + } + + public void connectToProxyServer(String name, int port) { + ProxyServerConnection oldConn = connections.get(name); + if (oldConn != null) + oldConn.shutdown(); + connections.put(name, new ProxyServerConnection(name, port, this)); + } + + + protected void resubscribe(ProxyServerConnection conn) { + synchronized (proxyPool) { + for (Integer key : proxyPool.keySet()) { + ProxyMessage sub = new ProxyMessage(key.intValue(), ProxyMessage.ADDPATH, false); + Logger.msg(5, "Subscribing to item "+key); + conn.sendMessage(sub); + } + } + } + + /** + * @param sub + */ + private void sendMessage(ProxyMessage sub) { + for (ProxyServerConnection element : connections.values()) { + element.sendMessage(sub); + } + + } + + public void shutdown() { + Logger.msg("ProxyManager.shutdown() - flagging shutdown of server connections"); + for (ProxyServerConnection element : connections.values()) { + element.shutdown(); + } + } + + protected void processMessage(ProxyMessage thisMessage) throws InvalidDataException { + if (Logger.doLog(9)) Logger.msg(9, thisMessage.toString()); + + if (thisMessage.getPath().equals(ProxyMessage.PINGPATH)) // ping response + return; + + if (thisMessage.getSysKey() == ProxyMessage.NA) // must be domain path info + informTreeSubscribers(thisMessage.getState(), thisMessage.getPath()); + else { + // proper proxy message + Logger.msg(5, "Received proxy message: "+thisMessage.toString()); + Integer key = new Integer(thisMessage.getSysKey()); + ItemProxy relevant = proxyPool.get(key); + if (relevant == null) + Logger.warning("Received proxy message for sysKey "+thisMessage.getSysKey()+" which we don't have a proxy for."); + else + try { + relevant.notify(thisMessage); + } catch (Throwable ex) { + Logger.error("Error caught notifying proxy listener "+relevant.toString()+" of "+thisMessage.toString()); + Logger.error(ex); + } + } + } + + private void informTreeSubscribers(boolean state, String path) { + DomainPath last = new DomainPath(path); + DomainPath parent; boolean first = true; + synchronized(treeSubscribers) { + while((parent = last.getParent()) != null) { + ArrayList currentKeys = new ArrayList(); + currentKeys.addAll(treeSubscribers.keySet()); + for (DomainPathSubscriber sub : currentKeys) { + DomainPath interest = treeSubscribers.get(sub); + if (interest!= null && interest.equals(parent)) { + if (state == ProxyMessage.ADDED) + sub.pathAdded(last); + else if (first) + sub.pathRemoved(last); + } + } + last = parent; + first = false; + } + } + } + + public void subscribeTree(DomainPathSubscriber sub, DomainPath interest) { + synchronized(treeSubscribers) { + treeSubscribers.put(sub, interest); + } + } + + public void unsubscribeTree(DomainPathSubscriber sub) { + synchronized(treeSubscribers) { + treeSubscribers.remove(sub); + } + } + + /************************************************************************** + * + **************************************************************************/ + private ItemProxy createProxy( org.omg.CORBA.Object ior, + int systemKey, + boolean isItem ) + throws ObjectNotFoundException + { + + ItemProxy newProxy = null; + + Logger.msg(5, "ProxyManager::creating proxy on Item " + systemKey); + + if( isItem ) + { + newProxy = new ItemProxy(ior, systemKey); + } + else + { + newProxy = new AgentProxy(ior, systemKey); + } + + // subscribe to changes from server + ProxyMessage sub = new ProxyMessage(systemKey, ProxyMessage.ADDPATH, false); + sendMessage(sub); + reportCurrentProxies(9); + return ( newProxy ); + } + + protected void removeProxy( int systemKey ) + { + ProxyMessage sub = new ProxyMessage(systemKey, ProxyMessage.DELPATH, true); + Logger.msg(5,"ProxyManager.removeProxy() - Unsubscribing to proxy informer for "+systemKey); + sendMessage(sub); + } + + + /************************************************************************** + * Called by the other GetProxy methods. Fills in either the ior or the + * SystemKey + **************************************************************************/ + private ItemProxy getProxy( org.omg.CORBA.Object ior, + int systemKey, + boolean isItem ) + throws ObjectNotFoundException + { + Integer key = new Integer(systemKey); + + synchronized(proxyPool) { + ItemProxy newProxy; + // return it if it exists + newProxy = proxyPool.get(key); + if (newProxy == null) { + // create a new one + newProxy = createProxy(ior, systemKey, isItem ); + proxyPool.put(key, newProxy); + } + return newProxy; + + } + } + + /************************************************************************** + * ItemProxy getProxy( String ) + * + * Proxy from Alias + **************************************************************************/ + public ItemProxy getProxy( Path path ) + throws ObjectNotFoundException + { + + //convert namePath to dn format + Logger.msg(8,"ProxyManager::getProxy(" + path.toString() + ")"); + boolean isItem = !(path.getEntity() instanceof AgentPath); + return getProxy( Gateway.getLDAPLookup().getIOR(path), + path.getSysKey(), + isItem ); + + } + + /************************************************************************** + * void reportCurrentProxies() + * + * A utility to Dump the current proxies loaded + **************************************************************************/ + public void reportCurrentProxies(int logLevel) + { + if (!Logger.doLog(logLevel)) return; + Logger.msg(logLevel, "Current proxies: "); + try { + synchronized(proxyPool) { + Iterator i = proxyPool.keySet().iterator(); + + for( int count=0; i.hasNext(); count++ ) + { + Integer nextProxy = i.next(); + ItemProxy thisProxy = proxyPool.get(nextProxy); + if (thisProxy != null) + Logger.msg(logLevel, + "" + count + ": " + + proxyPool.get(nextProxy).getClass().getName() + + ": " + nextProxy); + } + } + } catch (ConcurrentModificationException ex) { + Logger.msg(logLevel, "Proxy cache modified. Aborting."); + } + } + + + /************************************************************************** + * Static Proxy Server methods + **************************************************************************/ + + /** + * Initialises the Proxy event UDP server listening on 'Host.Proxy.port' from c2kprops + * @param c2kProps + */ + public static void initServer() + { + Logger.msg(5, "ProxyManager::initServer - Starting....."); + int port = Gateway.getProperties().getInt("ItemServer.Proxy.port", 0); + serverName = Gateway.getProperties().getProperty("ItemServer.name"); + if (port == 0) { + Logger.error("ItemServer.Proxy.port not defined in connect file. Remote proxies will not be informed of changes."); + return; + } + + // set up the proxy server + try { + Logger.msg(5, "ProxyManager::initServer - Initialising proxy informer on port "+port); + proxyServer = new SimpleTCPIPServer(port, ProxyClientConnection.class, 200); + proxyServer.startListening(); + } catch (Exception ex) { + Logger.error("Error setting up Proxy Server. Remote proxies will not be informed of changes."); + Logger.error(ex); + } + } + + public static void sendProxyEvent(ProxyMessage message) { + if (proxyServer != null && message.getPath() != null) + synchronized(proxyClients) { + for (ProxyClientConnection client : proxyClients) { + client.sendMessage(message); + } + } + } + + public static void reportConnections(int logLevel) { + synchronized(proxyClients) { + Logger.msg(logLevel, "Currently connected proxy clients:"); + for (ProxyClientConnection client : proxyClients) { + Logger.msg(logLevel, " "+client); + } + } + } + + public static void shutdownServer() { + if (proxyServer != null) { + Logger.msg(1, "ProxyManager: Closing Server."); + proxyServer.stopListening(); + } + } + + public static void registerProxyClient(ProxyClientConnection client) { + synchronized(proxyClients) { + proxyClients.add(client); + } + } + + public static void unRegisterProxyClient(ProxyClientConnection client) { + synchronized(proxyClients) { + proxyClients.remove(client); + } + } +} + diff --git a/src/main/java/com/c2kernel/entity/proxy/ProxyObserver.java b/src/main/java/com/c2kernel/entity/proxy/ProxyObserver.java new file mode 100644 index 0000000..b15a972 --- /dev/null +++ b/src/main/java/com/c2kernel/entity/proxy/ProxyObserver.java @@ -0,0 +1,27 @@ +package com.c2kernel.entity.proxy; + +import com.c2kernel.entity.C2KLocalObject; + + + +public interface ProxyObserver +{ + /************************************************************************** + * Subscribed items are broken apart and fed one by one to these methods. + * Replacement after an event is done by feeding the new memberbase with the same id. + * ID could be an XPath? + **************************************************************************/ + public void add(V contents); + + /************************************************************************** + * the 'type' parameter should be an indication of the type of object + * supplied so that the subscriber can associate the call back with + * one of its subscriptions. If we go with an Xpath subscription form, + * then the id will probably be sufficient. + * Should be comparable (substring whatever) with the parameter given to + * the subscribe method of ItemProxy. + **************************************************************************/ + public void remove(String id); + + public void control(String control, String msg); +} diff --git a/src/main/java/com/c2kernel/entity/proxy/ProxyServerConnection.java b/src/main/java/com/c2kernel/entity/proxy/ProxyServerConnection.java index 6807953..54ca787 100644 --- a/src/main/java/com/c2kernel/entity/proxy/ProxyServerConnection.java +++ b/src/main/java/com/c2kernel/entity/proxy/ProxyServerConnection.java @@ -29,7 +29,7 @@ public class ProxyServerConnection extends Thread String serverName; int serverPort; Socket serverConnection; - EntityProxyManager manager; + ProxyManager manager; // for talking to the proxy server PrintWriter serverStream; boolean listening = false; @@ -38,7 +38,7 @@ public class ProxyServerConnection extends Thread /** * Create an entity proxy manager to listen for proxy events and reap unused proxies */ - public ProxyServerConnection(String host, int port, EntityProxyManager manager) + public ProxyServerConnection(String host, int port, ProxyManager manager) { Logger.msg(5, "ProxyServerConnection - Initialising connection to "+host+":"+port); serverName = host; diff --git a/src/main/java/com/c2kernel/entity/transfer/TransferItem.java b/src/main/java/com/c2kernel/entity/transfer/TransferItem.java index 520063a..df81721 100644 --- a/src/main/java/com/c2kernel/entity/transfer/TransferItem.java +++ b/src/main/java/com/c2kernel/entity/transfer/TransferItem.java @@ -9,7 +9,7 @@ import com.c2kernel.entity.C2KLocalObject; import com.c2kernel.entity.TraceableEntity; import com.c2kernel.lifecycle.instance.Workflow; import com.c2kernel.lookup.DomainPath; -import com.c2kernel.lookup.EntityPath; +import com.c2kernel.lookup.ItemPath; import com.c2kernel.lookup.Path; import com.c2kernel.persistency.ClusterStorage; import com.c2kernel.persistency.outcome.Outcome; @@ -89,7 +89,7 @@ public class TransferItem { } // create item - EntityPath entityPath = new EntityPath(sysKey); + ItemPath entityPath = new ItemPath(sysKey); TraceableEntity newItem = (TraceableEntity)Gateway.getCorbaServer().createEntity(entityPath); Gateway.getLDAPLookup().add(entityPath); @@ -107,7 +107,10 @@ public class TransferItem { throw new Exception("No workflow found in import for "+sysKey); // init item - newItem.initialise(importAgentId, Gateway.getMarshaller().marshall(props), Gateway.getMarshaller().marshall(wf.search("workflow/domain"))); + newItem.initialise(importAgentId, + Gateway.getMarshaller().marshall(props), + Gateway.getMarshaller().marshall(wf.search("workflow/domain")), + null); // store objects importByType(ClusterStorage.COLLECTION, objects); diff --git a/src/main/java/com/c2kernel/entity/transfer/TransferSet.java b/src/main/java/com/c2kernel/entity/transfer/TransferSet.java index 7a3ba2e..a7d81b6 100644 --- a/src/main/java/com/c2kernel/entity/transfer/TransferSet.java +++ b/src/main/java/com/c2kernel/entity/transfer/TransferSet.java @@ -3,7 +3,7 @@ package com.c2kernel.entity.transfer; import java.io.File; import java.util.ArrayList; -import com.c2kernel.lookup.EntityPath; +import com.c2kernel.lookup.ItemPath; import com.c2kernel.lookup.NextKeyManager; import com.c2kernel.process.Gateway; import com.c2kernel.utils.FileStringUtility; @@ -85,7 +85,7 @@ public class TransferSet { try { // find the current last key NextKeyManager nextKeyMan = Gateway.getLDAPLookup().getNextKeyManager(); - EntityPath lastKey = nextKeyMan.getLastEntityPath(); + ItemPath lastKey = nextKeyMan.getLastEntityPath(); Logger.msg(1, "Last key imported was "+packageLastKey+". LDAP lastkey was "+lastKey.getSysKey()); -- cgit v1.2.3