soapexternalservice.java

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

     * @return The action value for the specified operation
     */
    private String getAction(String operation) {
        String action = _converter.getWSAInputAction(operation);
        if (action == null || "".equals(action)) {
            action = _converter.getSoapAction(operation);
        }
        return action;
    }

    /**
     * Extracts endpoint information from ODE message exchange to stuff them into Axis MessageContext.
     */
    private void writeHeader(MessageContext ctxt, PartnerRoleMessageExchange odeMex) {
        Options options = ctxt.getOptions();
        WSAEndpoint targetEPR = EndpointFactory.convertToWSA((MutableEndpoint) odeMex.getEndpointReference());
        WSAEndpoint myRoleEPR = EndpointFactory.convertToWSA((MutableEndpoint) odeMex.getMyRoleEndpointReference());

        String partnerSessionId = odeMex.getProperty(MessageExchange.PROPERTY_SEP_PARTNERROLE_SESSIONID);
        String myRoleSessionId = odeMex.getProperty(MessageExchange.PROPERTY_SEP_MYROLE_SESSIONID);

        if (partnerSessionId != null) {
            if (__log.isDebugEnabled()) {
                __log.debug("Partner session identifier found for WSA endpoint: " + partnerSessionId);
            }
            targetEPR.setSessionId(partnerSessionId);
        }
        options.setProperty("targetSessionEndpoint", targetEPR);

        if (myRoleEPR != null) {
            if (myRoleSessionId != null) {
                if (__log.isDebugEnabled()) {
                    __log.debug("MyRole session identifier found for myrole (callback) WSA endpoint: "
                            + myRoleSessionId);
                }
                myRoleEPR.setSessionId(myRoleSessionId);
            }
            options.setProperty("callbackSessionEndpoint", odeMex.getMyRoleEndpointReference());
        } else {
            __log.debug("My-Role EPR not specified, SEP will not be used.");
        }

        String action = getAction(odeMex.getOperationName());
        ctxt.setSoapAction(action);

        if (MessageExchange.MessageExchangePattern.REQUEST_RESPONSE == odeMex.getMessageExchangePattern()) {
            EndpointReference annonEpr =
                    new EndpointReference(Namespaces.WS_ADDRESSING_ANON_URI);
            ctxt.setReplyTo(annonEpr);
            ctxt.setMessageID("uuid:" + new UUID().toString());
        }
    }

    public org.apache.ode.bpel.iapi.EndpointReference getInitialEndpointReference() {
return endpointReference;
    }

    public void close() {
        // nothing
    }

    public String getPortName() {
        return _portName;
    }

    public QName getServiceName() {
        return _serviceName;
    }

    private void replyWithFailure(final String odeMexId, final FailureType error, final String errmsg,
                                  final Element details) {
        // ODE MEX needs to be invoked in a TX.
        try {
            _sched.execIsolatedTransaction(new Callable<Void>() {
                public Void call() throws Exception {
                    PartnerRoleMessageExchange odeMex = (PartnerRoleMessageExchange) _server.getEngine().getMessageExchange(odeMexId);
                    odeMex.replyWithFailure(error, errmsg, details);
                    return null;
                }
            });

        } catch (Exception e) {
            String emsg = "Error executing replyWithFailure transaction; reply will be lost.";
            __log.error(emsg, e);

        }

    }

    private void reply(final String odeMexId, final Operation operation, final MessageContext reply, final boolean fault) {
        // ODE MEX needs to be invoked in a TX.
        try {
            _sched.execIsolatedTransaction(new Callable<Void>() {
                public Void call() throws Exception {
                    PartnerRoleMessageExchange odeMex = (PartnerRoleMessageExchange) _server.getEngine().getMessageExchange(odeMexId);
                    // Setting the response
                    try {
                        if (__log.isDebugEnabled()) __log.debug("Received response for MEX " + odeMex);
                        if (fault) {
                            Document odeMsg = DOMUtils.newDocument();
                            Element odeMsgEl = odeMsg.createElementNS(null, "message");
                            odeMsg.appendChild(odeMsgEl);
                            QName faultType = _converter.parseSoapFault(odeMsgEl, reply.getEnvelope(), operation);
                            if (__log.isDebugEnabled()) __log.debug("Reply is a fault, found type: " + faultType);

                            if (faultType != null) {
                                if (__log.isWarnEnabled())
                                    __log.warn("Fault response: faultType=" + faultType + "\n" + DOMUtils.domToString(odeMsgEl));
                                Message response = odeMex.createMessage(faultType);
                                response.setMessage(odeMsgEl);

                                odeMex.replyWithFault(faultType, response);
                            } else {
                                if (__log.isWarnEnabled())
                                    __log.warn("Fault response: faultType=(unkown)\n" + reply.getEnvelope().toString());
                                odeMex.replyWithFailure(FailureType.OTHER, reply.getEnvelope().getBody()
                                        .getFault().getText(), OMUtils.toDOM(reply.getEnvelope().getBody()));
                            }
                        } else {
                            Message response = odeMex.createMessage(odeMex.getOperation().getOutput().getMessage().getQName());
                            _converter.parseSoapResponse(response, reply.getEnvelope(), operation);
                            if (__log.isInfoEnabled()) __log.info("Response:\n" + (response.getMessage() != null ?
                                    DOMUtils.domToString(response.getMessage()) : "empty"));
                            odeMex.reply(response);
                        }
                    } catch (Exception ex) {
                        String errmsg = "Unable to process response: " + ex.getMessage();
                        __log.error(errmsg, ex);
                        odeMex.replyWithFailure(FailureType.OTHER, errmsg, null);
                    }
                    return null;
                }
            });

        } catch (Exception e) {
            String errmsg = "Error executing reply transaction; reply will be lost.";
            __log.error(errmsg, e);
        }

    }


    /**
     * This class wraps a {@link org.apache.axis2.client.ServiceClient} and watches changes (deletions,creations,updates)
     * on a  Axis2 service config file named {service-name}.axis2.<p/>
     * The {@link org.apache.axis2.client.ServiceClient} instance is created from the main Axis2 config instance and
     * this service-specific config file.
     */
    private class CachedServiceClient extends FileWatchDog {
        ServiceClient client;

        protected CachedServiceClient(File file, long delay) {
            super(file, delay);
        }

        protected boolean isInitialized() {
            return client != null;
        }

        protected void init() {
            try {
                client = new ServiceClient(new ConfigurationContext(_axisConfig), null);
            } catch (AxisFault axisFault) {
                throw new RuntimeException(axisFault);
            }
        }

        protected void doOnUpdate() {
            // axis2 service configuration
            // if the config file has been modified (i.e added or updated), re-create a ServiceClient
            // and load the new config.
            init(); // create a new ServiceClient instance
            try {
                InputStream ais = file.toURI().toURL().openStream();
                if (ais != null) {
                    if (__log.isDebugEnabled()) __log.debug("Configuring service " + _serviceName + " using: " + file);
                    ServiceBuilder builder = new ServiceBuilder(ais, new ConfigurationContext(client.getAxisConfiguration()), client.getAxisService());
                    builder.populateService(builder.buildOM());
                }
            } catch (Exception e) {
                if (__log.isWarnEnabled()) __log.warn("Exception while configuring service: " + _serviceName, e);
            }
        }
    }

    private class CachedOptions extends WatchDog<Map> {

        Options options;

        private CachedOptions() {
            super(new WatchDog.Mutable<Map>() {
                // ProcessConf#getProperties(String...) cannot return ull (by contract)
                public boolean exists() {
                    return true;
                }

                public boolean hasChangedSince(Map since) {
                    Map latest = lastModified();  // cannot be null but better be prepared
                    // check if mappings are equal
                    return !CollectionUtils.equals(latest, since);
                }

                public Map lastModified() {
                    return _pconf.getEndpointProperties(endpointReference);
                }
            });
        }

        protected boolean isInitialized() {
            return options != null;
        }

        protected void init() {
            options = new Options();
        }

        protected void doOnUpdate() {
            init();

            // note: don't make this map an instance attribute, so we always get the latest version
            final Map<String, String> properties = _pconf.getEndpointProperties(endpointReference);
            Properties.Axis2.translate(properties, options);

            // set defaults values
            options.setExceptionToBeThrownOnSOAPFault(false);

            // this value does NOT override Properties.PROP_HTTP_CONNECTION_TIMEOUT
            // nor Properties.PROP_HTTP_SOCKET_TIMEOUT.
            // it will be applied only if the laters are not set.
            options.setTimeOutInMilliSeconds(60000);
        }
    }

}