in core/camel-management/src/main/java/org/apache/camel/management/DefaultManagementObjectStrategy.java [317:507]
public Object getManagedObjectForProcessor(
CamelContext context, Processor processor,
NamedNode node, Route route) {
ManagedProcessor answer = null;
ProcessorDefinition<?> definition = (ProcessorDefinition<?>) node;
if (definition instanceof RecipientListDefinition) {
// special for RecipientListDefinition, as the processor is wrapped in a pipeline as last
Pipeline pipeline = (Pipeline) processor;
for (Processor value : pipeline.next()) {
processor = value;
}
}
// unwrap delegates as we want the real target processor
Processor target = processor;
while (target != null) {
// skip error handlers
if (target instanceof ErrorHandler) {
return false;
}
if (target instanceof ConvertBodyProcessor) {
answer = new ManagedConvertBody(context, (ConvertBodyProcessor) target, definition);
} else if (target instanceof ConvertHeaderProcessor) {
answer = new ManagedConvertHeader(context, (ConvertHeaderProcessor) target, definition);
} else if (target instanceof ConvertVariableProcessor) {
answer = new ManagedConvertVariable(context, (ConvertVariableProcessor) target, definition);
} else if (target instanceof ChoiceProcessor) {
answer = new ManagedChoice(context, (ChoiceProcessor) target, definition);
} else if (target instanceof ClaimCheckProcessor) {
answer = new ManagedClaimCheck(context, (ClaimCheckProcessor) target, definition);
} else if (target instanceof Delayer) {
answer = new ManagedDelayer(context, (Delayer) target, definition);
} else if (target instanceof DisabledProcessor) {
answer = new ManagedDisabled(context, (DisabledProcessor) target, definition);
} else if (target instanceof TryProcessor tryProc) {
answer = new ManagedDoTry(context, tryProc, cast(definition));
} else if (target instanceof CatchProcessor) {
answer = new ManagedDoCatch(context, (CatchProcessor) target, cast(definition));
} else if (target instanceof FinallyProcessor) {
answer = new ManagedDoFinally(context, (FinallyProcessor) target, cast(definition));
} else if (target instanceof Throttler) {
answer = new ManagedThrottler(context, (Throttler) target, definition);
} else if (target instanceof DynamicRouter) {
answer = new ManagedDynamicRouter(context, (DynamicRouter) target, cast(definition));
} else if (target instanceof RoutingSlip) {
answer = new ManagedRoutingSlip(context, (RoutingSlip) target, cast(definition));
} else if (target instanceof FilterProcessor) {
answer = new ManagedFilter(context, (FilterProcessor) target, (ExpressionNode) definition);
} else if (target instanceof LogProcessor) {
answer = new ManagedLog(context, (LogProcessor) target, definition);
} else if (target instanceof LoopProcessor) {
answer = new ManagedLoop(context, (LoopProcessor) target, cast(definition));
} else if (target instanceof MarshalProcessor) {
answer = new ManagedMarshal(context, (MarshalProcessor) target, cast(definition));
} else if (target instanceof UnmarshalProcessor) {
answer = new ManagedUnmarshal(context, (UnmarshalProcessor) target, cast(definition));
} else if (target instanceof FailOverLoadBalancer) {
answer = new ManagedFailoverLoadBalancer(
context, (FailOverLoadBalancer) target, cast(definition));
} else if (target instanceof RandomLoadBalancer) {
answer = new ManagedRandomLoadBalancer(
context, (RandomLoadBalancer) target, cast(definition));
} else if (target instanceof RoundRobinLoadBalancer) {
answer = new ManagedRoundRobinLoadBalancer(
context, (RoundRobinLoadBalancer) target, cast(definition));
} else if (target instanceof StickyLoadBalancer) {
answer = new ManagedStickyLoadBalancer(
context, (StickyLoadBalancer) target, cast(definition));
} else if (target instanceof TopicLoadBalancer) {
answer = new ManagedTopicLoadBalancer(context, (TopicLoadBalancer) target, cast(definition));
} else if (target instanceof WeightedLoadBalancer) {
answer = new ManagedWeightedLoadBalancer(
context, (WeightedLoadBalancer) target, cast(definition));
} else if (target instanceof RecipientList) {
answer = new ManagedRecipientList(context, (RecipientList) target, cast(definition));
} else if (target instanceof Splitter) {
answer = new ManagedSplitter(context, (Splitter) target, cast(definition));
} else if (target instanceof MulticastProcessor) {
answer = new ManagedMulticast(context, (MulticastProcessor) target, definition);
} else if (target instanceof SamplingThrottler) {
answer = new ManagedSamplingThrottler(context, (SamplingThrottler) target, definition);
} else if (target instanceof Resequencer) {
answer = new ManagedResequencer(context, (Resequencer) target, definition);
} else if (target instanceof RollbackProcessor) {
answer = new ManagedRollback(context, (RollbackProcessor) target, definition);
} else if (target instanceof StreamResequencer) {
answer = new ManagedResequencer(context, (StreamResequencer) target, definition);
} else if (target instanceof SetBodyProcessor) {
answer = new ManagedSetBody(context, (SetBodyProcessor) target, cast(definition));
} else if (target instanceof RemoveHeaderProcessor) {
answer = new ManagedRemoveHeader(context, (RemoveHeaderProcessor) target, definition);
} else if (target instanceof RemoveHeadersProcessor) {
answer = new ManagedRemoveHeaders(context, (RemoveHeadersProcessor) target, definition);
} else if (target instanceof SetHeaderProcessor) {
answer = new ManagedSetHeader(context, (SetHeaderProcessor) target, cast(definition));
} else if (target instanceof SetHeadersProcessor) {
answer = new ManagedSetHeaders(context, (SetHeadersProcessor) target, cast(definition));
} else if (target instanceof SetVariableProcessor) {
answer = new ManagedSetVariable(context, (SetVariableProcessor) target, cast(definition));
} else if (target instanceof SetVariablesProcessor) {
answer = new ManagedSetVariables(context, (SetVariablesProcessor) target, cast(definition));
} else if (target instanceof RemovePropertyProcessor) {
answer = new ManagedRemoveProperty(context, (RemovePropertyProcessor) target, definition);
} else if (target instanceof RemovePropertiesProcessor) {
answer = new ManagedRemoveProperties(context, (RemovePropertiesProcessor) target, definition);
} else if (target instanceof RemoveVariableProcessor) {
answer = new ManagedRemoveVariable(context, (RemoveVariableProcessor) target, definition);
} else if (target instanceof SetPropertyProcessor) {
answer = new ManagedSetProperty(context, (SetPropertyProcessor) target, cast(definition));
} else if (target instanceof ExchangePatternProcessor) {
answer = new ManagedSetExchangePattern(context, (ExchangePatternProcessor) target, definition);
} else if (target instanceof ScriptProcessor) {
answer = new ManagedScript(context, (ScriptProcessor) target, cast(definition));
} else if (target instanceof StepProcessor) {
answer = new ManagedStep(context, (StepProcessor) target, definition);
} else if (target instanceof StopProcessor) {
answer = new ManagedStop(context, (StopProcessor) target, definition);
} else if (target instanceof ThreadsProcessor) {
answer = new ManagedThreads(context, (ThreadsProcessor) target, definition);
} else if (target instanceof ThrowExceptionProcessor) {
answer = new ManagedThrowException(context, (ThrowExceptionProcessor) target, definition);
} else if (target instanceof TransformProcessor) {
answer = new ManagedTransformer(context, target, cast(definition));
} else if (target instanceof DataTypeProcessor && definition instanceof TransformDefinition) {
answer = new ManagedTransformer(context, target, (TransformDefinition) definition);
} else if (target instanceof PredicateValidatingProcessor) {
answer = new ManagedValidate(context, (PredicateValidatingProcessor) target, cast(definition));
} else if (target instanceof WireTapProcessor) {
answer = new ManagedWireTapProcessor(context, (WireTapProcessor) target, definition);
} else if (target instanceof SendDynamicProcessor) {
answer = new ManagedSendDynamicProcessor(context, (SendDynamicProcessor) target, definition);
} else if (target instanceof SendProcessor sp) {
// special for sending to throughput logger
if (sp.getDestination() instanceof LogEndpoint le && le.getLogger() instanceof ThroughputLogger tl) {
answer = new ManagedThroughputLogger(context, tl, definition);
}
// regular send processor
if (answer == null) {
answer = new ManagedSendProcessor(context, (SendProcessor) target, definition);
}
} else if (target instanceof BeanProcessor) {
answer = new ManagedBeanProcessor(context, (BeanProcessor) target, definition);
} else if (target instanceof IdempotentConsumer) {
answer = new ManagedIdempotentConsumer(
context, (IdempotentConsumer) target, cast(definition));
} else if (target instanceof AggregateProcessor) {
answer = new ManagedAggregateProcessor(context, (AggregateProcessor) target, cast(definition));
} else if (target instanceof Enricher) {
answer = new ManagedEnricher(context, (Enricher) target, cast(definition));
} else if (target instanceof PollProcessor) {
answer = new ManagedPoll(context, (PollProcessor) target, cast(definition));
} else if (target instanceof PollEnricher) {
answer = new ManagedPollEnricher(context, (PollEnricher) target, cast(definition));
}
// special for custom load balancer
if (definition instanceof LoadBalanceDefinition lb) {
if (lb.getLoadBalancerType() instanceof CustomLoadBalancerDefinition) {
answer = new ManagedCustomLoadBalancer(context, (LoadBalancer) target, (LoadBalanceDefinition) definition);
}
}
if (answer != null) {
// break out as we found an answer
break;
}
// no answer yet, so unwrap any delegates and try again
if (target instanceof DelegateProcessor) {
target = ((DelegateProcessor) target).getProcessor();
} else {
// no delegate so we dont have any target to try next
break;
}
}
if (answer == null && definition instanceof ProcessDefinition) {
answer = new ManagedProcess(context, target, (ProcessDefinition) definition);
} else if (answer == null) {
// fallback to a generic processor
answer = new ManagedProcessor(context, target, definition);
}
answer.setRoute(route);
answer.init(context.getManagementStrategy());
return answer;
}