Java 类org.apache.camel.util.toolbox.AggregationStrategies 实例源码

protected RouteBuilder createRouteBuilder() {
    return new RouteBuilder() {
        public void configure() {
            from("direct:a")
                .setHeader("bar", constant("no"))
                    .to("mock:a")
                .multicast(AggregationStrategies.useOriginal()).to("direct:foo").end()
                .to("mock:result");

            from("direct:foo")
                .setHeader("foo", constant("yes"))
                .removeHeader("bar")
                .transform().simple("Foo was here ${body}")
                .to("mock:foo");
        }
    };
}

@Override
protected CamelContext createCamelContext() throws Exception {
    SimpleRegistry registry = new SimpleRegistry();
    registry.put("aggStrategy", AggregationStrategies.groupedExchange());

    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
    connectionFactory.setBrokerURL(broker.getTcpConnectorUri());

    SjmsComponent sjmsComponent = new SjmsComponent();
    sjmsComponent.setConnectionFactory(connectionFactory);

    SjmsBatchComponent sjmsBatchComponent = new SjmsBatchComponent();
    sjmsBatchComponent.setConnectionFactory(connectionFactory);

    CamelContext context = new DefaultCamelContext(registry);
    context.addComponent("sjms-batch", sjmsBatchComponent);
    context.addComponent("sjms", sjmsComponent);

    return context;
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                // do a little logging
                .log("Sending ${body} with correlation key ${header.myId}")
                // aggregate based on header correlation key
                // use class MyAggregationStrategy for aggregation
                // and complete when we have aggregated 3 messages
                .aggregate(header("myId"), AggregationStrategies.bean(new MyAggregationStrategyPojo()))
                    .completionSize(3)
                    // do a little logging for the published message
                    .log("Sending out ${body}")
                    // and send it to the mock
                    .to("mock:result");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .aggregate(constant(true), AggregationStrategies.bean(MyBodyAppender.class, "append"))
                    .completionSize(3)
                    .to("mock:result");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .pollEnrich("seda:foo", 100, AggregationStrategies.bean(appender))
                    .to("mock:result");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .setHeader("foo", constant("yes"))
                .aggregate(constant(true), AggregationStrategies.bean(appender, "appendWithHeadersAndProperties"))
                    .completionSize(3)
                    .to("mock:result");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .aggregate(constant(true), AggregationStrategies.bean(MyBodyAppender.class, "append"))
                    .completionSize(3)
                    .to("mock:result");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .aggregate(constant(true), AggregationStrategies.bean(appender))
                    .completionSize(3)
                    .to("mock:result");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .aggregate(constant(true), AggregationStrategies.beanAllowNull(appender, "addUsers"))
                    .completionSize(3)
                    .to("mock:result");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .aggregate(constant(true), AggregationStrategies.bean(appender, "append"))
                    .completionSize(3)
                    .to("mock:result");
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .aggregate(constant(true), AggregationStrategies.bean(appender, "appendWithHeaders"))
                    .completionSize(3)
                    .to("mock:result");
        }
    };
}

public void configure() throws Exception {

        fromF("dataset:createCommandDataset?initialDelay=%d", initialDelay)
                .routeId("dataset:createCommandsDataset")
                .startupOrder(1).autoStartup(true)
                .process(new MarshalProcessor()).to(targetEndpoint)
                .aggregate(constant(0), AggregationStrategies.useLatest()).completionSize(datasetSize)
                    .log("finished")
                    .log("will start next dataset")
                    .to("controlbus:route?routeId=dataset:increaseCommandsDataset&action=start")
                .end()
                ;

        fromF("dataset:increaseCommandDataset?initialDelay=%d", delayBetweenDatasets)
                .routeId("dataset:increaseCommandsDataset")
                .startupOrder(2).autoStartup(false)
                .process(new MarshalProcessor()).to(targetEndpoint)
                .aggregate(constant(0), AggregationStrategies.useLatest()).completionSize(datasetSize)
                    .log("finished")
                    .log("will start next dataset")
                    .to("controlbus:route?routeId=dataset:decreaseCommandsDataset&action=start")
                .end();
                ;

        fromF("dataset:decreaseCommandDataset?initialDelay=%d", delayBetweenDatasets)
                .routeId("dataset:decreaseCommandsDataset")
                .startupOrder(3).autoStartup(false)
                .process(new MarshalProcessor()).to(targetEndpoint)
                .aggregate(constant(0), AggregationStrategies.useLatest()).completionSize(datasetSize)
                    .log("finished")
                .end();
        ;

    }