integrationcontrol.java

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

    }
    return id;
  }

  public Map getCallers() {
    return callers;
  }

  public Caller createCaller(InvokeAction invokeAction, Token token) {
    PartnerLinkDefinition partnerLinkDefinition = invokeAction.getPartnerLink();
    PartnerLinkInstance partnerLinkInstance = partnerLinkDefinition.getInstance(token);

    // retrieve partner reference
    EndpointReference partnerReference = partnerLinkInstance.getPartnerReference();
    if (partnerReference == null) {
      // create reference whose sole selection criterion is the port type
      partnerReference = createPartnerReference(partnerLinkDefinition);
      partnerLinkInstance.setPartnerReference(partnerReference);
    }

    // select a port from the service catalog with the criteria known at this point
    Port port = partnerReference.selectPort(getPartnerCatalog());
    log.debug("selected port " + port.getName() + " on " + invokeAction + " for " + token);

    // create a client for that port
    Caller caller = new SoapCaller(port);
    callers.put(new Caller.Key(invokeAction.getId(), token.getId()), caller);
    return caller;
  }

  EndpointReference createPartnerReference(PartnerLinkDefinition partnerLink) {
    PartnerLinkEntry entry = getPartnerLinkEntry(partnerLink);
    InitiateMode initiateMode = entry.getInitiateMode();

    EndpointReference partnerReference;
    if (InitiateMode.PULL.equals(initiateMode)) {
      EndpointReferenceFactory referenceFactory = EndpointReferenceFactory.getInstance(
          IntegrationConstants.DEFAULT_REFERENCE_NAME, null);
      partnerReference = referenceFactory.createEndpointReference();
    }
    else if (InitiateMode.STATIC.equals(initiateMode))
      partnerReference = entry.getPartnerReference();
    else
      throw new BpelFaultException(BpelConstants.FAULT_UNINITIALIZED_PARTNER_ROLE);

    return partnerReference;
  }

  public Caller removeCaller(InvokeAction invokeAction, Token token) {
    return (Caller) callers.remove(new Caller.Key(invokeAction.getId(), token.getId()));
  }

  public ServiceCatalog getPartnerCatalog() {
    return getDeploymentDescriptor().getServiceCatalog();
  }

  public List getStartListeners() {
    return startListeners;
  }

  public void addStartListener(StartListener startListener) {
    startListeners.add(startListener);
  }

  public void removeStartListener(StartListener startListener) {
    startListeners.remove(startListener);
  }

  public DefinitionCatalog getMyCatalog() {
    return myCatalog;
  }

  /**
   * Prepares inbound message activities annotated to create a process instance for receiving
   * requests.
   */
  public void enableInboundMessageActivities(JbpmContext jbpmContext) throws NamingException,
      JMSException {
    InitialContext initialContext = new InitialContext();
    try {
      // publish partner link information to JNDI
      BpelProcessDefinition processDefinition = getDeploymentDescriptor().findProcessDefinition(
          jbpmContext);
      buildPartnerLinkEntries(initialContext, processDefinition);

      // open a jms connection
      openJmsConnection(initialContext);

      try {
        // enable start IMAs
        StartListenersBuilder builder = new StartListenersBuilder(this);
        builder.visit(processDefinition);

        // note: upon creation, start listeners add themselves to this control
        if (startListeners.isEmpty())
          throw new BpelException(processDefinition + " has no start activities");

        // enable outstanding IMAs
        IntegrationSession integrationSession = IntegrationSession.getContextInstance(jbpmContext);
        JmsIntegrationService integrationService = JmsIntegrationService.get(jbpmContext);

        // receive
        for (Iterator i = integrationSession.findReceiveTokens(processDefinition).iterator(); i.hasNext();) {
          Token token = (Token) i.next();
          Receive receive = (Receive) token.getNode();
          integrationService.jmsReceive(receive.getReceiveAction(), token, this, true);
        }

        // pick
        for (Iterator i = integrationSession.findPickTokens(processDefinition).iterator(); i.hasNext();) {
          Token token = (Token) i.next();
          // pick points activity token to begin mark
          Begin begin = (Begin) token.getNode();
          Pick pick = (Pick) begin.getCompositeActivity();
          integrationService.jmsReceive(pick.getOnMessages(), token, this);
        }

        // event
        for (Iterator t = integrationSession.findEventTokens(processDefinition).iterator(); t.hasNext();) {
          Token token = (Token) t.next();
          // scope points events token to itself
          Scope scope = (Scope) token.getNode();
          List onEvents = scope.getOnEvents();
          for (int i = 0, n = onEvents.size(); i < n; i++) {
            OnEvent onEvent = (OnEvent) onEvents.get(i);
            integrationService.jmsReceive(onEvent.getReceiveAction(), token, this, false);
          }
        }

        // start message delivery
        jmsConnection.start();
      }
      catch (JMSException e) {
        jmsConnection.close();
        throw e;
      }
    }
    finally {
      initialContext.close();
    }
  }

  /**
   * Prevents inbound message activities annotated to create a process instance from further
   * receiving requests.
   */
  public void disableInboundMessageActivities() throws JMSException {
    // disable start IMAs
    synchronized (startListeners) {
      for (int i = 0, n = startListeners.size(); i < n; i++) {
        StartListener startListener = (StartListener) startListeners.get(i);
        startListener.close();
      }
      startListeners.clear();
    }

    // disable outstanding IMAs
    synchronized (requestListeners) {
      for (Iterator i = requestListeners.values().iterator(); i.hasNext();) {
        RequestListener requestListener = (RequestListener) i.next();
        requestListener.close();
      }
      requestListeners.clear();
    }

    // release jms connection
    closeJmsConnection();
  }

  void buildPartnerLinkEntries(InitialContext initialContext, BpelProcessDefinition process)
      throws NamingException {
    // match scopes with their descriptors
    ScopeMatcher scopeMatcher = new ScopeMatcher(process);
    scopeMatcher.visit(getDeploymentDescriptor());
    Map scopeDescriptors = scopeMatcher.getScopeDescriptors();
    // lookup destinations & bind port entries
    PartnerLinkEntriesBuilder builder = new PartnerLinkEntriesBuilder(scopeDescriptors,
        getJmsContext(initialContext), integrationServiceFactory.getRequestDestination());
    builder.visit(process);
    partnerLinkEntries = builder.getPartnerLinkEntries();
  }

  private static Context getJmsContext(InitialContext initialContext) {
    Context jmsContext;
    try {
      jmsContext = (Context) initialContext.lookup(JMS_CONTEXT);
      log.debug("retrieved jms context: " + JMS_CONTEXT);
    }
    catch (NamingException e) {
      log.debug("jms context not found: " + JMS_CONTEXT);
      jmsContext = initialContext;
      log.debug("fell back on initial context");
    }
    return jmsContext;
  }

  void openJmsConnection(InitialContext initialContext) throws NamingException, JMSException {
    ConnectionFactory jmsConnectionFactory = getConnectionFactory(initialContext);
    jmsConnection = jmsConnectionFactory.createConnection();
  }

  private ConnectionFactory getConnectionFactory(InitialContext initialContext)
      throws NamingException {
    Context jmsContext = getJmsContext(initialContext);
    ConnectionFactory jmsConnectionFactory;
    try {
      jmsConnectionFactory = (ConnectionFactory) jmsContext.lookup(CONNECTION_FACTORY_NAME);
      log.debug("retrieved jms connection factory: " + CONNECTION_FACTORY_NAME);
    }
    catch (NamingException e) {
      log.debug("jms connection factory not found: " + CONNECTION_FACTORY_NAME);
      jmsConnectionFactory = integrationServiceFactory.getConnectionFactory();
      if (jmsConnectionFactory == null)
        throw e;
      log.debug("fell back on default connection factory");
    }
    return jmsConnectionFactory;
  }

  void closeJmsConnection() throws JMSException {
    if (jmsConnection != null) {
      jmsConnection.close();
      jmsConnection = null;
    }
  }

  void reset() {
    deploymentDescriptor = null;

    partnerLinkEntries = Collections.EMPTY_LIST;

    startListeners.clear();
    requestListeners.clear();
    outstandingRequests.clear();

    callers.clear();
  }
}