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

1 files changed, 232 insertions, 53 deletions

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<MemberSubscription<?>, 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<MemberSubscription<?>, 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<? extends CollectionMember> getCollection(String collName) throws ObjectNotFoundException {

-    	return (Collection<? extends CollectionMember>)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<result.length-1) retString.append(",");

+				}

+				Logger.msg(7, "EntityProxy.queryData() - "+retString.toString());

+				return retString.toString();

+			}

+			C2KLocalObject target = Gateway.getStorage().get(mSystemKey, path, null);

+			return Gateway.getMarshaller().marshall(target);

+        } catch (ObjectNotFoundException e) {

+            throw e;

+		} catch (Exception e) {

+			Logger.error(e);

+			return "<ERROR>"+e.getMessage()+"</ERROR>";

+		}

+    }

+

+    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<MemberSubscription<?>> 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<MemberSubscription<?>> 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());

+            }

+        }

+    }

 }