threadedbehaviourfactory.java

java实现的P2P多agent中间件

		 getParent() method.
		 */
		protected void setParent(CompositeBehaviour parent) {
			super.setParent(parent);
			myBehaviour.setThreadedParent(parent);
		}
		
		public void setDataStore(DataStore ds) {
			myBehaviour.setDataStore(ds);
		}
		
		public DataStore getDataStore() {
			return myBehaviour.getDataStore();
		}
		
		public void reset() {
			restarted = false;
			finished = false;
			suspended = false;
			myBehaviour.reset();
			super.reset();
		}
		
		/**
		 Propagate a restart() call (typically this happens when this 
		 ThreadedBehaviourWrapped is directly added to the agent Scheduler
		 and a message is received) to the wrapped threaded behaviour.
		 */
		public void restart() {
			myBehaviour.restart();
		}
		
		/**
		 Propagate a DOWNWARDS event (typically this happens when this
		 ThreadedBehaviourWrapper is added as a child of a CompositeBehaviour
		 and the latter, or an ancestor, is blocked/restarted)
		 to the wrapped threaded behaviour.
		 If the event is a restart, also notify the dedicated thread.
		 */
		protected void handle(RunnableChangedEvent rce) {
			super.handle(rce);
			if (!rce.isUpwards()) {
				myBehaviour.handle(rce);
				if (rce.isRunnable()) {
					go();
				}
			}
		}
		
		private synchronized void go() {
			restarted = true;
			notifyAll();
		}
		
		// Only the dedicated thread can suspend a threaded behaviour
		private synchronized void suspend() {
			if (Thread.currentThread() == myThread) {
				suspended = true;
			}
		}
		
		private synchronized void resume() {
			if (suspended) {
				suspended = false;
				if (myThread == null) {
					start();
				}
			}
		}
		
		public void run() {
			if (threadState == CREATED_STATE) {
				threadedBehaviours.addElement(this);
			}
			else if (threadState == SUSPENDED_STATE) {
				invokeMethod(myBehaviour, "handleResumed");
			}
			threadState = RUNNING_STATE;
			try {
				while (true) {
					restarted = false;
					myBehaviour.actionWrapper();
					
					synchronized (this) {
						// If the behaviour was restarted from outside during the action()
						// method, give it another chance
						if (restarted && (!myBehaviour.isRunnable())) {
							// We can't just set the runnable state of myBehaviour to true since, if myBehaviour
							// is a CompositeBehaviour, we may end up with myBehaviour runnable, but some of its children not runnable. 
							// However we can't call myBehaviour.restart() here because there could be a deadlock between a thread
							// posting a message and the current thread (monitors are this and the agent scheduler)
							myBehaviour.myEvent.init(true, Behaviour.NOTIFY_DOWN);
							myBehaviour.handle(myBehaviour.myEvent);
						}
						
						if (myBehaviour.done()) {
							break;
						}
						else {
							// If we were interrupted, avoid doing anything else and terminate
							if (Thread.currentThread().isInterrupted() || threadState == INTERRUPTED_STATE) {
								throw new InterruptedException();
							}
							// If the Behaviour suspended itself during the action() method --> Release the embedded Thread
							if (suspended) {
								threadState = SUSPENDED_STATE;
								myThread = null;
								// If the Behaviour defined a handleSuspended() method invoke it from within the terminating thread 
								// to give it a chance to clean up any allocated resources
								invokeMethod(myBehaviour, "handleSuspended");
								return;
							}
							if (!myBehaviour.isRunnable()) {
								threadState = BLOCKED_STATE;
								wait();
							}
						}
					}
					threadState = RUNNING_STATE;
				}
				exitValue = myBehaviour.onEnd();
				threadState = TERMINATED_STATE;
			}
			catch (InterruptedException ie) {
				threadState = INTERRUPTED_STATE;
				System.out.println("Threaded behaviour "+myBehaviour.getBehaviourName()+" interrupted before termination");
			}
			catch (Agent.Interrupted ae) {
				threadState = INTERRUPTED_STATE;
				System.out.println("Threaded behaviour "+myBehaviour.getBehaviourName()+" interrupted before termination");
			}
			catch (ThreadDeath td) {
				threadState = INTERRUPTED_STATE;
				System.out.println("Threaded behaviour "+myBehaviour.getBehaviourName()+" stopped before termination");
				// ThreadDeath errors should always be propagated so that the top level handler can perform the necessary clean up
				terminate();
				throw td;
			}
			catch (Throwable t) {
				threadState = ERROR_STATE;
				t.printStackTrace();
			}
			terminate();
		}
		
		/**
		 * Interrupt a threaded behaviour. This method should be used instead of 
		 * getThread().interrupt() because the latter may have no effect if called just after 
		 * the threaded behaviour suspended itself
		 */
		private synchronized Thread interrupt() {
			if (myThread != null) {
				threadState = INTERRUPTED_STATE;
				myThread.interrupt();
				return myThread;
			}
			else {
				if (threadState == SUSPENDED_STATE) {
					threadState = INTERRUPTED_STATE;
					terminate();
				}
				return null;
			}
		}

		private void terminate() {
			if (Thread.currentThread() == myThread) {
				if (threadState == INTERRUPTED_STATE || threadState == ERROR_STATE) {
					// If the Behaviour defined a handleAborted() method invoke it from within the terminating thread 
					// to give it a chance to clean up any allocated resources
					invokeMethod(myBehaviour, "handleAborted");
				}
			}
			finished = true;
			// Restart the wrapper so that it terminates too and is removed from the Agent scheduler
			super.restart();
			threadedBehaviours.removeElement(this);
			synchronized(ThreadedBehaviourFactory.this) {
				ThreadedBehaviourFactory.this.notifyAll();
			}
		}
		
		public final Thread getThread() {
			return myThread;
		}
		
		public final Behaviour getBehaviour() {
			return myBehaviour;
		}
		
		public final String getThreadState() {
			return threadState;
		}
	} // END of inner class ThreadedBehaviourWrapper
	//#APIDOC_EXCLUDE_END
	
	private void invokeMethod(Object obj, String methodName) {
		// FIXME: To be implemented 
	}
	
	/**
	 Inner class DummyParentBehaviour.
	 This class has the only purpose of propagating restart events
	 in the actual wrapped behaviour to the ThreadedBehaviourWrapper.
	 */
	private class DummyParentBehaviour extends CompositeBehaviour {
		private ThreadedBehaviourWrapper myChild;
		
		private DummyParentBehaviour(Agent a, ThreadedBehaviourWrapper b) {
			super(a);
			myChild = b;
		}
		
		public boolean isRunnable() {
			return false;
		}
		
		protected void handle(RunnableChangedEvent rce) {
			// This is always an UPWARDS event from the threaded behaviour, but
			// there is no need to propagate it to the wrapper since it will 
			// immediately block again. It would be just a waste of time.
			if (rce.isRunnable()) {
				myChild.go();
			}
		}
		
		/**
		 * Redefine the root() method so that both the DummyParentBehaviour
		 * and the ThreadedBehaviourWrapper are invisible in the behaviours hierarchy
		 */
		public Behaviour root() {
			Behaviour r = myChild.root();
			if (r == myChild) {
				return myChild.getBehaviour();
			}
			else {
				return r;
			}
		}
		
		protected void scheduleFirst() {
		}
		
		protected void scheduleNext(boolean currentDone, int currentResult) {
		}
		
		protected boolean checkTermination(boolean currentDone, int currentResult) {
			return false;
		}
		
		protected Behaviour getCurrent() {
			return null;
		}
		
		public jade.util.leap.Collection getChildren() {
			return null;
		}
	} // END of inner class DummyParentBehaviour
}