import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
/**
 * Shutdown all the consumers immediately.
 *
 * @param routes the routes to shutdown
 */
protected void shutdownRoutesNow(List<RouteStartupOrder> routes) {
    for (RouteStartupOrder order : routes) {

        // set the route to shutdown as fast as possible by stopping after
        // it has completed its current task
        ShutdownRunningTask current = order.getRoute().getRouteContext().getShutdownRunningTask();
        if (current != ShutdownRunningTask.CompleteCurrentTaskOnly) {
            LOG.debug("Changing shutdownRunningTask from {} to " +  ShutdownRunningTask.CompleteCurrentTaskOnly
                + " on route {} to shutdown faster", current, order.getRoute().getId());
            order.getRoute().getRouteContext().setShutdownRunningTask(ShutdownRunningTask.CompleteCurrentTaskOnly);
        }

        for (Consumer consumer : order.getInputs()) {
            shutdownNow(consumer);
        }
    }
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
public int getPendingExchangesSize() {
    int answer;
    // only return the real pending size in case we are configured to complete all tasks
    if (ShutdownRunningTask.CompleteAllTasks == shutdownRunningTask) {
        answer = pendingExchanges;
    } else {
        answer = 0;
    }

    if (answer == 0 && isPolling()) {
        // force at least one pending exchange if we are polling as there is a little gap
        // in the processBatch method and until an exchange gets enlisted as in-flight
        // which happens later, so we need to signal back to the shutdown strategy that
        // there is a pending exchange. When we are no longer polling, then we will return 0
        LOG.trace("Currently polling so returning 1 as pending exchanges");
        answer = 1;
    }

    return answer;
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
public boolean isBatchAllowed() {
    // stop if we are not running
    boolean answer = isRunAllowed();
    if (!answer) {
        return false;
    }

    if (shutdownRunningTask == null) {
        // we are not shutting down so continue to run
        return true;
    }

    // we are shutting down so only continue if we are configured to complete all tasks
    return ShutdownRunningTask.CompleteAllTasks == shutdownRunningTask;
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            CronScheduledRoutePolicy policy = new CronScheduledRoutePolicy();
            policy.setRouteStartTime("* 0/2 * * * ?");
            policy.setRouteStopTime("* 1/2 * * * ?");
            policy.setRouteStopGracePeriod(250);
            policy.setTimeUnit(TimeUnit.SECONDS);

            from(ftp)
                .noAutoStartup().routePolicy(policy).shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
                .log("Processing ${file:name}")
                .to("log:done");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
    // store a reference what to do in case when shutting down and we have pending messages
    this.shutdownRunningTask = shutdownRunningTask;
    // do not defer shutdown
    return false;
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
public ShutdownRunningTask getShutdownRunningTask() {
    if (shutdownRunningTask != null) {
        return shutdownRunningTask;
    } else {
        // fallback to the option from camel context
        return getCamelContext().getShutdownRunningTask();
    }
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        // START SNIPPET: e1
        public void configure() throws Exception {
            from(url).routeId("foo").noAutoStartup()
                // let it complete all tasks during shutdown
                .shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
                .process(new MyProcessor())
                .to("mock:bar");
        }
        // END SNIPPET: e1
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from(url)
                // let it complete only current task so we shutdown faster
                .shutdownRunningTask(ShutdownRunningTask.CompleteCurrentTaskOnly)
                .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2").to("mock:bar");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
public void testShutdownCompleteAllTasks() throws Exception {
    final String url = "file:target/seda";
    template.sendBodyAndHeader(url, "A", Exchange.FILE_NAME, "a.txt");
    template.sendBodyAndHeader(url, "B", Exchange.FILE_NAME, "b.txt");
    template.sendBodyAndHeader(url, "C", Exchange.FILE_NAME, "c.txt");
    template.sendBodyAndHeader(url, "D", Exchange.FILE_NAME, "d.txt");
    template.sendBodyAndHeader(url, "E", Exchange.FILE_NAME, "e.txt");
    
    // give it 20 seconds to shutdown
    context.getShutdownStrategy().setTimeout(20);

    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from(url).routeId("route1")
                // let it complete all tasks during shutdown
                .shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
                .to("log:delay")
                .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2")
                .to("log:bar")
                .to("mock:bar");
        }
    });
    context.start();

    MockEndpoint bar = getMockEndpoint("mock:bar");
    bar.expectedMinimumMessageCount(1);

    assertMockEndpointsSatisfied();

    // shutdown during processing
    context.stop();

    // should route all 5
    assertEquals("Should complete all messages", 5, bar.getReceivedCounter());
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("ibatis:selectAllAccounts").noAutoStartup().routeId("route1")
                 // let it complete all tasks
                 .shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
                 .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2").to("mock:bar");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("ibatis:selectAllAccounts").routeId("route1")
                 // let it complete only current task so we shutdown faster
                 .shutdownRunningTask(ShutdownRunningTask.CompleteCurrentTaskOnly)
                 .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2").to("mock:bar");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from(getFtpUrl()).routeId("route1")
                // let it complete all tasks during shutdown
                .shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
                .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2").to("mock:bar");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from(getFtpUrl()).routeId("route1")
                // let it complete only current task so we shutdown faster
                .shutdownRunningTask(ShutdownRunningTask.CompleteCurrentTaskOnly)
                .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2").to("mock:bar");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("mybatis:selectAllAccounts").routeId("route1")
                 // let it complete only current task so we shutdown faster
                 .shutdownRunningTask(ShutdownRunningTask.CompleteCurrentTaskOnly)
                 .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2").to("mock:bar");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("mybatis:selectAllAccounts").noAutoStartup().routeId("route1")
                 // let it complete all tasks
                 .shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
                 .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2").to("mock:bar");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        public void configure() throws Exception {
            from("pop3://[email protected]?password=secret").routeId("route1")
                     // let it complete only current task so we shutdown faster
                     .shutdownRunningTask(ShutdownRunningTask.CompleteCurrentTaskOnly)
                     .delay(1000).to("seda:foo");

            from("seda:foo").routeId("route2").to("mock:bar");
        }
    };
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Test
public void testShutdownCompleteAllTasks() throws Exception {
    // give it 20 seconds to shutdown
    context.getShutdownStrategy().setTimeout(20);
    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("pop3://[email protected]?password=secret&initialDelay=2s").routeId("route1")
                    // let it complete all tasks during shutdown
                    .shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks)
                    .delay(500).to("mock:bar");
        }
    });
    context.start();

    MockEndpoint bar = getMockEndpoint("mock:bar");
    bar.expectedMinimumMessageCount(1);

    assertMockEndpointsSatisfied();

    int batch = bar.getReceivedExchanges().get(0).getProperty(Exchange.BATCH_SIZE, int.class);

    // shutdown during processing
    context.stop();

    // should route all
    assertEquals("Should complete all messages", batch, bar.getReceivedCounter());
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
@Test
public void testShutdownCompleteAllTasks() throws Exception {
    final String url = "file:target/disruptor";
    template.sendBodyAndHeader(url, "A", Exchange.FILE_NAME, "a.txt");
    template.sendBodyAndHeader(url, "B", Exchange.FILE_NAME, "b.txt");
    template.sendBodyAndHeader(url, "C", Exchange.FILE_NAME, "c.txt");
    template.sendBodyAndHeader(url, "D", Exchange.FILE_NAME, "d.txt");
    template.sendBodyAndHeader(url, "E", Exchange.FILE_NAME, "e.txt");

    // give it 20 seconds to shutdown
    context.getShutdownStrategy().setTimeout(20 * 100000);

    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from(url).routeId("route1")
                    // let it complete all tasks during shutdown
                    .shutdownRunningTask(ShutdownRunningTask.CompleteAllTasks).to("log:delay").delay(1000)
                    .to("disruptor:foo?size=8");

            from("disruptor:foo?size=8").routeId("route2").to("log:bar").to("mock:bar");
        }
    });
    context.start();

    final MockEndpoint bar = getMockEndpoint("mock:bar");
    bar.expectedMinimumMessageCount(1);

    assertMockEndpointsSatisfied();

    // shutdown during processing
    context.stop();

    // should route all 5
    assertEquals("Should complete all messages", 5, bar.getReceivedCounter());
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
/**
 * To control how to shutdown the route.
 */
public ShutdownRunningTask getShutdownRunningTask() {
    return shutdownRunningTask;
}
 

import org.apache.camel.ShutdownRunningTask; //导入依赖的package包/类
/**
 * To control how to shutdown the route.
 */
@XmlAttribute @Metadata(defaultValue = "CompleteCurrentTaskOnly")
public void setShutdownRunningTask(ShutdownRunningTask shutdownRunningTask) {
    this.shutdownRunningTask = shutdownRunningTask;
}