Dispatches the given messages to the message endpoint.

@param messages
the messages to dispatch
@throws ResourceException
if the dispatch fails
358461
synchronized void dispatch(final List messages, AsynchDispatchScheduler scheduler, SibRaListener listener) throws ResourceException {

    	final String methodName = "dispatch";
    	if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled()) {
    		SibTr.entry(this, TRACE, methodName, new Object[] {messages, scheduler, listener});
    	}

    	if (_session == null) {

    		/*
    		 * The dispatcher has been closed - return safe in the knowledge
    		 * that the messages will have already been unlocked by the close of
    		 * the session
    		 */

    		if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled()) {
    			SibTr.exit(this, TRACE, methodName);
    		}
    		return;

    	}

    	/*
    	 * Keep a list of the messages for which delivery has not been attempted
    	 */

    	final List undeliveredDispatchableMessages = new ArrayList(messages);


    	/**
    	 * The endpoint to deliver the message to.
    	 */
    	MessageEndpoint endpoint = null;

    	try {

    		/*
    		 * Obtain an endpoint
    		 */
    		endpoint = createEndpoint();

    		/*
    		 * Obtain information for request metrics and handlers
    		 */
    		final SIDestinationAddress destination = _session
    				.getDestinationAddress();
    		final String busName = destination.getBusName();
    		final String meName = _connection.getMeName();
    		final String destinationName = destination.getDestinationName();
    		final String messageSelector = _endpointConfiguration
    				.getMessageSelector();
    		final ActivationSpec activationSpec = _endpointConfiguration
    				.getActivationSpec();

    		for (final Iterator iterator = messages.iterator(); iterator
    				.hasNext()
    				&& !isCancelled ();) {

    			final DispatchableMessage dispatchableMessage = (DispatchableMessage) iterator.next();

    			final SIBusMessage message = dispatchableMessage.getMessage();
    			boolean success = false;
    			try {

    				beforeDelivery(message, endpoint);

    				undeliveredDispatchableMessages.remove(dispatchableMessage);


    				try {

    					// 304501.ra - Call the pre MDB invoke on XD
    					MessagePacingControl mpc = MessagePacingControlFactory.getInstance();
    					if (mpc != null && mpc.isActive()) {

    						mpc.preMdbInvoke(this, scheduler);

    					}

    					_reliabilityPreInvoke.put(message.getMessageHandle(), message.getReliability());

    					success = _invoker.invokeEndpoint(endpoint,
    							message, _session, getTransaction(), _debugMEName);

    					// 304501.ra - Call the post MDB invoke method on XD
    					if (mpc != null && mpc.isActive()) {

    						mpc.postMdbInvoke (this, scheduler);

    					}

    				} finally {

    					afterDelivery(message, endpoint, success);

    				}

    			} finally {


    				// Call isTransactionRolledBack to clear up
    				// the data stored in the SibRaXaResource hashtable even if
    				// we are not using PMI
    				boolean tranRolledBack = isTransactionRolledBack ();

    				// Success should only be true if the endpoint
    				// was successfully invoked and the transaction
    				// was not rolled back.
    				success = success && !tranRolledBack;

    				if (success)
    				{
    					//This message was sucessfully delivered so reset all sequential message counters
    					clearSequentialFailureMessageCount();
    				}
    				else
    				{
    					// Make sure we have a message and the redelivered count exists
    					if ((message != null) && (((JsMessage)message).getRedeliveredCount() != null))
    					{
    						// message failed check its redelivery threshold
    						// we need to check for n - 1 where n is the max failed deliveries. The redeliveredCount means the 
    						// number of REdeliveries so we need to add one to this to align with the maxfaileddeliveries count.
    						if ( (((JsMessage)message).getRedeliveredCount().intValue() + 1) == (_maxFailedDeliveries - 1))
    						{
    							//We are about to hit the threshold
    							addToSequentialFailureMessageCount();
    						}
    					}
    					else
    					{
    						if (TraceComponent.isAnyTracingEnabled() && TRACE.isDebugEnabled()) 
    						{
    							if (null == message)
    							{
    								SibTr.debug(TRACE, "The message was null");
    							}
    							else
    							{
    								SibTr.debug(TRACE, "Unable to obtain the redelivered count from message " + message);
    							}
    						}
    					}
    				}

    			}


    		}

    	} finally {

    		cleanup();

    		try {

    			// If possible, unlock any undelivered messages

    			if (undeliveredDispatchableMessages.size() > 0) {

    				try {

    					// If we delete best effort non persistent messages then
    					// remove any best effort non persitent messages from the
    					// list.
    					if (_deleteUnrecoverableMessages) {

    						for (final Iterator iterator = undeliveredDispatchableMessages
    								.iterator(); iterator.hasNext();) {

    							final DispatchableMessage dispatchableMsg = (DispatchableMessage) iterator.next();
    							final SIBusMessage msg = dispatchableMsg.getMessage();

    							if (_unrecoverableReliability.compareTo(msg.getReliability ()) >= 0) {

    								if (TraceComponent.isAnyTracingEnabled() && TRACE.isDebugEnabled()) {
    									SibTr.debug(this,
    											TRACE,
    											"Removing the following message as it has already been deleted: ",
    											msg);
    								}

    								iterator.remove();

    							}

    						}

    					}

    					// Make sure that we still have some undelivered messages
    					// since they may have all been deleted
    					if (undeliveredDispatchableMessages.size() > 0) {

    						if (TraceComponent.isAnyTracingEnabled() && TRACE.isDebugEnabled()) {
    							SibTr.debug(this, TRACE,
    									"Unlocking undelivered messages:",
    									undeliveredDispatchableMessages);
    						}

    						unlockMessages(undeliveredDispatchableMessages, false);

    					}

    				} catch (final ResourceException exception) {

    					// No FFDC code needed
    					if (TraceComponent.isAnyTracingEnabled() && TRACE.isEventEnabled()) {
    						SibTr.exception(this, TRACE, exception);
    					}

    				}

    			}

    		} finally {

    			if (endpoint != null) {

    				endpoint.release();

    			}
    		}


    	}

    	if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled()) {
    		SibTr.exit(this, TRACE, methodName);
    	}

    }
328318
@Transactional
	public void processMessage(ProcessInstance processInstance,
			Message<?> message, MessageMapper messageMapper) {
		Object context = getContextDAO().getContext(processInstance.getObjectInstance());
		processInstance.setStatus(TaskStatus.GO);
		try {
			messageMapper.unpackMessage(message, context);
		} catch (Exception e) {
			processInstance.setStatus(TaskStatus.ABORTING);
			processInstance.setComment(e.getMessage());
		}
		log.debug("set status to processInstance {} to {} {}",processInstance.getId(),processInstance.getStatus(),context);
		getContextDAO().mergeContext(context);
		getWorkflowDAO().mergeProcessInstance(processInstance);
		getWorkflowDAO().flush();
		if (log.isDebugEnabled()) {
			getWorkflowDAO().getActiveProcesses();
		}
	}
392481
protected Call call( Message message ) throws NoSuchMessageException
    {
        Call call = new Call( message );
        call.addInput( arguments() );
        return call;
    }
852407
public void setBatchPutMessageErrorEntries(java.util.Collection<BatchPutMessageErrorEntry> batchPutMessageErrorEntries) {
        if (batchPutMessageErrorEntries == null) {
            this.batchPutMessageErrorEntries = null;
            return;
        }

        this.batchPutMessageErrorEntries = new java.util.ArrayList<BatchPutMessageErrorEntry>(batchPutMessageErrorEntries);
    }
666534
@Override
	public final void send(Message message) throws JMSException
    {
    	if (this.destination == null)
    		throw new UnsupportedOperationException("Destination was not set at creation time");
    	
    	// Setup message fields
    	setupMessage(destination,message,defaultDeliveryMode,defaultPriority,defaultTimeToLive);
    	
    	// Handle foreign message implementations
        message = MessageTools.normalize(message);

        sendToDestination(destination,false,message,defaultDeliveryMode,defaultPriority,defaultTimeToLive);
    }
579123
public void dispatchMessage (ChatMessage message, String localType)
    {
        setClientInfo(message, localType);
        dispatchPreparedMessage(message);
    }
768951
public final void dispatch( AbstractMessage message ) throws JMSException
    {
        // Security
        LocalConnection conn = (LocalConnection)getConnection();
        if (conn.isSecurityEnabled())
        {
            Destination destination = message.getJMSDestination();
            if (destination instanceof Queue)
            {
                String queueName = ((Queue)destination).getQueueName();
                if (conn.isRegisteredTemporaryQueue(queueName))
                {
                    // OK, temporary destination
                }
                else
                if (queueName.equals(FFMQConstants.ADM_REQUEST_QUEUE))
                {
                    conn.checkPermission(Resource.SERVER, Action.REMOTE_ADMIN);
                }
                else
                if (queueName.equals(FFMQConstants.ADM_REPLY_QUEUE))
                {
                    // Only the internal admin thread can produce on this queue
                    if (conn.getSecurityContext() != null)
                        throw new FFMQException("Access denied to administration queue "+queueName,"ACCESS_DENIED");
                }
                else
                {
                    // Standard queue
                    conn.checkPermission(destination,Action.PRODUCE);
                }
            }
            else
            if (destination instanceof Topic)
            {
                String topicName = ((Topic)destination).getTopicName();
                if (conn.isRegisteredTemporaryTopic(topicName))
                {
                    // OK, temporary destination
                }
                else
                {
                    // Standard topic
                    conn.checkPermission(destination,Action.PRODUCE);   
                }
            }
            else
                throw new InvalidDestinationException("Unsupported destination : "+destination);
        }
        
        if (debugEnabled)
            log.debug(this+" [PUT] in "+message.getJMSDestination()+" - "+message);
        
        externalAccessLock.readLock().lock();
    	try
        {
            checkNotClosed();
            
            pendingPuts.add(message);
            
        	if (!transacted)
        		commitUpdates(false, null, true); // FIXME Async commit ?
        }
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
768146
public void setResourceTags(java.util.Collection<ResourceTag> resourceTags) {
        if (resourceTags == null) {
            this.resourceTags = null;
            return;
        }

        this.resourceTags = new com.amazonaws.internal.SdkInternalList<ResourceTag>(resourceTags);
    }
824076
public void sendMessage(String message)
    throws ProcessException {
        try {
            HttpHelper httpHelper = new HttpHelper(getURL(destination));
            httpHelper.post(message);
        } catch (Exception e) {
            throw new ProcessException(-1, e.getMessage(), e);
        }
    }
564865
public void onMessage(MutateRowsResponse message) {
    for (Entry entry : message.getEntriesList()) {
      int index = (int) entry.getIndex();

      // Sanity check to make sure that the index returned from the server is valid.
      if (index >= mapToOriginalIndex.length || index < 0) {
        messageIsInvalid = true;
        break;
      }

      // Set the result.
      results[mapToOriginalIndex[index]] = entry.getStatus();
    }
  }
227492
@Override
    void addMessagingEngine(JsMessagingEngine messagingEngine) throws ResourceException
    {
        final String methodName = "addMessagingEngine";
        if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled())
        {
            SibTr.entry(this, TRACE, methodName, messagingEngine);
        }
        if ((_destinationStrategy.isNonDurablePubSub()) && (_connections.size() > 0))
        {
            // Ignore for non durable pub sub
            if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled())
            {
                SibTr.exit(this, TRACE, methodName);
            }
            return;
        }
        else
        {
            clearTimer();
            checkMEs(new JsMessagingEngine[] { messagingEngine });
        }

        if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled())
        {
            SibTr.exit(this, TRACE, methodName);
        }
    }
275415
@Override
    public void send(final Object message) throws MessageTransportException {
        if (shutdown.get())
            throw new MessageTransportException("send called on shutdown queue.");

        if (blocking) {
            while (true) {
                try {
                    queue.put(message);
                    if (statsCollector != null)
                        statsCollector.messageSent(message);
                    break;
                } catch (final InterruptedException ie) {
                    if (shutdown.get())
                        throw new MessageTransportException("Shutting down durring send.");
                }
            }
        } else {
            if (!queue.offer(message)) {
                if (statsCollector != null)
                    statsCollector.messageNotSent();
                throw new MessageTransportException("Failed to queue message due to capacity.");
            } else if (statsCollector != null)
                statsCollector.messageSent(message);
        }
    }
1103349
def dispatch(self, message: Message) -> Iterator[Any]:
        
        if "text" in message:
            text = message["text"] or ""
        elif "message" in message:
            text = message["message"].get("text", "")
        else:
            text = ""

        msg_subtype = message.get("subtype")

        for subtype, matchs in itertools.chain(
            self._routes[message["channel"]].items(), self._routes["*"].items()
        ):
            if msg_subtype == subtype or subtype is None:
                for match, endpoints in matchs.items():
                    if match.search(text):
                        yield from endpoints
1364178
@SuppressWarnings("unchecked")
    private void dispatchMessage(ArrayListMultimap<Integer, MessageListener> listenerMap, Message message) {
        List<MessageListener> messageListeners;

        // If the message is a Channel Message, hit the channel based listeners first.
        if (message instanceof ChannelMessage) {
            messageListeners = listenerMap.get(((ChannelMessage) message).getChannelInt());
            if (messageListeners != null) {
                for (MessageListener listener : messageListeners) {
                    listener.messageReceived(message);
                }
            }
        }

        // Then send the message to any listeners registered to all channels
        //   or to listeners for messages that do not support channels.
        messageListeners = listenerMap.get(null);
        if (messageListeners != null) {
            for (MessageListener listener : messageListeners) {
                listener.messageReceived(message);
            }
        }
    }
502158
@Override
	public final void send(Message message, int deliveryMode, int priority, long timeToLive) throws JMSException
    {
    	if (this.destination == null)
    		throw new UnsupportedOperationException("Destination was not set at creation time");

        // Setup message fields
    	setupMessage(destination,message,deliveryMode,priority,timeToLive);
    	
    	// Handle foreign message implementations
        message = MessageTools.normalize(message);
        
    	sendToDestination(destination,false,message,deliveryMode,priority,timeToLive);
    }
23574
public synchronized void changeEndpoints(Endpoints endpoints) {
        this.endpoints = endpoints;
        this.apiBroker.send(new ApiBroker.ChangeEndpoints(endpoints));
    }
1024499
public void setResourceServers(java.util.Collection<ResourceServerType> resourceServers) {
        if (resourceServers == null) {
            this.resourceServers = null;
            return;
        }

        this.resourceServers = new java.util.ArrayList<ResourceServerType>(resourceServers);
    }