001/* 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.camel.management; 018 019import java.util.concurrent.ThreadPoolExecutor; 020 021import org.apache.camel.CamelContext; 022import org.apache.camel.Component; 023import org.apache.camel.Consumer; 024import org.apache.camel.DelegateProcessor; 025import org.apache.camel.Endpoint; 026import org.apache.camel.NamedNode; 027import org.apache.camel.Processor; 028import org.apache.camel.Producer; 029import org.apache.camel.Route; 030import org.apache.camel.Service; 031import org.apache.camel.cluster.CamelClusterService; 032import org.apache.camel.component.bean.BeanProcessor; 033import org.apache.camel.component.log.LogEndpoint; 034import org.apache.camel.health.HealthCheckRegistry; 035import org.apache.camel.management.mbean.ManagedAggregateProcessor; 036import org.apache.camel.management.mbean.ManagedBeanProcessor; 037import org.apache.camel.management.mbean.ManagedBrowsableEndpoint; 038import org.apache.camel.management.mbean.ManagedCamelContext; 039import org.apache.camel.management.mbean.ManagedCamelHealth; 040import org.apache.camel.management.mbean.ManagedChoice; 041import org.apache.camel.management.mbean.ManagedClaimCheck; 042import org.apache.camel.management.mbean.ManagedClusterService; 043import org.apache.camel.management.mbean.ManagedComponent; 044import org.apache.camel.management.mbean.ManagedConsumer; 045import org.apache.camel.management.mbean.ManagedConvertBody; 046import org.apache.camel.management.mbean.ManagedConvertHeader; 047import org.apache.camel.management.mbean.ManagedConvertVariable; 048import org.apache.camel.management.mbean.ManagedCustomLoadBalancer; 049import org.apache.camel.management.mbean.ManagedDataFormat; 050import org.apache.camel.management.mbean.ManagedDataTypeTransformer; 051import org.apache.camel.management.mbean.ManagedDelayer; 052import org.apache.camel.management.mbean.ManagedDisabled; 053import org.apache.camel.management.mbean.ManagedDoCatch; 054import org.apache.camel.management.mbean.ManagedDoFinally; 055import org.apache.camel.management.mbean.ManagedDoTry; 056import org.apache.camel.management.mbean.ManagedDynamicRouter; 057import org.apache.camel.management.mbean.ManagedEndpoint; 058import org.apache.camel.management.mbean.ManagedEnricher; 059import org.apache.camel.management.mbean.ManagedEventNotifier; 060import org.apache.camel.management.mbean.ManagedFailoverLoadBalancer; 061import org.apache.camel.management.mbean.ManagedFilter; 062import org.apache.camel.management.mbean.ManagedIdempotentConsumer; 063import org.apache.camel.management.mbean.ManagedLog; 064import org.apache.camel.management.mbean.ManagedLoop; 065import org.apache.camel.management.mbean.ManagedMarshal; 066import org.apache.camel.management.mbean.ManagedMulticast; 067import org.apache.camel.management.mbean.ManagedPoll; 068import org.apache.camel.management.mbean.ManagedPollEnricher; 069import org.apache.camel.management.mbean.ManagedProcess; 070import org.apache.camel.management.mbean.ManagedProcessor; 071import org.apache.camel.management.mbean.ManagedProducer; 072import org.apache.camel.management.mbean.ManagedRandomLoadBalancer; 073import org.apache.camel.management.mbean.ManagedRecipientList; 074import org.apache.camel.management.mbean.ManagedRemoveHeader; 075import org.apache.camel.management.mbean.ManagedRemoveHeaders; 076import org.apache.camel.management.mbean.ManagedRemoveProperties; 077import org.apache.camel.management.mbean.ManagedRemoveProperty; 078import org.apache.camel.management.mbean.ManagedRemoveVariable; 079import org.apache.camel.management.mbean.ManagedResequencer; 080import org.apache.camel.management.mbean.ManagedRollback; 081import org.apache.camel.management.mbean.ManagedRoundRobinLoadBalancer; 082import org.apache.camel.management.mbean.ManagedRoute; 083import org.apache.camel.management.mbean.ManagedRouteController; 084import org.apache.camel.management.mbean.ManagedRouteGroup; 085import org.apache.camel.management.mbean.ManagedRoutingSlip; 086import org.apache.camel.management.mbean.ManagedSamplingThrottler; 087import org.apache.camel.management.mbean.ManagedScheduledPollConsumer; 088import org.apache.camel.management.mbean.ManagedScript; 089import org.apache.camel.management.mbean.ManagedSendDynamicProcessor; 090import org.apache.camel.management.mbean.ManagedSendProcessor; 091import org.apache.camel.management.mbean.ManagedService; 092import org.apache.camel.management.mbean.ManagedSetBody; 093import org.apache.camel.management.mbean.ManagedSetExchangePattern; 094import org.apache.camel.management.mbean.ManagedSetHeader; 095import org.apache.camel.management.mbean.ManagedSetHeaders; 096import org.apache.camel.management.mbean.ManagedSetProperty; 097import org.apache.camel.management.mbean.ManagedSetVariable; 098import org.apache.camel.management.mbean.ManagedSetVariables; 099import org.apache.camel.management.mbean.ManagedSplitter; 100import org.apache.camel.management.mbean.ManagedStep; 101import org.apache.camel.management.mbean.ManagedStickyLoadBalancer; 102import org.apache.camel.management.mbean.ManagedStop; 103import org.apache.camel.management.mbean.ManagedSupervisingRouteController; 104import org.apache.camel.management.mbean.ManagedSuspendableRoute; 105import org.apache.camel.management.mbean.ManagedThreadPool; 106import org.apache.camel.management.mbean.ManagedThreads; 107import org.apache.camel.management.mbean.ManagedThrottler; 108import org.apache.camel.management.mbean.ManagedThroughputLogger; 109import org.apache.camel.management.mbean.ManagedThrowException; 110import org.apache.camel.management.mbean.ManagedTopicLoadBalancer; 111import org.apache.camel.management.mbean.ManagedTransformer; 112import org.apache.camel.management.mbean.ManagedUnmarshal; 113import org.apache.camel.management.mbean.ManagedValidate; 114import org.apache.camel.management.mbean.ManagedWeightedLoadBalancer; 115import org.apache.camel.management.mbean.ManagedWireTapProcessor; 116import org.apache.camel.model.ExpressionNode; 117import org.apache.camel.model.LoadBalanceDefinition; 118import org.apache.camel.model.ProcessDefinition; 119import org.apache.camel.model.ProcessorDefinition; 120import org.apache.camel.model.RecipientListDefinition; 121import org.apache.camel.model.loadbalancer.CustomLoadBalancerDefinition; 122import org.apache.camel.processor.CatchProcessor; 123import org.apache.camel.processor.ChoiceProcessor; 124import org.apache.camel.processor.ClaimCheckProcessor; 125import org.apache.camel.processor.Delayer; 126import org.apache.camel.processor.DisabledProcessor; 127import org.apache.camel.processor.DynamicRouter; 128import org.apache.camel.processor.Enricher; 129import org.apache.camel.processor.ExchangePatternProcessor; 130import org.apache.camel.processor.FilterProcessor; 131import org.apache.camel.processor.FinallyProcessor; 132import org.apache.camel.processor.LogProcessor; 133import org.apache.camel.processor.LoopProcessor; 134import org.apache.camel.processor.MulticastProcessor; 135import org.apache.camel.processor.Pipeline; 136import org.apache.camel.processor.PollEnricher; 137import org.apache.camel.processor.PollProcessor; 138import org.apache.camel.processor.RecipientList; 139import org.apache.camel.processor.RemoveHeaderProcessor; 140import org.apache.camel.processor.RemoveHeadersProcessor; 141import org.apache.camel.processor.RemovePropertiesProcessor; 142import org.apache.camel.processor.RemovePropertyProcessor; 143import org.apache.camel.processor.RemoveVariableProcessor; 144import org.apache.camel.processor.Resequencer; 145import org.apache.camel.processor.RollbackProcessor; 146import org.apache.camel.processor.RoutingSlip; 147import org.apache.camel.processor.SamplingThrottler; 148import org.apache.camel.processor.ScriptProcessor; 149import org.apache.camel.processor.SendDynamicProcessor; 150import org.apache.camel.processor.SendProcessor; 151import org.apache.camel.processor.SetBodyProcessor; 152import org.apache.camel.processor.SetHeaderProcessor; 153import org.apache.camel.processor.SetHeadersProcessor; 154import org.apache.camel.processor.SetPropertyProcessor; 155import org.apache.camel.processor.SetVariableProcessor; 156import org.apache.camel.processor.SetVariablesProcessor; 157import org.apache.camel.processor.Splitter; 158import org.apache.camel.processor.StepProcessor; 159import org.apache.camel.processor.StopProcessor; 160import org.apache.camel.processor.StreamResequencer; 161import org.apache.camel.processor.ThreadsProcessor; 162import org.apache.camel.processor.Throttler; 163import org.apache.camel.processor.ThrowExceptionProcessor; 164import org.apache.camel.processor.TransformProcessor; 165import org.apache.camel.processor.TryProcessor; 166import org.apache.camel.processor.WireTapProcessor; 167import org.apache.camel.processor.aggregate.AggregateProcessor; 168import org.apache.camel.processor.idempotent.IdempotentConsumer; 169import org.apache.camel.processor.loadbalancer.FailOverLoadBalancer; 170import org.apache.camel.processor.loadbalancer.LoadBalancer; 171import org.apache.camel.processor.loadbalancer.RandomLoadBalancer; 172import org.apache.camel.processor.loadbalancer.RoundRobinLoadBalancer; 173import org.apache.camel.processor.loadbalancer.StickyLoadBalancer; 174import org.apache.camel.processor.loadbalancer.TopicLoadBalancer; 175import org.apache.camel.processor.loadbalancer.WeightedLoadBalancer; 176import org.apache.camel.processor.transformer.DataTypeProcessor; 177import org.apache.camel.spi.BrowsableEndpoint; 178import org.apache.camel.spi.DataFormat; 179import org.apache.camel.spi.ErrorHandler; 180import org.apache.camel.spi.EventNotifier; 181import org.apache.camel.spi.ManagementObjectStrategy; 182import org.apache.camel.spi.RouteController; 183import org.apache.camel.spi.SupervisingRouteController; 184import org.apache.camel.support.ScheduledPollConsumer; 185import org.apache.camel.support.processor.ConvertBodyProcessor; 186import org.apache.camel.support.processor.ConvertHeaderProcessor; 187import org.apache.camel.support.processor.ConvertVariableProcessor; 188import org.apache.camel.support.processor.MarshalProcessor; 189import org.apache.camel.support.processor.PredicateValidatingProcessor; 190import org.apache.camel.support.processor.ThroughputLogger; 191import org.apache.camel.support.processor.UnmarshalProcessor; 192 193/** 194 * Default {@link org.apache.camel.spi.ManagementObjectStrategy}. 195 */ 196public class DefaultManagementObjectStrategy implements ManagementObjectStrategy { 197 198 @Override 199 public Object getManagedObjectForCamelContext(CamelContext context) { 200 ManagedCamelContext mc = new ManagedCamelContext(context); 201 mc.init(context.getManagementStrategy()); 202 return mc; 203 } 204 205 @Override 206 public Object getManagedObjectForCamelHealth(CamelContext context, HealthCheckRegistry healthCheckRegistry) { 207 ManagedCamelHealth mch = new ManagedCamelHealth(context, healthCheckRegistry); 208 mch.init(context.getManagementStrategy()); 209 return mch; 210 } 211 212 @Override 213 public Object getManagedObjectForComponent(CamelContext context, Component component, String name) { 214 ManagedComponent mc = new ManagedComponent(name, component); 215 mc.init(context.getManagementStrategy()); 216 return mc; 217 } 218 219 @Override 220 public Object getManagedObjectForDataFormat(CamelContext context, DataFormat dataFormat) { 221 ManagedDataFormat md = new ManagedDataFormat(context, dataFormat); 222 md.init(context.getManagementStrategy()); 223 return md; 224 } 225 226 @Override 227 public Object getManagedObjectForEndpoint(CamelContext context, Endpoint endpoint) { 228 // we only want to manage singleton endpoints 229 if (!endpoint.isSingleton()) { 230 return null; 231 } 232 233 if (endpoint instanceof BrowsableEndpoint browsableEndpoint) { 234 ManagedBrowsableEndpoint me = new ManagedBrowsableEndpoint(browsableEndpoint); 235 me.init(context.getManagementStrategy()); 236 return me; 237 } else { 238 ManagedEndpoint me = new ManagedEndpoint(endpoint); 239 me.init(context.getManagementStrategy()); 240 return me; 241 } 242 } 243 244 @Override 245 public Object getManagedObjectForRouteController(CamelContext context, RouteController routeController) { 246 ManagedService mrc; 247 if (routeController instanceof SupervisingRouteController supervisingRouteController) { 248 mrc = new ManagedSupervisingRouteController(context, supervisingRouteController); 249 } else { 250 mrc = new ManagedRouteController(context, routeController); 251 } 252 mrc.init(context.getManagementStrategy()); 253 return mrc; 254 } 255 256 @Override 257 public Object getManagedObjectForRoute(CamelContext context, Route route) { 258 ManagedRoute mr; 259 if (route.supportsSuspension()) { 260 mr = new ManagedSuspendableRoute(context, route); 261 } else { 262 mr = new ManagedRoute(context, route); 263 } 264 mr.init(context.getManagementStrategy()); 265 return mr; 266 } 267 268 @Override 269 public Object getManagedObjectForRouteGroup(CamelContext context, String group) { 270 if (group == null) { 271 return null; 272 } 273 ManagedRouteGroup mr = new ManagedRouteGroup(context, group); 274 mr.init(context.getManagementStrategy()); 275 return mr; 276 } 277 278 @Override 279 public Object getManagedObjectForThreadPool( 280 CamelContext context, ThreadPoolExecutor threadPool, 281 String id, String sourceId, String routeId, String threadPoolProfileId) { 282 ManagedThreadPool mtp = new ManagedThreadPool(context, threadPool, id, sourceId, routeId, threadPoolProfileId); 283 mtp.init(context.getManagementStrategy()); 284 return mtp; 285 } 286 287 @Override 288 public Object getManagedObjectForEventNotifier(CamelContext context, EventNotifier eventNotifier) { 289 ManagedEventNotifier men = new ManagedEventNotifier(context, eventNotifier); 290 men.init(context.getManagementStrategy()); 291 return men; 292 } 293 294 @Override 295 public Object getManagedObjectForConsumer(CamelContext context, Consumer consumer) { 296 ManagedConsumer mc; 297 if (consumer instanceof ScheduledPollConsumer scheduledPollConsumer) { 298 mc = new ManagedScheduledPollConsumer(context, scheduledPollConsumer); 299 } else { 300 mc = new ManagedConsumer(context, consumer); 301 } 302 mc.init(context.getManagementStrategy()); 303 return mc; 304 } 305 306 @Override 307 public Object getManagedObjectForProducer(CamelContext context, Producer producer) { 308 ManagedProducer mp = new ManagedProducer(context, producer); 309 mp.init(context.getManagementStrategy()); 310 return mp; 311 } 312 313 @Override 314 public Object getManagedObjectForService(CamelContext context, Service service) { 315 ManagedService mc = new ManagedService(context, service); 316 mc.init(context.getManagementStrategy()); 317 return mc; 318 } 319 320 @Override 321 public Object getManagedObjectForClusterService(CamelContext context, CamelClusterService service) { 322 ManagedClusterService mcs = new ManagedClusterService(context, service); 323 mcs.init(context.getManagementStrategy()); 324 return mcs; 325 } 326 327 @Override 328 public Object getManagedObjectForProcessor( 329 CamelContext context, Processor processor, 330 NamedNode node, Route route) { 331 ManagedProcessor answer = null; 332 333 ProcessorDefinition<?> definition = (ProcessorDefinition<?>) node; 334 335 if (definition instanceof RecipientListDefinition && processor instanceof Pipeline pipeline) { 336 // special for RecipientListDefinition, as the processor is wrapped in a pipeline as last 337 for (Processor value : pipeline.next()) { 338 processor = value; 339 } 340 } 341 342 // unwrap delegates as we want the real target processor 343 Processor target = processor; 344 while (target != null) { 345 346 // skip error handlers 347 if (target instanceof ErrorHandler) { 348 return false; 349 } 350 351 if (target instanceof ConvertBodyProcessor cbp) { 352 answer = new ManagedConvertBody(context, cbp, definition); 353 } else if (target instanceof ConvertHeaderProcessor chp) { 354 answer = new ManagedConvertHeader(context, chp, definition); 355 } else if (target instanceof ConvertVariableProcessor cvp) { 356 answer = new ManagedConvertVariable(context, cvp, definition); 357 } else if (target instanceof ChoiceProcessor cp) { 358 answer = new ManagedChoice(context, cp, definition); 359 } else if (target instanceof ClaimCheckProcessor ccp) { 360 answer = new ManagedClaimCheck(context, ccp, definition); 361 } else if (target instanceof Delayer d) { 362 answer = new ManagedDelayer(context, d, definition); 363 } else if (target instanceof DisabledProcessor dp) { 364 answer = new ManagedDisabled(context, dp, definition); 365 } else if (target instanceof TryProcessor tryProc) { 366 answer = new ManagedDoTry(context, tryProc, cast(definition)); 367 } else if (target instanceof CatchProcessor catchProc) { 368 answer = new ManagedDoCatch(context, catchProc, cast(definition)); 369 } else if (target instanceof FinallyProcessor fp) { 370 answer = new ManagedDoFinally(context, fp, cast(definition)); 371 } else if (target instanceof Throttler t) { 372 answer = new ManagedThrottler(context, t, definition); 373 } else if (target instanceof DynamicRouter dr) { 374 answer = new ManagedDynamicRouter(context, dr, cast(definition)); 375 } else if (target instanceof RoutingSlip rs) { 376 answer = new ManagedRoutingSlip(context, rs, cast(definition)); 377 } else if (target instanceof FilterProcessor fp) { 378 answer = new ManagedFilter(context, fp, (ExpressionNode) definition); 379 } else if (target instanceof LogProcessor lp) { 380 answer = new ManagedLog(context, lp, definition); 381 } else if (target instanceof LoopProcessor loopProc) { 382 answer = new ManagedLoop(context, loopProc, cast(definition)); 383 } else if (target instanceof MarshalProcessor mp) { 384 answer = new ManagedMarshal(context, mp, cast(definition)); 385 } else if (target instanceof UnmarshalProcessor up) { 386 answer = new ManagedUnmarshal(context, up, cast(definition)); 387 } else if (target instanceof FailOverLoadBalancer folb) { 388 answer = new ManagedFailoverLoadBalancer(context, folb, cast(definition)); 389 } else if (target instanceof RandomLoadBalancer rlb) { 390 answer = new ManagedRandomLoadBalancer(context, rlb, cast(definition)); 391 } else if (target instanceof RoundRobinLoadBalancer rrlb) { 392 answer = new ManagedRoundRobinLoadBalancer(context, rrlb, cast(definition)); 393 } else if (target instanceof StickyLoadBalancer slb) { 394 answer = new ManagedStickyLoadBalancer(context, slb, cast(definition)); 395 } else if (target instanceof TopicLoadBalancer tlb) { 396 answer = new ManagedTopicLoadBalancer(context, tlb, cast(definition)); 397 } else if (target instanceof WeightedLoadBalancer wlb) { 398 answer = new ManagedWeightedLoadBalancer(context, wlb, cast(definition)); 399 } else if (target instanceof RecipientList rl) { 400 answer = new ManagedRecipientList(context, rl, cast(definition)); 401 } else if (target instanceof Splitter s) { 402 answer = new ManagedSplitter(context, s, cast(definition)); 403 } else if (target instanceof MulticastProcessor mcp) { 404 answer = new ManagedMulticast(context, mcp, definition); 405 } else if (target instanceof SamplingThrottler st) { 406 answer = new ManagedSamplingThrottler(context, st, definition); 407 } else if (target instanceof Resequencer r) { 408 answer = new ManagedResequencer(context, r, definition); 409 } else if (target instanceof RollbackProcessor rp) { 410 answer = new ManagedRollback(context, rp, definition); 411 } else if (target instanceof StreamResequencer sr) { 412 answer = new ManagedResequencer(context, sr, definition); 413 } else if (target instanceof SetBodyProcessor sbp) { 414 answer = new ManagedSetBody(context, sbp, cast(definition)); 415 } else if (target instanceof RemoveHeaderProcessor rhp) { 416 answer = new ManagedRemoveHeader(context, rhp, definition); 417 } else if (target instanceof RemoveHeadersProcessor rhsp) { 418 answer = new ManagedRemoveHeaders(context, rhsp, definition); 419 } else if (target instanceof SetHeaderProcessor shp) { 420 answer = new ManagedSetHeader(context, shp, cast(definition)); 421 } else if (target instanceof SetHeadersProcessor shsp) { 422 answer = new ManagedSetHeaders(context, shsp, cast(definition)); 423 } else if (target instanceof SetVariableProcessor svp) { 424 answer = new ManagedSetVariable(context, svp, cast(definition)); 425 } else if (target instanceof SetVariablesProcessor svsp) { 426 answer = new ManagedSetVariables(context, svsp, cast(definition)); 427 } else if (target instanceof RemovePropertyProcessor rpp) { 428 answer = new ManagedRemoveProperty(context, rpp, definition); 429 } else if (target instanceof RemovePropertiesProcessor rpsp) { 430 answer = new ManagedRemoveProperties(context, rpsp, definition); 431 } else if (target instanceof RemoveVariableProcessor rvp) { 432 answer = new ManagedRemoveVariable(context, rvp, definition); 433 } else if (target instanceof SetPropertyProcessor spp) { 434 answer = new ManagedSetProperty(context, spp, cast(definition)); 435 } else if (target instanceof ExchangePatternProcessor epp) { 436 answer = new ManagedSetExchangePattern(context, epp, definition); 437 } else if (target instanceof ScriptProcessor scrp) { 438 answer = new ManagedScript(context, scrp, cast(definition)); 439 } else if (target instanceof StepProcessor stepp) { 440 answer = new ManagedStep(context, stepp, definition); 441 } else if (target instanceof StopProcessor stopp) { 442 answer = new ManagedStop(context, stopp, definition); 443 } else if (target instanceof ThreadsProcessor tp) { 444 answer = new ManagedThreads(context, tp, definition); 445 } else if (target instanceof ThrowExceptionProcessor tep) { 446 answer = new ManagedThrowException(context, tep, definition); 447 } else if (target instanceof TransformProcessor) { 448 answer = new ManagedTransformer(context, target, cast(definition)); 449 } else if (target instanceof DataTypeProcessor) { 450 answer = new ManagedDataTypeTransformer(context, target, cast(definition)); 451 } else if (target instanceof PredicateValidatingProcessor pvp) { 452 answer = new ManagedValidate(context, pvp, cast(definition)); 453 } else if (target instanceof WireTapProcessor wtp) { 454 answer = new ManagedWireTapProcessor(context, wtp, definition); 455 } else if (target instanceof SendDynamicProcessor sdp) { 456 answer = new ManagedSendDynamicProcessor(context, sdp, definition); 457 } else if (target instanceof SendProcessor sp) { 458 // special for sending to throughput logger 459 if (sp.getDestination() instanceof LogEndpoint le && le.getLogger() instanceof ThroughputLogger tl) { 460 answer = new ManagedThroughputLogger(context, tl, definition); 461 } 462 // regular send processor 463 if (answer == null) { 464 answer = new ManagedSendProcessor(context, sp, definition); 465 } 466 } else if (target instanceof BeanProcessor bp) { 467 answer = new ManagedBeanProcessor(context, bp, definition); 468 } else if (target instanceof IdempotentConsumer ic) { 469 answer = new ManagedIdempotentConsumer(context, ic, cast(definition)); 470 } else if (target instanceof AggregateProcessor ap) { 471 answer = new ManagedAggregateProcessor(context, ap, cast(definition)); 472 } else if (target instanceof Enricher e) { 473 answer = new ManagedEnricher(context, e, cast(definition)); 474 } else if (target instanceof PollProcessor pp) { 475 answer = new ManagedPoll(context, pp, cast(definition)); 476 } else if (target instanceof PollEnricher pe) { 477 answer = new ManagedPollEnricher(context, pe, cast(definition)); 478 } 479 480 // special for custom load balancer 481 if (definition instanceof LoadBalanceDefinition lb) { 482 if (lb.getLoadBalancerType() instanceof CustomLoadBalancerDefinition 483 && target instanceof LoadBalancer loadBalancer) { 484 answer = new ManagedCustomLoadBalancer(context, loadBalancer, lb); 485 } 486 } 487 488 if (answer != null) { 489 // break out as we found an answer 490 break; 491 } 492 493 // no answer yet, so unwrap any delegates and try again 494 if (target instanceof DelegateProcessor delegateProcessor) { 495 target = delegateProcessor.getProcessor(); 496 } else { 497 // no delegate so we dont have any target to try next 498 break; 499 } 500 } 501 502 if (answer == null && definition instanceof ProcessDefinition pd) { 503 answer = new ManagedProcess(context, target, pd); 504 } else if (answer == null) { 505 // fallback to a generic processor 506 answer = new ManagedProcessor(context, target, definition); 507 } 508 509 answer.setRoute(route); 510 answer.init(context.getManagementStrategy()); 511 return answer; 512 } 513 514 @SuppressWarnings("unchecked") 515 private <T extends ProcessorDefinition<?>> T cast(ProcessorDefinition<?> definition) { 516 return (T) definition; 517 } 518 519}