integrationcontrol.java

jbpm-bpel-1.1.Beta3 JBoss jBPM Starters Kit 是一个综合包

/*
 * JBoss, Home of Professional Open Source
 * Copyright 2005, JBoss Inc., and individual contributors as indicated
 * by the @authors tag.
 *
 * This is free software; you can redistribute it and/or modify it
 * under the terms of the JBPM BPEL PUBLIC LICENSE AGREEMENT as
 * published by JBoss Inc.; either version 1.0 of the License, or
 * (at your option) any later version.
 *
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 */
package org.jbpm.bpel.integration.jms;

import java.util.Collections;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NameNotFoundException;
import javax.naming.NamingException;
import javax.wsdl.Operation;
import javax.wsdl.Port;
import javax.xml.namespace.QName;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.jbpm.JbpmConfiguration;
import org.jbpm.JbpmContext;
import org.jbpm.bpel.app.AppDescriptor;
import org.jbpm.bpel.app.ScopeMatcher;
import org.jbpm.bpel.app.AppPartnerRole.InitiateMode;
import org.jbpm.bpel.db.IntegrationSession;
import org.jbpm.bpel.endpointref.EndpointReference;
import org.jbpm.bpel.endpointref.EndpointReferenceFactory;
import org.jbpm.bpel.endpointref.wsa.WsaConstants;
import org.jbpm.bpel.graph.basic.Receive;
import org.jbpm.bpel.graph.def.BpelDefinition;
import org.jbpm.bpel.graph.exe.BpelFaultException;
import org.jbpm.bpel.graph.scope.OnEvent;
import org.jbpm.bpel.graph.scope.Scope;
import org.jbpm.bpel.graph.struct.Pick;
import org.jbpm.bpel.graph.struct.StructuredActivity.Begin;
import org.jbpm.bpel.integration.IntegrationService;
import org.jbpm.bpel.integration.catalog.ServiceCatalog;
import org.jbpm.bpel.integration.client.SoapClient;
import org.jbpm.bpel.integration.def.PartnerLinkDefinition;
import org.jbpm.bpel.integration.def.Receiver;
import org.jbpm.bpel.integration.def.Replier;
import org.jbpm.bpel.integration.exe.PartnerLinkInstance;
import org.jbpm.bpel.xml.BpelConstants;
import org.jbpm.graph.exe.Token;
import org.jbpm.svc.Services;

/**
 * @author Alejandro Gu韟ar
 * @version $Revision: 1.7 $ $Date: 2007/01/22 17:27:03 $
 */
public class IntegrationControl {

  private final JmsIntegrationServiceFactory integrationServiceFactory;

  private Connection jmsConnection;
  private AppDescriptor appDescriptor;

  private Map partnerLinkEntriesById;
  private Map partnerLinkEntriesByHandle;

  private List startListeners = Collections.EMPTY_LIST;
  private Map requestListeners = new Hashtable();
  private Map outstandingRequests = new Hashtable();
  private Map partnerClients = new HashMap();

  /**
   * jBPM configuration value of an alternate JNDI subcontext containing JMS
   * objects, relative to the initial context.
   */
  public static final String NAME_JMS_CONTEXT = "jbpm.bpel.jms.context";

  /**
   * Default JNDI subcontext containing JMS objects, relative to the initial
   * context.
   */
  public static final String DEFAULT_JMS_CONTEXT = "java:comp/env/jms";

  /** JNDI name bound to the JMS connection factory, relative to the JMS context. */
  public static final String CONNECTION_FACTORY_NAME = "IntegrationConnectionFactory";

  private static final QName DEFAULT_REFERENCE_NAME = new QName(
      WsaConstants.NS_ADDRESSING, WsaConstants.ELEM_ENDPOINT_REFERENCE);

  public static final Log log = LogFactory.getLog(IntegrationControl.class);
  private static final long serialVersionUID = 1L;

  IntegrationControl(JmsIntegrationServiceFactory integrationServiceFactory) {
    this.integrationServiceFactory = integrationServiceFactory;
  }

  public JmsIntegrationServiceFactory getIntegrationServiceFactory() {
    return integrationServiceFactory;
  }

  public Connection getJmsConnection() {
    return jmsConnection;
  }

  public AppDescriptor getAppDescriptor() {
    return appDescriptor;
  }

  public void setAppDescriptor(AppDescriptor appDescriptor) {
    this.appDescriptor = appDescriptor;
  }

  public PartnerLinkEntry getPartnerLinkEntry(PartnerLinkDefinition partnerLink) {
    return (PartnerLinkEntry) partnerLinkEntriesById.get(new Long(
        partnerLink.getId()));
  }

  public PartnerLinkEntry getPartnerLinkEntry(String handle) {
    return (PartnerLinkEntry) partnerLinkEntriesByHandle.get(handle);
  }

  public Map getRequestListeners() {
    return Collections.unmodifiableMap(requestListeners);
  }

  public void addRequestListener(RequestListener requestListener) {
    Object listenerKey = createKey(requestListener.getReceiverId(),
        requestListener.getTokenId());
    requestListeners.put(listenerKey, requestListener);
  }

  public RequestListener removeRequestListener(Receiver receiver, Token token) {
    Object listenerKey = createKey(receiver.getId(), token.getId());
    return (RequestListener) requestListeners.remove(listenerKey);
  }

  private static Object createKey(long receiverId, long tokenId) {
    return new RequestListener.Key(receiverId, tokenId);
  }

  public Map getOutstandingRequests() {
    return Collections.unmodifiableMap(outstandingRequests);
  }

  public void addOutstandingRequest(Receiver receiver, Token token,
      OutstandingRequest request) {
    Object key = createKey(receiver.getPartnerLink().getInstance(token),
        receiver.getOperation(), receiver.getMessageExchange());
    if (outstandingRequests.put(key, request) != null) {
      throw new BpelFaultException(BpelConstants.FAULT_CONFLICTING_REQUEST);
    }
    log.debug("added outstanding request: receiver="
        + receiver
        + ", token="
        + token
        + ", request="
        + request);
  }

  public OutstandingRequest removeOutstandingRequest(Replier replier,
      Token token) {
    Object key = createKey(replier.getPartnerLink().getInstance(token),
        replier.getOperation(), replier.getMessageExchange());
    OutstandingRequest request = (OutstandingRequest) outstandingRequests.remove(key);
    if (request == null)
      throw new BpelFaultException(BpelConstants.FAULT_MISSING_REQUEST);

    log.debug("removed outstanding request: replier="
        + replier
        + ", token="
        + token
        + ", request="
        + request);
    return request;
  }

  private static Object createKey(PartnerLinkInstance partnerLinkInstance,
      Operation operation, String messageExchange) {
    return new OutstandingRequest.Key(getOrAssignId(partnerLinkInstance),
        operation.getName(), messageExchange);
  }

  public Map getPartnerClients() {
    return Collections.unmodifiableMap(partnerClients);
  }

  public SoapClient getPartnerClient(PartnerLinkInstance instance) {
    Long instanceId = new Long(getOrAssignId(instance));
    SoapClient partnerClient;
    synchronized (partnerClients) {
      // retrieve cached port consumer
      partnerClient = (SoapClient) partnerClients.get(instanceId);
      if (partnerClient == null) {
        // no cached consumer, create one from partner endpoint reference
        EndpointReference partnerRef = instance.getPartnerReference();
        if (partnerRef == null) {
          // no partner reference, create one containing only the port type as
          // selection criterion
          partnerRef = createPartnerReference(instance.getDefinition());
          instance.setPartnerReference(partnerRef);
          log.debug("initialized partner reference: instance="