activityguard.java

bpel执行引擎用来执行bpel业务流程

    private static ACTIVITY createActivity(ActivityInfo activity, ScopeFrame scopeFrame, LinkFrame linkFrame) {
        return __activityTemplateFactory.createInstance(activity.o,activity, scopeFrame, linkFrame);
    }

    private ACTIVITY createActivity(ActivityInfo activity) {
        return createActivity(activity,_scopeFrame, _linkFrame);
    }

    private void startGuardedActivity() {
        ActivityInfo activity = new ActivityInfo(genMonotonic(),_self.o,_self.self, newChannel(ParentScopeChannel.class));
        instance(createActivity(activity));
        instance(new TCONDINTERCEPT(activity.parent));
    }


    /**
     * Intercepts the
     * {@link ParentScopeChannel#completed(org.apache.ode.bpel.runtime.channels.FaultData, java.util.Set<org.apache.ode.bpel.runtime.CompensationHandler>)}
     * call, to evaluate transition conditions before returning to the parent.
     */
    private class TCONDINTERCEPT extends BpelJacobRunnable {
        private static final long serialVersionUID = 4014873396828400441L;
        ParentScopeChannel _in;

        public TCONDINTERCEPT(ParentScopeChannel in) {
            _in = in;
        }

        public void run() {
            object(new ParentScopeChannelListener(_in) {
                private static final long serialVersionUID = 2667359535900385952L;

                public void compensate(OScope scope, SynchChannel ret) {
                    _self.parent.compensate(scope,ret);
                    instance(TCONDINTERCEPT.this);
                }

                public void completed(FaultData faultData, Set<CompensationHandler> compensations) {
                    sendEvent(new ActivityExecEndEvent());
                    if (faultData != null) {
                        dpe(_oactivity.sourceLinks);
                        _self.parent.completed(faultData, compensations);
                    } else {
                        FaultData fault = null;
                        for (Iterator<OLink> i = _oactivity.sourceLinks.iterator();i.hasNext();) {
                            OLink olink = i.next();
                            LinkInfo linfo = _linkFrame.resolve(olink);
                            try {
                                boolean val = evaluateTransitionCondition(olink.transitionCondition);
                                linfo.pub.linkStatus(val);
                            } catch (FaultException e) {
                                linfo.pub.linkStatus(false);
                                __log.error(e);
                                if (fault == null)
                                    fault = createFault(e.getQName(),olink.transitionCondition);
                            }
                        }
                        _self.parent.completed(fault, compensations);
                    }
                }

                public void cancelled() {
                    sendEvent(new ActivityExecEndEvent());
                    dpe(_oactivity.outgoingLinks);
                    dpe(_oactivity.sourceLinks);
                    // Implicit scope can tell the difference between cancelled and completed.
                    _self.parent.cancelled();
                }

                public void failure(String reason, Element data) {
                    if (_failure == null)
                        _failure = new ActivityFailure();
                    _failure.dateTime = new Date();
                    _failure.reason = reason;
                    _failure.data = data;

                    OFailureHandling failureHandling = _oactivity.getFailureHandling();
                    if (failureHandling != null && failureHandling.faultOnFailure) {
                      // No attempt to retry or enter activity recovery state, simply fault.
                        if (__log.isDebugEnabled())
                            __log.debug("ActivityRecovery: Activity " + _self.aId + " faulting on failure");
                        FaultData faultData = createFault(OFailureHandling.FAILURE_FAULT_NAME, _oactivity, reason);
                        completed(faultData, CompensationHandler.emptySet());
                        return;
                    }
                    if (failureHandling == null || _failure.retryCount >= failureHandling.retryFor) {
                        requireRecovery();
                        return;
                    }
        
                    if (__log.isDebugEnabled())
                        __log.debug("ActivityRecovery: Retrying activity " + _self.aId);
                    Date future = new Date(new Date().getTime() + 
                        (failureHandling == null ? 0L : failureHandling.retryDelay * 1000));
                    final TimerResponseChannel timerChannel = newChannel(TimerResponseChannel.class);
                    getBpelRuntimeContext().registerTimer(timerChannel, future);
                    object(false, new TimerResponseChannelListener(timerChannel) {
                        private static final long serialVersionUID = -261911108068231376L;
                            public void onTimeout() {
                                ++_failure.retryCount;
                                startGuardedActivity();
                            }
                            public void onCancel() {
                                requireRecovery();
                            }
                    });
                }

                private void requireRecovery() {
                    if (__log.isDebugEnabled())
                        __log.debug("ActivityRecovery: Activity " + _self.aId + " requires recovery");
                    sendEvent(new ActivityFailureEvent(_failure.reason));
                    final ActivityRecoveryChannel recoveryChannel = newChannel(ActivityRecoveryChannel.class);
                    getBpelRuntimeContext().registerActivityForRecovery(
                        recoveryChannel, _self.aId, _failure.reason, _failure.dateTime, _failure.data,
                        new String[] { "retry", "cancel", "fault" }, _failure.retryCount);
                    object(false, new ActivityRecoveryChannelListener(recoveryChannel) {
                        private static final long serialVersionUID = 8397883882810521685L;
                        public void retry() {
                            if (__log.isDebugEnabled())
                                __log.debug("ActivityRecovery: Retrying activity " + _self.aId + " (user initiated)");
                            sendEvent(new ActivityRecoveryEvent("retry"));
                            getBpelRuntimeContext().unregisterActivityForRecovery(recoveryChannel);
                            ++_failure.retryCount;
                            startGuardedActivity();
                        }
                        public void cancel() {
                            if (__log.isDebugEnabled())
                                __log.debug("ActivityRecovery: Cancelling activity " + _self.aId + " (user initiated)");
                            sendEvent(new ActivityRecoveryEvent("cancel"));
                            getBpelRuntimeContext().unregisterActivityForRecovery(recoveryChannel);
                            cancelled();
                        }
                        public void fault(FaultData faultData) {
                            if (__log.isDebugEnabled())
                                __log.debug("ActivityRecovery: Faulting activity " + _self.aId + " (user initiated)");
                            sendEvent(new ActivityRecoveryEvent("fault"));
                            getBpelRuntimeContext().unregisterActivityForRecovery(recoveryChannel);
                            if (faultData == null)
                                faultData = createFault(OFailureHandling.FAILURE_FAULT_NAME, _self.o, _failure.reason);
                            completed(faultData, CompensationHandler.emptySet());
                        }
                    }.or(new TerminationChannelListener(_self.self) {
                        private static final long serialVersionUID = 2148587381204858397L;

                        public void terminate() {
                            if (__log.isDebugEnabled())
                                __log.debug("ActivityRecovery: Cancelling activity " + _self.aId + " (terminated by scope)");
                            getBpelRuntimeContext().unregisterActivityForRecovery(recoveryChannel);
                            cancelled();
                        }
                    }));
                }
            });

        }
    }

    static class ActivityFailure implements Serializable {
        private static final long serialVersionUID = 1L;

        Date    dateTime;
        String  reason;
        Element data;
        int     retryCount;
    }
    
}