001: package net.refractions.udig.catalog.ui.workflow;
002:
003: import java.io.IOException;
004: import java.lang.reflect.InvocationTargetException;
005: import java.util.HashMap;
006: import java.util.HashSet;
007: import java.util.LinkedList;
008: import java.util.Map;
009:
010: import net.refractions.udig.catalog.ui.CatalogUIPlugin;
011: import net.refractions.udig.catalog.ui.internal.Messages;
012: import net.refractions.udig.ui.PlatformGIS;
013:
014: import org.eclipse.core.runtime.IProgressMonitor;
015: import org.eclipse.core.runtime.NullProgressMonitor;
016: import org.eclipse.core.runtime.SubProgressMonitor;
017: import org.eclipse.jface.operation.IRunnableWithProgress;
018:
019: public class Workflow {
020:
021: /** set of primary states **/
022: State[] states;
023:
024: /** map of class to objects for states **/
025: Map<Class<State>, State> lookup;
026:
027: /** queue of primary states**/
028: LinkedList<State> queue;
029:
030: /** current state **/
031: State current;
032:
033: /** listeners **/
034: HashSet<Listener> listeners = new HashSet<Listener>();
035:
036: /** concurrent access lock **/
037: Lock lock = new Lock();
038:
039: /** flag to indicate wither the pipe is started/finished **/
040: boolean started = false;
041: boolean finished = false;
042:
043: /** context object, states use this object as a seed to perform work **/
044: Object context;
045:
046: /**
047: * Creates an empty workflow. When using this constructor the states of the
048: * workflow must be set before the workflow can be started.
049: *
050: */
051: public Workflow() {
052: //do nothing
053: }
054:
055: /**
056: * Creates a workflow from a set of workflow states.
057: *
058: * @param states The states of the workflow.
059: */
060: public Workflow(State[] states) {
061: setStates(states);
062: }
063:
064: /**
065: * Adds a listener to the workflow. The listener collection is a set which
066: * prevents duplicates. For this resason clients may call this method
067: * multiple times with the same object.
068: *
069: * @param l The listening object.
070: */
071: public void addListener(Listener l) {
072: listeners.add(l);
073: }
074:
075: public void removeListener(Listener l) {
076: listeners.remove(l);
077: }
078:
079: /**
080: * Returns an object representing a context for which the states can feed
081: * off of. The context is often provided via a workbench selection.
082: *
083: * @return The context object, or null if none has been set
084: */
085: public Object getContext() {
086: return context;
087: }
088:
089: /**
090: * Sets the object representing a context for which states can feed off of.
091: * The context is often provided via a workbench selection.
092: *
093: * @param context The context object to set.
094: */
095: public void setContext(Object context) {
096: this .context = context;
097: }
098:
099: /**
100: * Sets the primary set of states of the workflow.
101: *
102: * @param states An array of states.
103: */
104: @SuppressWarnings("unchecked")
105: public void setStates(State[] states) {
106: int i2 = 0;
107: if (states != null)
108: i2 = states.length;
109: State[] s = new State[i2];
110: if (states != null)
111: System.arraycopy(states, 0, s, 0, s.length);
112:
113: this .states = s;
114: queue = new LinkedList<State>();
115: lookup = new HashMap<Class<State>, State>();
116: for (int i = 0; i < s.length; i++) {
117: s[i].setWorkflow(this );
118: queue.addLast(s[i]);
119: lookup.put((Class<State>) s[i].getClass(), s[i]);
120: }
121: }
122:
123: /**
124: * @return the primary set of states of the workflow.
125: */
126: public State[] getStates() {
127: State[] s = new State[states.length];
128: System.arraycopy(states, 0, s, 0, s.length);
129: return s;
130: }
131:
132: /**
133: * Returns a state of a specific class.
134: *
135: * @param <T> The type of the state.
136: * @param c The class of the state.
137: *
138: * @return The state instance, or null if none exists.
139: */
140: public <T> T getState(Class<T> c) {
141: return c.cast(lookup.get(c));
142: }
143:
144: /**
145: * Starts the workflow by moving to the first state. This method must only
146: * be called once. This method executes asynchronously performing work in a
147: * seperate thread and does not block.
148: */
149: public void start() {
150:
151: IRunnableWithProgress runnable = new IRunnableWithProgress() {
152:
153: public void run(IProgressMonitor monitor)
154: throws InvocationTargetException,
155: InterruptedException {
156:
157: start(monitor);
158: }
159: };
160:
161: //synchronized (mutex) {
162: //lock.acquire();
163:
164: //try {
165: //assertNotStarted();
166: PlatformGIS.run(runnable);
167: // }
168: // catch(IllegalStateException e) {
169: // lock.release();
170: // throw new IllegalStateException(e);
171: // }
172: // catch(Throwable t) {
173: // lock.release();
174: // CatalogUIPlugin.log(t.getLocalizedMessage(),t);
175: // }
176: //}
177: }
178:
179: /**
180: * Starts the workflow by moving to the first state. This method must only
181: * be called once. This method executes synchronously performing work in
182: * the current thread and blocks.
183: */
184: public void start(IProgressMonitor monitor) {
185: //synchronized (mutex) {
186: //lock.steal();
187: lock.acquire();
188:
189: try {
190: //move to first state
191: current = queue.removeFirst();
192: current.setPrevious(null);
193: current.init(monitor);
194:
195: started = true;
196: dispatchStarted(current);
197: } catch (IOException e) {
198: throw new RuntimeException(e);
199: } finally {
200: lock.release();
201: }
202: //}
203: }
204:
205: /**
206: * Moves the workflow to the next state. This method executes asynchronously
207: * performing work in a seperate thread and does not block.
208: *
209: */
210: public void next() {
211:
212: IRunnableWithProgress runnable = new IRunnableWithProgress() {
213:
214: public void run(IProgressMonitor monitor)
215: throws InvocationTargetException,
216: InterruptedException {
217:
218: next(monitor);
219: }
220: };
221:
222: PlatformGIS.run(runnable);
223:
224: }
225:
226: /**
227: * Moves the workflow to the next state. This method executes synchronously
228: * performing work in the current thread and blocks.
229: *
230: * @return True if the state
231: */
232: @SuppressWarnings("unchecked")
233: public void next(IProgressMonitor monitor) {
234: IProgressMonitor monitor2 = monitor;
235: if (monitor2 == null)
236: monitor2 = new NullProgressMonitor();
237: //synchronized (mutex) {
238: //lock.steal();
239: lock.acquire();
240:
241: try {
242: assertStarted();
243: assertNotFinished();
244:
245: String name = getCurrentState().getName();
246: String string = name != null ? name
247: : Messages.Workflow_busy;
248: monitor2.beginTask(string, 20);
249: monitor2.setTaskName(string);
250:
251: if (queue == null) {
252: String msg = "No states"; //$NON-NLS-1$
253: throw new IllegalStateException(msg);
254: }
255:
256: //run it
257: boolean ok = false;
258: SubProgressMonitor subProgressMonitor = new SubProgressMonitor(
259: monitor2, 10);
260: try {
261: ok = current.run(subProgressMonitor)
262: && !monitor2.isCanceled();
263: } catch (Throwable t) {
264: CatalogUIPlugin.log(t.getLocalizedMessage(), t);
265: } finally {
266: subProgressMonitor.done();
267: }
268:
269: if (ok) {
270: //dispatch the event
271: dispatchPassed(current);
272:
273: //grab the next state, try pulling one from the current state
274: State next = current.next();
275: if (next == null) {
276: //try pulling from the queue
277: if (!queue.isEmpty())
278: next = queue.removeFirst();
279: } else {
280: //add to lookup tables
281: lookup.put((Class<State>) next.getClass(), next);
282: }
283:
284: if (next != null) {
285: //set the back pointer and call lifecyclmutexe events
286: next.setWorkflow(this );
287: next.setPrevious(current);
288: try {
289: subProgressMonitor = new SubProgressMonitor(
290: monitor2, 10);
291: next.init(subProgressMonitor);
292: } finally {
293: subProgressMonitor.done();
294: }
295: State prev = current;
296: current = next;
297:
298: dispatchForward(current, prev);
299: } else {
300: // no more states, we are finished
301: State last = current;
302: current = null;
303:
304: finished = true;
305: dispatchFinished(last);
306: }
307: } else {
308: //run did not succeed
309: dispatchFailed(current);
310: }
311: } catch (IOException e) {
312: throw new RuntimeException(e);
313: } finally {
314: lock.release();
315: }
316: //}
317: }
318:
319: /**
320: * Moves the workflow to the previous state. This method executes
321: * asynchronously performing work in a seperate thread and does not block.
322: */
323: public void previous() {
324:
325: IRunnableWithProgress runnable = new IRunnableWithProgress() {
326:
327: public void run(IProgressMonitor monitor)
328: throws InvocationTargetException,
329: InterruptedException {
330:
331: previous(monitor);
332: }
333: };
334:
335: //synchronized (mutex) {
336: lock.acquire();
337: try {
338: assertStarted();
339: assertNotFinished();
340: runnable.run(new NullProgressMonitor());
341: } catch (IllegalStateException e) {
342: //lock.release();
343: throw new IllegalStateException(e);
344: } catch (Throwable t) {
345: //lock.release();
346: CatalogUIPlugin.log(t.getLocalizedMessage(), t);
347: } finally {
348: lock.release();
349: }
350: //}
351: }
352:
353: /**
354: * Moves the workflow to the previous state. This method executes
355: * synchronously performing work in the current thread and blocks.
356: */
357: public void previous(IProgressMonitor monitor) {
358: //synchronized (mutex) {
359: lock.steal();
360:
361: try {
362: assertStarted();
363: assertNotFinished();
364:
365: if (current.getPreviousState() != null) {
366: // if this state is a "primary" state, place back in front of queue
367: if (isPrimaryState(current))
368: queue.addFirst(current);
369:
370: State next = current;
371: current = current.getPreviousState();
372:
373: //renitialize the state and dispatch the started event
374: current.init(monitor);
375: dispatchBackward(current, next);
376:
377: }
378: } catch (IOException e) {
379: throw new RuntimeException(e);
380: } finally {
381: lock.release();
382: }
383: //}
384: }
385:
386: /**
387: * @return True if the workflow has been started with a call to #start().
388: */
389: public boolean isStarted() {
390: return started;
391: }
392:
393: /**
394: * @return True if the workflow has been finished. The workflow is
395: * considered finished after the call to #next(), while in the final state.
396: */
397: public boolean isFinished() {
398: return finished;
399: }
400:
401: /**
402: * @return the current state of the workflow.
403: */
404: public State getCurrentState() {
405: return current;
406: }
407:
408: /**
409: * Determines if the workflow has more states. It is important to note
410: * that this method may not 100% accurate depending on the behaviour of
411: * states dynamically creating new states.
412: *
413: * @return True if there are more states, otherwise false.
414: */
415: public boolean hasMoreStates() {
416: //if the queue is not empty, we definitely have more states
417: if (!queue.isEmpty())
418: return true;
419:
420: //ask the current state
421: if (current != null)
422: return current.hasNext();
423:
424: return false;
425: }
426:
427: /**
428: * Runs the workflow from its current state. The workflow will continue
429: * to walk through the states while the state is finished.
430: *
431: * @param monitor A progress monitor.
432: *
433: * @return True if the pipe was able to run to completion, otherwise false.
434: */
435: public boolean run(IProgressMonitor monitor) {
436: Runner runner = new Runner(this );
437: return runner.run(monitor);
438: }
439:
440: /**
441: * Resets the workflow. This method may only be called if the workflow is
442: * in a finished state. Once reset the workflow lifecycle starts again with
443: * a call to @see DataPipeline#start().
444: */
445: public void reset() {
446: assertFinished();
447:
448: started = finished = false;
449: setStates(states);
450: }
451:
452: protected void assertStarted() {
453: if (!started) {
454: String msg = "Not started"; //$NON-NLS-1$
455: throw new IllegalStateException(msg);
456: }
457: }
458:
459: protected void assertNotStarted() {
460: if (started) {
461: String msg = "Already started"; //$NON-NLS-1$
462: throw new IllegalStateException(msg);
463: }
464: }
465:
466: protected void assertFinished() {
467: if (!finished) {
468: String msg = "Not finished"; //$NON-NLS-1$
469: throw new IllegalStateException(msg);
470: }
471: }
472:
473: protected void assertNotFinished() {
474: if (finished) {
475: String msg = "Already finished"; //$NON-NLS-1$
476: throw new IllegalStateException(msg);
477: }
478: }
479:
480: protected boolean isPrimaryState(State state) {
481: for (int i = 0; i < states.length; i++) {
482: if (states[i].equals(state))
483: return true;
484: }
485:
486: return false;
487: }
488:
489: protected void dispatchStarted(State start) {
490: try {
491: for (Listener l : listeners) {
492: l.started(start);
493: }
494: } catch (Throwable t) {
495: CatalogUIPlugin.log(t.getLocalizedMessage(), t);
496: }
497: }
498:
499: protected void dispatchForward(State current, State prev) {
500: try {
501: for (Listener l : listeners) {
502: l.forward(current, prev);
503: }
504: } catch (Throwable t) {
505: CatalogUIPlugin.log(t.getLocalizedMessage(), t);
506: }
507: }
508:
509: protected void dispatchBackward(State current, State next) {
510: try {
511: for (Listener l : listeners) {
512: l.backward(current, next);
513: }
514: } catch (Throwable t) {
515: CatalogUIPlugin.log(t.getLocalizedMessage(), t);
516: }
517: }
518:
519: protected void dispatchPassed(State state) {
520: try {
521: for (Listener l : listeners) {
522: l.statePassed(state);
523: }
524: } catch (Throwable t) {
525: CatalogUIPlugin.log(t.getLocalizedMessage(), t);
526: }
527: }
528:
529: protected void dispatchFailed(State state) {
530: for (Listener l : listeners) {
531: try {
532: l.stateFailed(state);
533: } catch (Throwable t) {
534: CatalogUIPlugin.log(t.getLocalizedMessage(), t);
535: }
536: }
537: }
538:
539: protected void dispatchFinished(State last) {
540: try {
541: for (Listener l : listeners) {
542: l.finished(last);
543: }
544: } catch (Throwable t) {
545: CatalogUIPlugin.log(t.getLocalizedMessage(), t);
546: }
547: }
548:
549: public static class Runner implements Listener {
550: Workflow pipe;
551: boolean stopped;
552:
553: Runner(Workflow pipe) {
554: this .pipe = pipe;
555: }
556:
557: public boolean run(IProgressMonitor monitor) {
558: try {
559: monitor.beginTask(Messages.Workflow_task_name,
560: IProgressMonitor.UNKNOWN);
561: stopped = false;
562: pipe.addListener(this );
563:
564: //first check if the pipe is already finished
565: if (pipe.isFinished())
566: return true;
567:
568: //may need to start
569: if (!pipe.isStarted()) {
570: pipe.start(monitor);
571: }
572:
573: while (!stopped && !pipe.isFinished()) {
574: pipe
575: .next(new SubProgressMonitor(
576: monitor,
577: 10,
578: SubProgressMonitor.PREPEND_MAIN_LABEL_TO_SUBTASK));
579: }
580:
581: pipe.removeListener(this );
582: return !stopped;
583: } finally {
584: monitor.done();
585: }
586: }
587:
588: public void forward(State current, State prev) {
589: // do nothing
590: }
591:
592: public void backward(State current, State next) {
593: // do nothing
594: }
595:
596: public void statePassed(State state) {
597: // do nothing
598: }
599:
600: public void stateFailed(State state) {
601: stopped = true;
602: }
603:
604: public void started(State first) {
605: // do nothing
606: }
607:
608: public void finished(State last) {
609: // do nothing
610: }
611: }
612:
613: public static abstract class State {
614:
615: /** previous state **/
616: State previous;
617:
618: /** the workflow **/
619: Workflow workflow;
620:
621: /**
622: * @param workflow The workflow containing the state.
623: */
624: public void setWorkflow(Workflow workflow) {
625: this .workflow = workflow;
626: }
627:
628: public abstract String getName();
629:
630: /**
631: * @return the worklow containing all the states.
632: */
633: public Workflow getWorkflow() {
634: return workflow;
635: }
636:
637: /**
638: * Sets the previous state. The first method in the lifecycle of the
639: * state which is called by the data workflow to track the states that
640: * have been completed. Should not be called by client code.
641: *
642: * @param previous The previous state.
643: */
644: public void setPrevious(State previous) {
645: this .previous = previous;
646: }
647:
648: /**
649: * Returns the previous state.
650: *
651: * @return The state previous to this state, or null if no such state
652: * exisits.
653: */
654: public State getPreviousState() {
655: return previous;
656: }
657:
658: /**
659: * Initialize the state. This is the second method in the lifecycle of
660: * the state. It is called after #setPrevious(). If the state needs to
661: * "seed" itself with any context, that should occur here.
662: */
663: public void init(IProgressMonitor monitor) throws IOException {
664: //do nothing
665: }
666:
667: /**
668: * Performs any "hard" work. This method is provided is provided for
669: * states which have to block to get work done. For instance, making
670: * a connection to a remote service. This method returns a boolean
671: * which signals wether the state was able to get the work done.
672: *
673: * @param monitor A progress monitor.
674: *
675: * @return True if the state was able to complete its job, otherwise
676: * false.
677: * @throws IOException
678: */
679: public boolean run(IProgressMonitor monitor) throws IOException {
680:
681: return true;
682: }
683:
684: /**
685: * Determines if the state can dynamically create a new state to be
686: * the next active state of the workflow. Note, in most cases this is
687: * equivalent to <code>next() != null</code>. However some
688: * implemtnations require that next() be called only once, as it is a
689: * lifecycle event.
690: *
691: * @return true if the state can create a new state, otherwise false.
692: */
693: public boolean hasNext() {
694: return false;
695: }
696:
697: /**
698: * The final method in the lifecycle of the state. This method is
699: * used for states to dynamically link to each other. This method
700: * returns null to indicate no state.
701: *
702: * @return A new state which is to become the next active state,
703: * otherwise null.
704: */
705: public State next() {
706: return null;
707: }
708: }
709:
710: public static interface Listener {
711:
712: /**
713: * Event thrown when the pipe moves to a new state in the forward
714: * direction.
715: *
716: * @param current The current state.
717: * @param prev The state before the current state.
718: */
719: void forward(State current, State prev);
720:
721: /**
722: * Event thrown when the pipe moves to a new state in a backward
723: * direction.
724: *
725: * @param current The curent state.
726: * @param next The state after the current state.
727: */
728: void backward(State current, State next);
729:
730: /**
731: * Event thrown when a state successfully completes its job.
732: *
733: * @param state The current state.
734: */
735: void statePassed(State state);
736:
737: /**
738: * Event thrown when a state can not complete its job.
739: *
740: * @param state The current state.
741: */
742: void stateFailed(State state);
743:
744: /**
745: * Event thrown when the workflow is started.
746: *
747: * @param first The first state of the pipe
748: */
749: void started(State first);
750:
751: /**
752: * Event thrown when workflow is finished.
753: *
754: * @param last The last state of the pipe
755: */
756: void finished(State last);
757: }
758:
759: private static class Lock {
760: boolean locked = false;
761: Thread owner = null;
762:
763: public void acquire() {
764: while (true) {
765: synchronized (this ) {
766: if (!locked) {
767: //grab the lock
768: locked = true;
769: owner = Thread.currentThread();
770: return;
771: }
772: }
773: try {
774: Thread.sleep(100);
775: } catch (InterruptedException e) {
776: throw new IllegalStateException(e);
777: }
778: }
779: }
780:
781: public synchronized void release() {
782: if (owner != Thread.currentThread())
783: throw new IllegalStateException("Not owner"); //$NON-NLS-1$
784: owner = null;
785: locked = false;
786: }
787:
788: public synchronized void steal() {
789: owner = Thread.currentThread();
790: locked = true;
791: }
792: }
793: }
|