Java 类org.apache.camel.spi.RoutePolicy 实例源码

public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
    LOG.debug("Running ScheduledJob: jobExecutionContext={}", jobExecutionContext);

    SchedulerContext schedulerContext = getSchedulerContext(jobExecutionContext);
    ScheduledJobState state = (ScheduledJobState) schedulerContext.get(jobExecutionContext.getJobDetail().getKey().toString());
    Action storedAction = state.getAction(); 
    Route storedRoute = state.getRoute();

    List<RoutePolicy> policyList = storedRoute.getRouteContext().getRoutePolicyList();
    for (RoutePolicy policy : policyList) {
        try {
            if (policy instanceof ScheduledRoutePolicy) {
                ((ScheduledRoutePolicy)policy).onJobExecute(storedAction, storedRoute);
            }
        } catch (Exception e) {
            throw new JobExecutionException("Failed to execute Scheduled Job for route " + storedRoute.getId()
                    + " with trigger name: " + jobExecutionContext.getTrigger().getKey(), e);
        }
    }
}

@Override
public void after(Exchange exchange, Object data) throws Exception {
    // do not invoke it if Camel is stopping as we don't want
    // the policy to start a consumer during Camel is stopping
    if (isCamelStopping(exchange.getContext())) {
        return;
    }

    for (RoutePolicy policy : routePolicies) {
        try {
            if (isRoutePolicyRunAllowed(policy)) {
                policy.onExchangeDone(route, exchange);
            }
        } catch (Exception e) {
            LOG.warn("Error occurred during onExchangeDone on RoutePolicy: " + policy
                    + ". This exception will be ignored", e);
        }
    }
}

public String getRoutePolicyList() {
    List<RoutePolicy> policyList = route.getRouteContext().getRoutePolicyList();

    if (policyList == null || policyList.isEmpty()) {
        // return an empty string to have it displayed nicely in JMX consoles
        return "";
    }

    StringBuilder sb = new StringBuilder();
    for (int i = 0; i < policyList.size(); i++) {
        RoutePolicy policy = policyList.get(i);
        sb.append(policy.getClass().getSimpleName());
        sb.append("(").append(ObjectHelper.getIdentityHashCode(policy)).append(")");
        if (i < policyList.size() - 1) {
            sb.append(", ");
        }
    }
    return sb.toString();
}

protected void doStart() throws Exception {
    warmUp();

    for (Route route : routes) {
        // start the route itself
        ServiceHelper.startService(route);

        // invoke callbacks on route policy
        if (route.getRouteContext().getRoutePolicyList() != null) {
            for (RoutePolicy routePolicy : route.getRouteContext().getRoutePolicyList()) {
                routePolicy.onStart(route);
            }
        }

        // fire event
        EventHelper.notifyRouteStarted(camelContext, route);
    }
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // create the flip route policy
            RoutePolicy policy = new FlipRoutePolicy("foo", "bar");

            // use the flip route policy in the foo route
            from("timer://foo")
                .routeId("foo").routePolicy(policy)
                .setBody().constant("Foo message")
                .to("log:foo")
                .to("mock:foo");

            // use the flip route policy in the bar route and do NOT start
            // this route on startup
            from("timer://bar")
                .routeId("bar").routePolicy(policy).noAutoStartup()
                .setBody().constant("Bar message")
                .to("log:bar")
                .to("mock:bar");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // create the flip route policy
            RoutePolicy policy = new FlipRoutePolicy("foo", "bar");

            // use the flip route policy in the foo route
            from("timer:foo?delay=500")
                .routeId("foo").routePolicy(policy)
                .setBody().constant("Foo message")
                .to("log:foo")
                .to("mock:foo");

            // use the flip route policy in the bar route and do NOT start
            // this route on startup
            from("timer:bar?delay=500")
                .routeId("bar").routePolicy(policy).noAutoStartup()
                .setBody().constant("Bar message")
                .to("log:bar")
                .to("mock:bar");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // create the flip route policy
            RoutePolicy policy = new FlipRoutePolicy("foo", "bar");

            // use the flip route policy in the foo route
            from("timer://foo?delay=500")
                .routeId("foo").routePolicy(policy)
                .setBody().constant("Foo message")
                .to("log:foo")
                .to("mock:foo");

            // use the flip route policy in the bar route and do NOT start
            // this route on startup
            from("timer://bar?delay=500")
                .routeId("bar").routePolicy(policy).noAutoStartup()
                .setBody().constant("Bar message")
                .to("log:bar")
                .to("mock:bar");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // create the flip route policy
            RoutePolicy policy = new FlipRoutePolicy("foo", "bar");

            // use the flip route policy in the foo route
            from("timer://foo")
                .routeId("foo").routePolicy(policy)
                .setBody().constant("Foo message")
                .to("log:foo")
                .to("mock:foo");

            // use the flip route policy in the bar route and do NOT start
            // this route on startup
            from("timer://bar")
                .routeId("bar").routePolicy(policy).noAutoStartup()
                .setBody().constant("Bar message")
                .to("log:bar")
                .to("mock:bar");
        }
    };
}

/**
 * Configures route policies for this route
 *
 * @param policies the route policies
 * @return the builder
 */
public RouteDefinition routePolicy(RoutePolicy... policies) {
    if (routePolicies == null) {
        routePolicies = new ArrayList<RoutePolicy>();
    }
    for (RoutePolicy policy : policies) {
        routePolicies.add(policy);
    }
    return this;
}

/**
 * Strategy to determine if this policy is allowed to run
 *
 * @param policy the policy
 * @return <tt>true</tt> to run
 */
protected boolean isRoutePolicyRunAllowed(RoutePolicy policy) {
    if (policy instanceof StatefulService) {
        StatefulService ss = (StatefulService) policy;
        return ss.isRunAllowed();
    }
    return true;
}

@Override
public Object before(Exchange exchange) throws Exception {
    // invoke begin
    for (RoutePolicy policy : routePolicies) {
        try {
            if (isRoutePolicyRunAllowed(policy)) {
                policy.onExchangeBegin(route, exchange);
            }
        } catch (Exception e) {
            LOG.warn("Error occurred during onExchangeBegin on RoutePolicy: " + policy
                    + ". This exception will be ignored", e);
        }
    }
    return null;
}

@Override
protected void doSuspend() throws Exception {
    // suspend and resume logic is provided by DefaultCamelContext which leverages ShutdownStrategy
    // to safely suspend and resume
    for (Route route : routes) {
        if (route.getRouteContext().getRoutePolicyList() != null) {
            for (RoutePolicy routePolicy : route.getRouteContext().getRoutePolicyList()) {
                routePolicy.onSuspend(route);
            }
        }
    }
}

@Override
protected void doResume() throws Exception {
    // suspend and resume logic is provided by DefaultCamelContext which leverages ShutdownStrategy
    // to safely suspend and resume
    for (Route route : routes) {
        if (route.getRouteContext().getRoutePolicyList() != null) {
            for (RoutePolicy routePolicy : route.getRouteContext().getRoutePolicyList()) {
                routePolicy.onResume(route);
            }
        }
    }
}

private RoutePolicy createRouteStartPolicy() {
    SimpleScheduledRoutePolicy policy = new SimpleScheduledRoutePolicy();
    long startTime = System.currentTimeMillis() + 3000L;
    policy.setRouteStartDate(new Date(startTime));
    policy.setRouteStartRepeatCount(1);
    policy.setRouteStartRepeatInterval(3000);

    return policy;
}

@SuppressWarnings("unchecked")
private CamelContext startRouteWithPolicy(String policyBeanName) throws Exception {
    CamelContext context = new DefaultCamelContext();
    List<RouteDefinition> routes = (List<RouteDefinition>)applicationContext.getBean("testRouteContext");
    RoutePolicy policy = applicationContext.getBean(policyBeanName, RoutePolicy.class);
    assertTrue(getTestType() == TestType.SIMPLE 
        ? policy instanceof SimpleScheduledRoutePolicy 
        : policy instanceof CronScheduledRoutePolicy);
    routes.get(0).routePolicy(policy);
    ((ModelCamelContext)context).addRouteDefinitions(routes);
    context.start();
    return context;
}

@Override
public RoutePolicy createRoutePolicy(CamelContext camelContext, String routeId, RouteDefinition routeDefinition) {
    MetricsRoutePolicy answer = new MetricsRoutePolicy();
    answer.setMetricsRegistry(getMetricsRegistry());
    answer.setUseJmx(isUseJmx());
    answer.setJmxDomain(getJmxDomain());
    answer.setPrettyPrint(isPrettyPrint());
    answer.setRateUnit(getRateUnit());
    answer.setDurationUnit(getDurationUnit());
    return answer;
}

private RoutePolicy createRouteStartPolicy() {
    SimpleScheduledRoutePolicy policy = new SimpleScheduledRoutePolicy();
    long startTime = System.currentTimeMillis() + 3000L;
    policy.setRouteStartDate(new Date(startTime));
    policy.setRouteStartRepeatCount(1);
    policy.setRouteStartRepeatInterval(3000);

    return policy;
}

@SuppressWarnings("unchecked")
private CamelContext startRouteWithPolicy(String policyBeanName) throws Exception {
    CamelContext context = new DefaultCamelContext();
    List<RouteDefinition> routes = (List<RouteDefinition>)applicationContext.getBean("testRouteContext");
    RoutePolicy policy = applicationContext.getBean(policyBeanName, RoutePolicy.class);
    assertTrue(getTestType() == TestType.SIMPLE 
        ? policy instanceof SimpleScheduledRoutePolicy 
        : policy instanceof CronScheduledRoutePolicy);
    routes.get(0).routePolicy(policy);
    ((ModelCamelContext)context).addRouteDefinitions(routes);
    context.start();
    return context;
}

@Override
public RoutePolicy createRoutePolicy(CamelContext camelContext, String s, RouteDefinition routeDefinition) {
    return new SingleMessageRoutePolicy();
}

@Override
public RoutePolicy createRoutePolicy(CamelContext camelContext, String s, RouteDefinition routeDefinition) {
    return new SingleMessageRoutePolicy();
}

@Override
public RoutePolicy createRoutePolicy(CamelContext camelContext, String routeId, RouteDefinition route) {
    return new SingleMessageRoutePolicy();
}

public List<RoutePolicy> getRoutePolicies() {
    return routePolicies;
}

@XmlTransient
public void setRoutePolicies(List<RoutePolicy> routePolicies) {
    this.routePolicies = routePolicies;
}

public RoutePolicyAdvice(List<RoutePolicy> routePolicies) {
    this.routePolicies = routePolicies;
}

protected void doStop() throws Exception {

        // if we are stopping CamelContext then we are shutting down
        boolean isShutdownCamelContext = camelContext.isStopping();

        if (isShutdownCamelContext || isRemovingRoutes()) {
            // need to call onRoutesRemove when the CamelContext is shutting down or Route is shutdown
            for (LifecycleStrategy strategy : camelContext.getLifecycleStrategies()) {
                strategy.onRoutesRemove(routes);
            }
        }

        for (Route route : routes) {
            LOG.debug("Stopping services on route: {}", route.getId());

            // gather list of services to stop as we need to start child services as well
            Set<Service> services = gatherChildServices(route, true);

            // stop services
            stopChildService(route, services, isShutdownCamelContext);

            // stop the route itself
            if (isShutdownCamelContext) {
                ServiceHelper.stopAndShutdownServices(route);
            } else {
                ServiceHelper.stopServices(route);
            }

            // invoke callbacks on route policy
            if (route.getRouteContext().getRoutePolicyList() != null) {
                for (RoutePolicy routePolicy : route.getRouteContext().getRoutePolicyList()) {
                    routePolicy.onStop(route);
                }
            }
            // fire event
            EventHelper.notifyRouteStopped(camelContext, route);
        }
        if (isRemovingRoutes()) {
            camelContext.removeRouteCollection(routes);
        }
        // need to warm up again
        warmUpDone.set(false);
    }

public void setRoutePolicyList(List<RoutePolicy> routePolicyList) {
    this.routePolicyList = routePolicyList;
}

public List<RoutePolicy> getRoutePolicyList() {
    return routePolicyList;
}

@Override
public RoutePolicy createRoutePolicy(CamelContext camelContext, String routeId, RouteDefinition route) {
    return new MyRoutePolicy(routeId);
}

private RoutePolicy createThrottlePolicy() {
    ThrottlingInflightRoutePolicy policy = new ThrottlingInflightRoutePolicy();
    policy.setMaxInflightExchanges(10);
    return policy;
}

@Override
public RoutePolicy createRoutePolicy(CamelContext camelContext, String routeId, RouteDefinition route) {
    return new ZipkinRoutePolicy(routeId);
}

private RoutePolicy createThrottlePolicy() {
    ThrottlingInflightRoutePolicy policy = new ThrottlingInflightRoutePolicy();
    policy.setMaxInflightExchanges(10);
    return policy;
}