jacobvpu.java

来自「bpel执行引擎用来执行bpel业务流程」· Java 代码 · 共 493 行 · 第 1/2 页

        if (list == null) {
            return "";
        }
        StringBuffer buf = new StringBuffer();
        for (int i = 0; i < list.length; ++i) {
            if (i > 0) {
                buf.append(',');
            }
            buf.append(list[i]);
        }
        return buf.toString();
    }

    public void setClassLoader(ClassLoader classLoader) {
        _classLoader = classLoader;
        if (_executionQueue != null) {
            _executionQueue.setClassLoader(classLoader);
        }
    }

    /**
     * Dump the state of the VPU for debugging purposes.
     */
    public void dumpState() {
        _statistics.printToStream(System.err);
        _executionQueue.dumpState(System.err);
    }

    public boolean isComplete() {
        return _executionQueue.isComplete();
    }

    private class JacobThreadImpl implements Runnable, JacobThread {
        private final JacobObject _methodBody;

        private final Object[] _args;

        private final Method _method;

        /** Text string identifying the left side of the reduction (for debug). */
        private String _source;

        /** Text string identifying the target class and method (for debug) . */
        private String _targetStr = "Unknown";

        JacobThreadImpl(Continuation rqe) {
            assert rqe != null;

            _methodBody = rqe.getClosure();
            _args = rqe.getArgs();
            _source = rqe.getDescription();
            _method = rqe.getMethod();

            if (__log.isDebugEnabled()) {
                StringBuffer buf = new StringBuffer(_methodBody.getClass().getName());
                buf.append('.');
                buf.append(rqe.getMethod());
                _targetStr = buf.toString();
            }
        }

        public void instance(JacobRunnable template) {
            String desc = null;
            if (__log.isTraceEnabled()) {
                __log.trace(_cycle + ": " + template);
                desc = template.toString();
            }
            _statistics.numReductionsStruct++;
            addReaction(template, REDUCE_METHOD, CollectionUtils.EMPTY_OBJECT_ARRAY, desc);
        }

        public Channel message(Channel channel, Method method, Object[] args) {
            if (__log.isTraceEnabled()) {
                __log.trace(_cycle + ": " + channel + " ! "
                        + method.getName() + "(" + stringify(args) + ")");
            }
            _statistics.messagesSent++;

            SynchChannel replyChannel = null;
            // Check for synchronous methods; create a synchronization channel
            if (method.getReturnType() != void.class) {
                if (method.getReturnType() != SynchChannel.class) {
                    throw new IllegalStateException(
                            "ChannelListener method can only return SynchChannel: " + method);
                }
                replyChannel = (SynchChannel) newChannel(SynchChannel.class, "", "Reply Channel");
                Object[] newArgs = new Object[args.length + 1];
                System.arraycopy(args, 0, newArgs, 0, args.length);
                newArgs[args.length] = replyChannel;
                args = newArgs;
            }
            CommChannel chnl = (CommChannel) ChannelFactory.getBackend(channel);
            CommGroup grp = new CommGroup(false);
            CommSend send = new CommSend(chnl, method, args);
            grp.add(send);
            _executionQueue.add(grp);
            return replyChannel;
        }

        public Channel newChannel(Class channelType, String creator, String description) {
            CommChannel chnl = new CommChannel(channelType);
            chnl.setDescription(description);
            _executionQueue.add(chnl);

            Channel ret = ChannelFactory.createChannel(chnl, channelType);
            if (__log.isTraceEnabled())
                __log.trace(_cycle + ": new " + ret);

            _statistics.channelsCreated++;
            return ret;
        }

        public String exportChannel(Channel channel) {
            if (__log.isTraceEnabled()) {
                __log.trace(_cycle + ": export<" + channel + ">");
            }
            CommChannel chnl = (CommChannel) ChannelFactory.getBackend(channel);
            return _executionQueue.createExport(chnl);
        }

        public Channel importChannel(String channelId, Class channelType) {
            CommChannel cframe = _executionQueue.consumeExport(channelId);
            return ChannelFactory.createChannel(cframe, channelType);
        }

        public void object(boolean replicate, ChannelListener[] ml) {
            if (__log.isTraceEnabled()) {
                StringBuffer msg = new StringBuffer();
                msg.append(_cycle);
                msg.append(": ");
                for (int i = 0; i < ml.length; ++i) {
                    if (i != 0) msg.append(" + ");
                    msg.append(ml[i].getChannel());
                    msg.append(" ? ");
                    msg.append(ml.toString());

                }
                __log.debug(msg.toString());
            }

            _statistics.numContinuations++;

            CommGroup grp = new CommGroup(replicate);
            for (int i = 0; i < ml.length; ++i) {
                CommChannel chnl = (CommChannel) ChannelFactory
                        .getBackend(ml[i].getChannel());
                // TODO see below..
                // oframe.setDebugInfo(fillDebugInfo());
                CommRecv recv = new CommRecv(chnl, ml[i]);
                grp.add(recv);
            }
            _executionQueue.add(grp);
        }

        public void object(boolean replicate, ChannelListener methodList)
                throws IllegalArgumentException {
            object(replicate, new ChannelListener[] { methodList });
        }

        /* UNUSED
         private DebugInfo fillDebugInfo() {
            // Some of the debug information is a bit lengthy, so lets not put
            // it in all the time... eh.
            DebugInfo frame = new DebugInfo();
            frame.setCreator(_source);
            Exception ex = new Exception();
            StackTraceElement[] st = ex.getStackTrace();
            if (st.length > 2) {
                StackTraceElement[] stcut = new StackTraceElement[st.length - 2];
                System.arraycopy(st, 2, stcut, 0, stcut.length);
                frame.setLocation(stcut);
            }

            return frame;
        }
        */

        public Object getExtension(Class extensionClass) {
            return _extensions.get(extensionClass);
        }

        public void run() {
            assert _methodBody != null;
            assert _method != null;
            assert _method.getDeclaringClass().isAssignableFrom(_methodBody.getClass());

            if (__log.isTraceEnabled()) {
                __log.trace(_cycle + ": " + _source);
            }

            Object[] args;
            SynchChannel synchChannel;
            if (_method.getReturnType() != void.class) {
                args = new Object[_args.length - 1];
                System.arraycopy(_args, 0, args, 0, args.length);
                synchChannel = (SynchChannel) _args[args.length];
            } else {
                args = _args;
                synchChannel = null;
            }
            stackThread();
            long ctime = System.currentTimeMillis();
            try {
                _method.invoke(_methodBody, args);
                if (synchChannel != null) {
                    synchChannel.ret();
                }
            } catch (IllegalAccessException iae) {
                String msg = __msgs.msgMethodNotAccessible(_method.getName(),
                        _method.getDeclaringClass().getName());
                __log.error(msg, iae);
                throw new RuntimeException(msg, iae);
            } catch (InvocationTargetException e) {
                String msg = __msgs.msgClientMethodException(_method.getName(),
                        _methodBody.getClass().getName());
                __log.error(msg, e.getTargetException());
                throw new RuntimeException(e.getTargetException());
            } finally {
                ctime = System.currentTimeMillis() - ctime;
                _statistics.totalClientTimeMs += ctime;
                unstackThread();
            }
        }

        public String toString() {
            return "PT[ " + _methodBody + " ]";
        }

        private void stackThread() {
            Stack<JacobThread> currStack = __activeJacobThread.get();
            if (currStack == null) {
                currStack = new Stack<JacobThread>();
                __activeJacobThread.set(currStack);
            }
            currStack.push(this);
        }

        private JacobThread unstackThread() {
            Stack<JacobThread> currStack = __activeJacobThread.get();
            assert currStack != null;
            return currStack.pop();
        }
    }

}