Associates a message processor with this instance.
<p>
The message processor cannot be started before. This method will start it.
</p>
<p>
This method must be invoked only once.
</p>
@param messageProcessor the message processor
329440
public void associateMessageProcessor( AbstractMessageProcessor<T> messageProcessor ) {

		if( this.messageProcessor != null )
			throw new IllegalArgumentException( "The message processor was already defined." );

		this.messageProcessor = messageProcessor;
		configureMessageProcessor( messageProcessor );
		this.messageProcessor.start();
	}
353553
void init(TransformerFactoryImpl processor)
  {
    m_stylesheetProcessor = processor;

    // Set the initial content handler.
    m_processors.push(m_schema.getElementProcessor());
    this.pushNewNamespaceSupport();

    // m_includeStack.push(SystemIDResolver.getAbsoluteURI(this.getBaseIdentifier(), null));
    // initXPath(processor, null);
  }
94504
void processMsgs(LockedMessageEnumeration msgEnumeration, ConsumerSession consumerSession)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "processMsgs", new Object[] { msgEnumeration, consumerSession });      

    // Remember that a callback is running
    asynchConsumerRunning = true;

    try
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "Entering asynchConsumerCallback.consumeMessages",
                     new Object[] {asynchConsumerCallback, msgEnumeration});      

      //Call the consumeMessages method on the registered
      //AsynchConsumerCallback object
      asynchConsumerCallback.consumeMessages(msgEnumeration);

      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "Exiting asynchConsumerCallback.consumeMessages");      
    }
    catch (Throwable e)
    {
      //Catch any exceptions thrown by consumeMessages.
      FFDCFilter.processException(
       e,
       "com.ibm.ws.sib.processor.impl.AsynchConsumer.processMsgs",
       "1:124:1.37",
       this);
       
      //Notify the consumer that this exception occurred.
      if(consumerSession!=null)
      {
        try
        {
          //Notify the consumer that this exception occurred.
          notifyExceptionListeners(e, consumerSession);
        }
        catch(Exception connectionClosed)
        {
          //No FFDC code needed
        }
            
      }
       
     //Trace the exception but otherwise swallow it since it was not
      //a failure in MP code or code upon which the MP is critically
      //dependent.
      if (e instanceof Exception)
        SibTr.exception(tc, (Exception)e);
      
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "Exception occurred in consumeMessages " + e);
         
      // We're not allowed to swallow this exception - let it work its way back to
      // the threadpool
      if(e instanceof ThreadDeath)
      {     
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
          SibTr.exit(tc, "processMsgs", e); 
        throw (ThreadDeath)e;
      }
        
    }
    finally
    {
      asynchConsumerRunning = false;
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "processMsgs");
  }
1291326
public void executeProcessors(JCas jcas, ProcessorManager.Priority prio) {
		if(!this.initialized) {
			Logger.printError(component, "Unable to execute Processors; initialization was not concluded successfully.");
			System.exit(-1);
		}
		
		LinkedList<GenericProcessor> myList = processors.get(prio);
		for(GenericProcessor gp : myList) {
			try {
				gp.process(jcas);
			} catch (Exception exception) {
				exception.printStackTrace();
				Logger.printError(component, "Unable to process registered Processor " + gp.getClass().getName() + ", got: " + exception.toString());
				System.exit(-1);
			}
		}
	}
99399
public EventHandlerGroup<T> and(final EventProcessor... processors)
    {
        Sequence[] combinedSequences = new Sequence[sequences.length + processors.length];

        for (int i = 0; i < processors.length; i++)
        {
            consumerRepository.add(processors[i]);
            combinedSequences[i] = processors[i].getSequence();
        }
        System.arraycopy(sequences, 0, combinedSequences, processors.length, sequences.length);

        return new EventHandlerGroup<>(disruptor, consumerRepository, combinedSequences);
    }
517176
public void forEachFlushedMessage(MessageProcessor processor) throws Exception {
        if (processor == null) {
            throw new NullPointerException("processor");
        }

        Entry entry = flushedEntry;
        if (entry == null) {
            return;
        }

        do {
            if (!entry.cancelled) {
                if (!processor.processMessage(entry.msg)) {
                    return;
                }
            }
            entry = entry.next;
        } while (isFlushedEntry(entry));
    }
23868
private void startProcessingRequest(final T data) {
        Logger.v("Task[%s] start processing request", this);

        if (mExecutor != null) {
            final ProcessingPrioritizable<T> prioritizable = new ProcessingPrioritizable<T>(this, data);
            final ProcessingRequest<T> request = new ProcessingRequest<T>(prioritizable, mPriority.ordinal(), this);
            mExecutor.executeProcessingRequest(request);
        } else {
            throw new IllegalStateException(Messages.NO_EXECUTOR);
        }
    }
114865
static ClassProcessor createProcessor(
            final ClassLoader classLoader,
            final String processorClassName)
            throws MojoExecutionException {
        try {
            final Object classProcessor = classLoader.loadClass(processorClassName).newInstance();
            if (classProcessor instanceof ClassProcessor) {
                return (ClassProcessor) classProcessor;
            }
            throw new MojoExecutionException("Invalid class processor: " + processorClassName);
        } catch (final ClassNotFoundException e) {
            throw new MojoExecutionException("Class processor not found: " + processorClassName, e);
            // CHECKSTYLE.OFF: IllegalCatch
        } catch (final Exception e) {
            // CHECKSTYLE.ON: IllegalCatch
            throw new MojoExecutionException("Unable to instantiate processor: " + processorClassName, e);
        }
    }
830327
public void registerJsonBeanProcessor( Class target, JsonBeanProcessor jsonBeanProcessor ) {
      if( target != null && jsonBeanProcessor != null ) {
         beanProcessorMap.put( target, jsonBeanProcessor );
      }
   }
554916
protected void setProcessor(final Processor processor) throws BuildException, NullPointerException {
    if (processor == null) {
      throw new NullPointerException("processor");
    }
    if (isReference()) {
      throw super.tooManyAttributes();
    }
    if (this.env == null && !this.newEnvironment) {
      this.processor = processor;
    } else {
      this.processor = processor.changeEnvironment(this.newEnvironment, this.env);
    }
  }
816773
private void startupProcessor() {
        Processor processor = processorRef.get();
        if (processor == null) {
            processor = new Processor();
            if (processorRef.compareAndSet(null, processor)) {
                executor.execute(new NamePreservingRunnable(processor,
                        threadName));
            }
        }

        // Just stop the select() and start it again, so that the processor
        // can be activated immediately. 
        wakeup();
    }
1229247
void addMessageHandler(SubscriptionMessageHandler messageHandler)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "addMessageHandler", messageHandler);

    final boolean inserted = _subscriptionMessagePool.add(messageHandler);
    // If the message wasn't inserted, then the pool was exceeded
    if (!inserted)
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(
          tc,
         "SubscriptionObjectPool size exceeded");

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "addMessageHandler");
  }
528609
public function addProcessor(ProcessorInterface $processor)
    {
        $this->processors[] = $processor;
        $this->logDebug('Added processor: ' . get_class($processor));
    }
106755
protected void addConfiguredProcessorParam(final ProcessorParam param) throws NullPointerException, BuildException {
    if (param == null) {
      throw new NullPointerException("param");
    }
    if (isReference()) {
      throw noChildrenAllowed();
    }
    this.processorParams.addElement(param);
  }
755795
public void setDefaultMessageProcessor(Map<String,Object> mapHeaderMessageInfo)
    {
        if (mapHeaderMessageInfo.get(TrxMessageHeader.MESSAGE_PROCESSOR_CLASS) == null)
        {
            String strMessageInfoTypeCode = (String)mapHeaderMessageInfo.get(TrxMessageHeader.MESSAGE_INFO_TYPE);
            String strMessageProcessTypeCode = (String)mapHeaderMessageInfo.get(TrxMessageHeader.MESSAGE_PROCESS_TYPE);
            if ((strMessageInfoTypeCode != null) && (strMessageProcessTypeCode != null))
            {
                String strProcessorClass = null;
                if ((MessageInfoType.REQUEST.equals(strMessageInfoTypeCode)) && (MessageType.MESSAGE_OUT.equals(strMessageProcessTypeCode)))
                    strProcessorClass = BaseMessageOutProcessor.class.getName();
                else if ((MessageInfoType.REQUEST.equals(strMessageInfoTypeCode)) && (MessageType.MESSAGE_IN.equals(strMessageProcessTypeCode)))
                    strProcessorClass = BaseMessageInProcessor.class.getName();
                else if ((MessageInfoType.REPLY.equals(strMessageInfoTypeCode)) && (MessageType.MESSAGE_OUT.equals(strMessageProcessTypeCode)))
                    strProcessorClass = BaseMessageReplyOutProcessor.class.getName();
                if ((MessageInfoType.REPLY.equals(strMessageInfoTypeCode)) && (MessageType.MESSAGE_IN.equals(strMessageProcessTypeCode)))
                    strProcessorClass = BaseMessageReplyInProcessor.class.getName();
                if (strProcessorClass != null)
                    mapHeaderMessageInfo.put(TrxMessageHeader.MESSAGE_PROCESSOR_CLASS, strProcessorClass);
            }
        }
    }
931292
public void addProcessorExecution(ProcessorExecution status) {
        lock.lock();
        try {
            processorExecutions.add(status);
        } finally {
            lock.unlock();
        }
    }
218789
public List<ConfigMessage> init(Processor processor) {
        List<ConfigMessage> configMessages = new ArrayList<>();

        if (action.getPredicate() != null) {
            try {
                predicate = ExpressionHandlerFactory.getHandler(action.getPredicate());

                predicate.init(processor, getAction(), true);

                if (!isUsesHeaders()) {
                    setUsesHeaders(predicate.isUsesHeaders());
                }
                if (!isUsesContent()) {
                    setUsesContent(predicate.isUsesContent());
                }

            } catch (Throwable t) {
                log.severe("Failed to initialise predicate for action:" + action, t);
                ConfigMessage configMessage = new ConfigMessage();
                configMessage.setSeverity(Severity.Error);
                configMessage.setMessage(t.getMessage());
                configMessage.setProcessor(processor.getDescription());
                configMessage.setAction(action.getDescription());
                configMessages.add(configMessage);
            }
        }

        return configMessages;
    }