Creates a processing item with a specific processor and paralellism level.

@param processor
@param parallelism
@return ProcessingItem
13968
private ProcessingItem createPi(Processor processor, int parallelism) {
        ProcessingItem pi = this.componentFactory.createPi(processor, parallelism);
        this.topology.addProcessingItem(pi, parallelism);
        return pi;
    }
1128920
public <IN, OUT> Processor<IN, OUT> newAsyncProcessor(Function<IN, OUT> processingFunction) {
        return newAsyncProcessor(processingFunction, scheduler, batchSize);
    }
137202
private CompletableFuture<ProcessorCollection> getProcessor(long streamSegmentId) {
        ProcessorCollection existingProcessor = this.processors.getOrDefault(streamSegmentId, null);
        if (existingProcessor != null) {
            if (closeIfNecessary(existingProcessor).isClosed()) {
                // Existing SegmentAggregator has become stale (most likely due to its SegmentMetadata being evicted),
                // so it has been closed and we need to create a new one.
                this.processors.remove(streamSegmentId);
            } else {
                return CompletableFuture.completedFuture(existingProcessor);
            }
        }

        // Get the SegmentAggregator's Metadata.
        UpdateableSegmentMetadata segmentMetadata = this.dataSource.getStreamSegmentMetadata(streamSegmentId);
        if (segmentMetadata == null) {
            return Futures.failedFuture(new DataCorruptionException(String.format(
                    "No StreamSegment with id '%d' is registered in the metadata.", streamSegmentId)));
        }

        // Then create the aggregator, and only register it after a successful initialization. Otherwise we risk
        // having a registered aggregator that is not initialized.
        SegmentAggregator newAggregator = new SegmentAggregator(segmentMetadata, this.dataSource, this.storage, this.config, this.timer, this.executor);
        ProcessorCollection pc = new ProcessorCollection(newAggregator, this.createProcessors.apply(segmentMetadata));
        try {
            CompletableFuture<Void> init = newAggregator.initialize(this.config.getFlushTimeout());
            Futures.exceptionListener(init, ex -> newAggregator.close());
            return init.thenApply(ignored -> {
                this.processors.put(streamSegmentId, pc);
                return pc;
            });
        } catch (Exception ex) {
            pc.close();
            throw ex;
        }
    }
316159
public static LazyReact parallelBuilder(final int parallelism) {
        return LazyReact.builder()
                        .executor(Executors.newFixedThreadPool(parallelism))
                        .build();
    }
466154
public <A extends Annotation> void registerProcessor(final Class<A> annoClass, final FieldProcessor<A> processor) {
        ArgUtils.notNull(annoClass, "annoClass");
        ArgUtils.notNull(processor, "processor");
        
        pocessorMap.put(annoClass, processor);
        
    }
128841
public ParallelTaskBuilder setHttpPollerProcessor(
            HttpPollerProcessor httpPollerProcessor) {
        this.httpMeta.setHttpPollerProcessor(httpPollerProcessor);
        this.httpMeta.setPollable(true);
        return this;
    }
1271497
public void setParallelism(int parallelism) {
		Preconditions.checkArgument(
				parallelism > 0 || parallelism == ExecutionConfig.PARALLELISM_DEFAULT,
				"The parallelism must be at least one, or ExecutionConfig.PARALLELISM_DEFAULT (use system default).");
		this.parallelism = parallelism;
	}
263719
public static NiftyProcessorFactory factoryFromTProcessor(final TProcessor standardThriftProcessor)
    {
        checkProcessMethodSignature();

        return new NiftyProcessorFactory()
        {
            @Override
            public NiftyProcessor getProcessor(TTransport transport)
            {
                return processorFromTProcessor(standardThriftProcessor);
            }
        };
    }
1199507
private Object invokeProcessorsFor(Object value, ResolvableType type) {

		Object currentValue = value;

		// Process actual value
		for (RepresentationModelProcessorInvoker.ProcessorWrapper wrapper : this.processors) {
			if (wrapper.supports(type, currentValue)) {
				currentValue = wrapper.invokeProcessor(currentValue);
			}
		}

		return currentValue;
	}
808002
@Override
    public Consumer createConsumer(Processor processor) throws Exception {
        LOG.debug(this.toString());

        if (direction == null || (direction.compareTo("") == 0)) {
            direction = "DIR_INPUT_ONLY";
        }
        if (mode == null || (mode.compareTo("") == 0)) {
            mode = "MODE_INPUT_PULL_UP";
        }
        if (trigger == null || (trigger.compareTo("") == 0)) {
            trigger = "TRIGGER_BOTH_EDGES";
        }

        int internalDirection = internalValueWithOR(direction);
        int internalMode = internalValueWithOR(mode);
        int internalTrigger = internalValueWithOR(trigger);

        GPIOPinConfig pinConfig = new GPIOPinConfig(0, Integer.parseInt(this.gpioId), internalDirection, internalMode,
                internalTrigger, state);

        GPIOPin pin = DeviceManager.open(GPIOPin.class, pinConfig);
        return new GPIOConsumer(this, processor, pin);
    }
1372096
def add_processor(self, processor_instance: Processor, priority=0) -> None:
        
        assert issubclass(processor_instance.__class__, Processor)
        processor_instance.priority = priority
        processor_instance.world = self
        self._processors.append(processor_instance)
        self._processors.sort(key=lambda proc: proc.priority, reverse=True)
331714
private Object processItem(Object itemRead) {
		logger.entering(sourceClass, "processItem", itemRead);
		Object processedItem = null;

		// if no processor defined for this chunk
		if (processorProxy == null){
			return itemRead;
		}

		try {

			// call process listeners before and after the actual process call
			for (ItemProcessListenerProxy processListenerProxy : itemProcessListeners) {
				processListenerProxy.beforeProcess(itemRead);
			}

			processedItem = processorProxy.processItem(itemRead);

			if (processedItem == null) {
				currentItemStatus.setFiltered(true);
			}

			for (ItemProcessListenerProxy processListenerProxy : itemProcessListeners) {
				processListenerProxy.afterProcess(itemRead, processedItem);
			}
		} catch (Exception e) {
			for (ItemProcessListenerProxy processListenerProxy : itemProcessListeners) {
				processListenerProxy.onProcessError(itemRead, e);
			}
			if(!currentChunkStatus.isRetryingAfterRollback()) {
				if (retryProcessException(e, itemRead)) {
					if (!retryHandler.isRollbackException(e)) {
						processedItem = processItem(itemRead);
					} else {
						currentChunkStatus.markForRollbackWithRetry(e);
					}
				}
				else if (skipProcessException(e, itemRead)) {
					currentItemStatus.setSkipped(true);
					stepContext.getMetric(MetricImpl.MetricType.PROCESS_SKIP_COUNT).incValue();
				}
				else {
					throw new BatchContainerRuntimeException(e);
				}
			}
			else {
				if (skipProcessException(e, itemRead)) {
					currentItemStatus.setSkipped(true);
					stepContext.getMetric(MetricImpl.MetricType.PROCESS_SKIP_COUNT).incValue();
				} else if (retryProcessException(e, itemRead)) {

					if (!retryHandler.isRollbackException(e)) {
						// retry without rollback
						processedItem = processItem(itemRead);
					} else {
						// retry with rollback
						currentChunkStatus.markForRollbackWithRetry(e);
					}
				} else {
					throw new BatchContainerRuntimeException(e);
				}
			}

		} catch (Throwable e) {
			throw new BatchContainerRuntimeException(e);
		}

		logger.exiting(sourceClass, "processItem", processedItem==null ? "<null>" : processedItem);
		return processedItem;
	}
662783
@Override
	public Processor newProcessor(Processor p) {
		AMRDefaultRuleProcessor oldProcessor = (AMRDefaultRuleProcessor) p;
		Builder builder = new Builder(oldProcessor);
		AMRDefaultRuleProcessor newProcessor = builder.build();
		newProcessor.resultStream = oldProcessor.resultStream;
		newProcessor.ruleStream = oldProcessor.ruleStream;
		return newProcessor;
	}
1364529
@Override
	public Processor newProcessor(Processor sourceProcessor) {
		LocalLearnerProcessor newProcessor = new LocalLearnerProcessor();
		LocalLearnerProcessor originProcessor = (LocalLearnerProcessor) sourceProcessor;

		if (originProcessor.getLearner() != null){
			newProcessor.setLearner(originProcessor.getLearner().create());
		}

		if (originProcessor.getChangeDetector() != null){
				newProcessor.setChangeDetector(originProcessor.getChangeDetector());
		}

		newProcessor.setOutputStream(originProcessor.getOutputStream());
		return newProcessor;
	}
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);
        }
    }
1303166
public static IElementProcessor unwrap(final IElementProcessor processor) {
        if (processor == null) {
            return null;
        }
        if (processor instanceof AbstractProcessorWrapper) {
            return (IElementProcessor)((AbstractProcessorWrapper) processor).unwrap();
        }
        return processor;
    }
98281
public static MessageBatcher createBatcher(String name,
			MessageProcessor processor) {
		MessageBatcher batcher = batcherMap.get(name);
		if (batcher == null) {
			synchronized (BatcherFactory.class) {
				batcher = batcherMap.get(name);
				if (batcher == null) {
					batcher = new MessageBatcher(name, processor);
					batcherMap.put(name, batcher);
				}
			}
		}
		return batcher;
	}