soaphandler.java

来自「ejb3 java session bean」· Java 代码 · 共 562 行 · 第 1/2 页

          continue;
        }

        // instantiate lookup class
        Object lookup = lookupClass.newInstance();
        // register lookup
        lookups.put(contextClassName, lookup);
        log.debug("registered endpoint metadata lookup: name="
            + contextClassName
            + ", class="
            + lookupClassName);
      }
      catch (ClassNotFoundException e) {
        log.debug("endpoint metadata lookup not found, skipping: " + lookupClassName);
      }
      catch (InstantiationException e) {
        log.warn("endpoint metadata lookup class not instantiable: " + lookupClassName, e);
      }
      catch (IllegalAccessException e) {
        log.warn("endpoint metadata lookup class or constructor not public: " + lookupClassName, e);
      }
    }
    return lookups;
  }

  public boolean handleResponse(MessageContext messageContext) throws JAXRPCException {
    Map parts = (Map) messageContext.getProperty(MESSAGE_PARTS_PROP);
    SOAPFaultException faultException = (SOAPFaultException) messageContext.getProperty(FAULT_EXCEPTION_PROP);

    // absence of both parts and fault means one-way operation
    if (parts == null && faultException == null)
      return true;

    String operationName = (String) messageContext.getProperty(OPERATION_NAME_PROP);
    SOAPMessage soapMessage = ((SOAPMessageContext) messageContext).getMessage();

    JbpmContext jbpmContext = integrationControl.getIntegrationServiceFactory()
        .getJbpmConfiguration()
        .createJbpmContext();
    try {
      lookupEndpointMetadata(messageContext);

      SOAPEnvelope envelope = soapMessage.getSOAPPart().getEnvelope();
      // remove existing body, it might have undesirable content
      SOAPBody body = envelope.getBody();
      body.detachNode();
      // re-create body
      body = envelope.addBody();

      if (faultException == null)
        writeOutput(operationName, soapMessage, parts);
      else {
        String faultName = (String) messageContext.getProperty(FAULT_NAME_PROP);
        writeFault(operationName, soapMessage, faultName, parts, faultException);
      }
    }
    /*
     * NO need to set jbpm context as rollback only for any exception, since operations in try-block
     * only read definitions from database
     */
    catch (SOAPException e) {
      throw new JAXRPCException("could not compose outbound soap message", e);
    }
    finally {
      jbpmContext.close();
    }
    return true;
  }

  public boolean handleFault(MessageContext messageContext) throws JAXRPCException {
    return true;
  }

  protected ObjectMessage sendRequest(SOAPMessage soapMessage, Session jmsSession,
      JbpmContext jbpmContext) throws SOAPException, JMSException {
    // create a jms message to deliver the incoming content
    ObjectMessage jmsRequest = jmsSession.createObjectMessage();

    // put the partner link identified by handle in a jms property
    PartnerLinkEntry partnerLinkEntry = integrationControl.getPartnerLinkEntry(portTypeName,
        serviceName, portName);
    long partnerLinkId = partnerLinkEntry.getId();
    jmsRequest.setLongProperty(IntegrationConstants.PARTNER_LINK_ID_PROP, partnerLinkId);

    Operation operation = determineOperation(soapMessage);
    if (operation == null)
      throw new SOAPException("could not determine operation to perform");

    // put the operation name in a jms property
    String operationName = operation.getName();
    jmsRequest.setStringProperty(IntegrationConstants.OPERATION_NAME_PROP, operationName);

    log.debug("received request: partnerLink=" + partnerLinkId + ", operation=" + operationName);

    // extract message content
    HashMap requestParts = new HashMap();
    formatter.readMessage(operationName, soapMessage, requestParts, MessageDirection.INPUT);
    jmsRequest.setObject(requestParts);

    // fill message properties
    BpelProcessDefinition process = integrationControl.getDeploymentDescriptor()
        .findProcessDefinition(jbpmContext);
    MessageType requestType = process.getImportDefinition().getMessageType(
        operation.getInput().getMessage().getQName());
    fillCorrelationProperties(requestParts, jmsRequest, requestType.getPropertyAliases());

    // set up producer
    MessageProducer producer = jmsSession.createProducer(partnerLinkEntry.getDestination());
    try {
      // is the exchange pattern request/response?
      if (operation.getOutput() != null) {
        Destination replyTo = integrationControl.getIntegrationServiceFactory()
            .getResponseDestination();
        jmsRequest.setJMSReplyTo(replyTo);

        // have jms discard request message if response timeout expires
        Number responseTimeout = getResponseTimeout(jbpmContext);
        if (responseTimeout != null)
          producer.setTimeToLive(responseTimeout.longValue());
      }
      else {
        // have jms discard message if one-way timeout expires
        Number oneWayTimeout = getOneWayTimeout(jbpmContext);
        if (oneWayTimeout != null)
          producer.setTimeToLive(oneWayTimeout.longValue());
      }

      // send request message
      producer.send(jmsRequest);
      log.debug("sent request: " + RequestListener.messageToString(jmsRequest));

      return jmsRequest;
    }
    finally {
      // release producer resources
      producer.close();
    }
  }

  private static Number getResponseTimeout(JbpmContext jbpmContext) {
    Object responseTimeout = jbpmContext.getObjectFactory().createObject(
        "jbpm.bpel.response.timeout");
    if (responseTimeout instanceof Number)
      return (Number) responseTimeout;
    else if (responseTimeout != null)
      log.warn("response timeout is not a number: " + responseTimeout);
    return null;
  }

  private static Number getOneWayTimeout(JbpmContext jbpmContext) {
    Object oneWayTimeout = jbpmContext.getObjectFactory().createObject("jbpm.bpel.oneway.timeout");
    if (oneWayTimeout instanceof Number)
      return (Number) oneWayTimeout;
    else if (oneWayTimeout != null)
      log.warn("one-way timeout is not a number: " + oneWayTimeout);
    return null;
  }

  private Operation determineOperation(SOAPMessage soapMessage) throws SOAPException {
    Binding binding = formatter.getBinding();

    SOAPBinding soapBinding = (SOAPBinding) WsdlUtil.getExtension(
        binding.getExtensibilityElements(),
        com.ibm.wsdl.extensions.soap.SOAPConstants.Q_ELEM_SOAP_BINDING);

    String style = soapBinding.getStyle();
    if (style == null) {
      // wsdlsoap:binding does not specify any style, assume 'document'
      style = SoapBindConstants.DOCUMENT_STYLE;
    }

    PortType portType = binding.getPortType();
    SOAPElement bodyElement = SoapUtil.getElement(soapMessage.getSOAPBody());

    if (style.equals(SoapBindConstants.RPC_STYLE)) {
      String operationName = bodyElement.getLocalName();
      return portType.getOperation(operationName, null, null);
    }

    List operations = portType.getOperations();
    for (int i = 0, n = operations.size(); i < n; i++) {
      Operation operation = (Operation) operations.get(i);
      Message inputMessage = operation.getInput().getMessage();
      QName docLitElementName = WsdlUtil.getDocLitElementName(inputMessage);

      if (XmlUtil.nodeNameEquals(bodyElement, docLitElementName))
        return operation;
    }
    return null;
  }

  /**
   * Gets the values of message properties from the request message parts and sets them in the
   * property fields of the JMS message.
   * @param requestParts the parts extracted from the request SOAP message
   * @param jmsRequest the JMS message whose properties will be set
   * @param propertyAliases the property aliases associated with the request message type
   * @throws JMSException
   */
  private static void fillCorrelationProperties(Map requestParts, ObjectMessage jmsRequest,
      Map propertyAliases) throws JMSException {
    // easy way out: no property aliases
    if (propertyAliases == null)
      return;
    // iterate through the property aliases associated with the message type
    for (Iterator i = propertyAliases.entrySet().iterator(); i.hasNext();) {
      Entry aliasEntry = (Entry) i.next();
      QName propertyName = (QName) aliasEntry.getKey();
      PropertyAlias alias = (PropertyAlias) aliasEntry.getValue();
      // get part accessor from operation wrapper
      String partName = alias.getPart();
      Object value = requestParts.get(partName);
      if (value == null) {
        log.debug("message part not found, cannot get property value: property="
            + propertyName
            + ", part="
            + partName);
        continue;
      }
      // evaluate the query against the part value, if any
      PropertyQuery query = alias.getQuery();
      if (query != null) {
        try {
          value = query.getEvaluator().evaluate((Element) value);
        }
        catch (BpelFaultException e) {
          // the most likely cause is a selection failure due to missing nodes
          log.debug("query evaluation failed, "
              + "cannot get property value: property="
              + propertyName
              + ", part="
              + partName
              + ", query="
              + query.getText(), e);
          continue;
        }
      }
      // set the value in a jms message property field
      jmsRequest.setObjectProperty(propertyName.getLocalPart(),
          value instanceof Node ? DatatypeUtil.toString((Node) value) : value);
    }
  }

  protected ObjectMessage receiveResponse(Session jmsSession, Destination replyTo,
      String requestId, JbpmContext jbpmContext) throws JMSException, SOAPFaultException {
    // set up consumer
    String selector = "JMSCorrelationID='" + requestId + '\'';
    MessageConsumer consumer = jmsSession.createConsumer(replyTo, selector);
    try {
      // receive response message
      log.debug("listening for response: destination=" + replyTo + ", requestId=" + requestId);
      Number responseTimeout = getResponseTimeout(jbpmContext);
      ObjectMessage jmsResponse = (ObjectMessage) (responseTimeout != null ? consumer.receive(responseTimeout.longValue())
          : consumer.receive());
      // did a message arrive in time?
      if (jmsResponse == null) {
        log.debug("response timeout expired: destination=" + replyTo + ", requestId" + requestId);
        throw new SOAPFaultException(SoapBindConstants.SERVER_FAULTCODE,
            SoapBindConstants.TIMEOUT_FAULTSTRING, null, null);
      }
      jmsResponse.acknowledge();
      log.debug("received response: " + RequestListener.messageToString(jmsResponse));
      return jmsResponse;
    }
    finally {
      // release consumer resources
      consumer.close();
    }
  }

  protected void writeOutput(String operationName, SOAPMessage soapMessage, Map responseParts)
      throws SOAPException {
    formatter.writeMessage(operationName, soapMessage, responseParts, MessageDirection.OUTPUT);
  }

  protected void writeFault(String operationName, SOAPMessage soapMessage, String faultName,
      Map faultParts, SOAPFaultException faultException) throws SOAPException {
    formatter.writeFault(operationName, soapMessage, faultName, faultParts,
        faultException.getFaultCode(), faultException.getFaultString());
  }
}