Java 类org.apache.camel.processor.aggregate.UseLatestAggregationStrategy 实例源码

项目：Camel 文件：SplitterParallelIssueTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                    .log("Start ${body}")
                    .split(body().tokenize("@"), new UseLatestAggregationStrategy()).parallelProcessing().streaming()
                    .process(new Processor() {
                        @Override
                        public void process(Exchange exchange) throws Exception {
                            int num = exchange.getIn().getBody(int.class);
                            final long sleep = num * delay;
                            log.info("Sleep for " + sleep + "ms");
                            Thread.sleep(sleep);
                        }
                    })
                    .end()
                    .log("End ${body}")
                    .to("mock:end");
        }
    };
}

项目：Camel 文件：AggregateTimeoutWithExecutorServiceTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // share 8 threads among the 20 routes
            ScheduledExecutorService threadPool = context.getExecutorServiceManager().newScheduledThreadPool(this, "MyThreadPool", 8);
            for (int i = 0; i < NUM_AGGREGATORS; ++i) {
                from("direct:start" + i)
                    // aggregate timeout after 3th seconds
                    .aggregate(header("id"), new UseLatestAggregationStrategy()).completionTimeout(3000).timeoutCheckerExecutorService(threadPool)
                    .to("mock:result" + i);
            }
        }
    };
}

项目：Camel 文件：AggregatorTest.java

protected RouteBuilder createRouteBuilder() {
    return new RouteBuilder() {
        public void configure() {
            // START SNIPPET: ex
            // in this route we aggregate all from direct:state based on the header id cheese
            from("direct:start")
                .aggregate(header("cheese"), new UseLatestAggregationStrategy()).completionTimeout(1000L)
                    .to("mock:result");

            from("seda:header").setHeader("visited", constant(true))
                .aggregate(header("cheese"), new UseLatestAggregationStrategy()).completionTimeout(1000L)
                    .to("mock:result");

            // in this sample we aggregate with a completion predicate
            from("direct:predicate")
                .aggregate(header("cheese"), new UseLatestAggregationStrategy()).completionTimeout(1000L)
                    .completionPredicate(header("cheese").isEqualTo(123))
                    .to("mock:result");
            // END SNIPPET: ex
        }
    };
}

项目：Camel 文件：AggregatorExceptionHandleTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            onException(IllegalArgumentException.class).handled(true).to("mock:handled");

            from("direct:start")
                .aggregate(header("id"), new UseLatestAggregationStrategy()).completionTimeout(1000L)
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        String body = exchange.getIn().getBody(String.class);
                        if ("Damn".equals(body)) {
                            throw new IllegalArgumentException("Damn");
                        }
                        exchange.getOut().setBody("Bye World");
                    }
                })
                .to("mock:result");

        }
    };
}

项目：Camel 文件：DefaultAggregatorCollectionTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        public void configure() throws Exception {
            // START SNIPPET: e1
            // our route is aggregating from the direct queue and sending the response to the mock
            from("direct:start")
                // aggregated by header id
                // as we have not configured more on the aggregator it will default to aggregate the
                // latest exchange only
                .aggregate(header("id")).aggregationStrategy(new UseLatestAggregationStrategy())
                // wait for 0.5 seconds to aggregate
                .completionTimeout(500L)
                .to("mock:result");
            // END SNIPPET: e1
        }
    };
}

项目：Camel 文件：DistributedCompletionIntervalTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // START SNIPPET: e1
            from("direct:start")
                .aggregate(header("id"), new UseLatestAggregationStrategy())
                    .aggregationRepository(sharedAggregationRepository)
                    .optimisticLocking()
                    // trigger completion every 5th second
                    .completionInterval(getName().equals("testCamelContext1Wins") ? 5000 : 10000)
                .to("mock:result");
            // END SNIPPET: e1
        }
    };
}

项目：Camel 文件：DistributedCompletionIntervalTest.java

@Override
protected RouteBuilder createRouteBuilder2() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // START SNIPPET: e1
            from("direct:start")
                .aggregate(header("id"), new UseLatestAggregationStrategy())
                    .aggregationRepository(sharedAggregationRepository)
                    .optimisticLocking()
                    // trigger completion every 5th second
                    .completionInterval(getName().equals("testCamelContext1Wins") ? 10000 : 5000)
                .to("mock:result");
            // END SNIPPET: e1
        }
    };
}

项目：Camel 文件：AggregatorExceptionTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            final String exceptionString = "This is an Error not an Exception";
            errorHandler(deadLetterChannel("mock:error"));

            from("direct:start")
                .aggregate(header("id"), new UseLatestAggregationStrategy())
                .completionSize(5)
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        throw new java.lang.NoSuchMethodError(exceptionString);   
                    }
                });
        }
    };
}

项目：Camel 文件：FileMulticastDeleteTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("file:target/inbox?delete=true")
                .multicast(new UseLatestAggregationStrategy()).shareUnitOfWork()
                    .to("direct:foo", "direct:bar")
                .end()
                .convertBodyTo(String.class)
                .to("mock:result");

            from("direct:foo")
                .to("log:foo")
                .aggregate(header(Exchange.FILE_NAME), new MyFileAggregator()).completionTimeout(1000)
                    .convertBodyTo(String.class)
                    .to("mock:foo")
                .end();

            from("direct:bar")
                .to("log:bar")
                .convertBodyTo(String.class)
                .to("mock:bar");
        }
    };
}

项目：Camel 文件：AggregratedJmsRouteTest.java

protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        public void configure() throws Exception {
            from(timeOutEndpointUri).to("jms:queue:test.b");

            from("jms:queue:test.b").aggregate(header("cheese"), new AggregationStrategy() {
                public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        LOG.error("aggregration delay sleep inturrepted", e);
                        fail("aggregration delay sleep inturrepted");
                    }
                    return newExchange;
                }
            }).completionTimeout(2000L).to("mock:result");

            from(multicastEndpointUri).to("jms:queue:point1", "jms:queue:point2", "jms:queue:point3");
            from("jms:queue:point1").process(new MyProcessor()).to("jms:queue:reply");
            from("jms:queue:point2").process(new MyProcessor()).to("jms:queue:reply");
            from("jms:queue:point3").process(new MyProcessor()).to("jms:queue:reply");
            from("jms:queue:reply").aggregate(header("cheese"), new UseLatestAggregationStrategy()).completionSize(3)
                .to("mock:reply");
        }
    };
}

项目：ipf-flow-manager 文件：SplitterCopyAspectTest.java

@Before
public void setUp() throws Exception {
    CamelContext context = new DefaultCamelContext();
    messages = new ArrayList<>();
    splitter = new Splitter(
            context,
            new TestExpression(),
            new TestProcessor(),
            new UseLatestAggregationStrategy());
}

项目：ipf-flow-manager 文件：MulticastCopyAspectTest.java

@Before
public void setUp() throws Exception {
    CamelContext context = new DefaultCamelContext();
    Processor processor = new TestProcessor();
    messages = new ArrayList<>();
    multicast = new MulticastProcessor(context, Collections.nCopies(5, processor),
            new UseLatestAggregationStrategy());
}

项目：Camel 文件：PollEnrichBridgeErrorHandlerTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // try at most 3 times and if still failing move to DLQ
            errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(3).redeliveryDelay(0));

            from("seda:start")
                // bridge the error handler when doing a polling so we can let Camel's error handler decide what to do
                .pollEnrich("file:target/foo?pollStrategy=#myPoll&consumer.bridgeErrorHandler=true", 10000, new UseLatestAggregationStrategy())
                .to("mock:result");
        }
    };
}

项目：Camel 文件：PollEnricherRefTest.java

@Override
protected JndiRegistry createRegistry() throws Exception {
    JndiRegistry jndi = super.createRegistry();
    jndi.bind("cool", cool);
    jndi.bind("agg", new UseLatestAggregationStrategy());
    return jndi;
}

项目：Camel 文件：EnricherRefTest.java

@Override
protected JndiRegistry createRegistry() throws Exception {
    JndiRegistry jndi = super.createRegistry();
    jndi.bind("cool", cool);
    jndi.bind("agg", new UseLatestAggregationStrategy());
    return jndi;
}

项目：Camel 文件：SplitterTest.java

protected RouteBuilder createRouteBuilder() {
    return new RouteBuilder() {
        public void configure() {
            onException(CamelException.class).to("mock:failed");

            from("direct:seqential").split(body().tokenize(","), new UseLatestAggregationStrategy()).to("mock:result");
            from("direct:parallel").split(body().tokenize(","), new MyAggregationStrategy()).parallelProcessing().to("mock:result");
            from("direct:parallelAggregate").split(body().tokenize(","), new MyAggregationStrategy()).parallelProcessing().parallelAggregate().to("mock:result");
            from("direct:streaming").split(body().tokenize(",")).streaming().to("mock:result");
            from("direct:parallel-streaming").split(body().tokenize(","), new MyAggregationStrategy()).parallelProcessing().streaming().to("mock:result");
            from("direct:exception")
                    .split(body().tokenize(","))
                    .aggregationStrategy(new MyAggregationStrategy())
                    .parallelProcessing()
                    .process(new Processor() {
                        public void process(Exchange exchange) throws Exception {
                            String string = exchange.getIn().getBody(String.class);
                            if ("Exception".equals(string)) {
                                throw new CamelException("Just want to throw exception here");
                            }

                        }
                    }).to("mock:result");
            from("direct:simple").split(body()).to("mock:result");
        }
    };
}

项目：Camel 文件：AggregateTimeoutManualTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // START SNIPPET: e1
            from("direct:start")
                // timeout after 3 seconds
                .aggregate(header("id"), new UseLatestAggregationStrategy()).completionTimeout(3000)
                .to("mock:result");
            // END SNIPPET: e1
        }
    };
}

项目：Camel 文件：AggregateCompletedByBatchConsumerSendEmptyMessageWhenIdleTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("file:target/batch?sendEmptyMessageWhenIdle=true&delay=250")
                .aggregate(constant(true), new UseLatestAggregationStrategy()).completionFromBatchConsumer()
                    .to("mock:result");

        }
    };
}

项目：Camel 文件：AggregatorLockingTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("seda:a?concurrentConsumers=2")
                .aggregate(header("myId"), new UseLatestAggregationStrategy())
                .completionSize(1)
                // N.B. *no* parallelProcessing() nor optimisticLocking() !
                // each thread releases 1 permit and then blocks waiting for other threads.
                // if there are <THREAD_COUNT> threads running in parallel, then all N threads will release
                // and we will proceed. If the threads are prevented from running simultaneously due to the
                // lock in AggregateProcessor.doProcess() then only 1 thread will run and will not release
                // the current thread, causing the test to time out.
                .log("Before await with thread: ${threadName} and body: ${body}")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {
                        latch.countDown();
                        // block until the other thread counts down as well
                        if (!latch.await(5, TimeUnit.SECONDS)) {
                            throw new RuntimeException("Took too long; assume threads are blocked and fail test");
                        }
                    }
                })
                .log("After await with thread: ${threadName} and body: ${body}")
                .to("mock:result");
        }
    };
}

项目：Camel 文件：AggregateCompletionIntervalTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // START SNIPPET: e1
            from("seda:start")
                .aggregate(header("id"), new UseLatestAggregationStrategy())
                    // trigger completion every 5th second
                    .completionInterval(5000)
                .to("mock:result");
            // END SNIPPET: e1
        }
    };
}

项目：Camel 文件：AggregatorExceptionInPredicateTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            onException(IllegalArgumentException.class).handled(true).to("mock:handled");

            from("direct:start")
                .aggregate(header("id"))
                .completionTimeout(500)
                .aggregationStrategy(new AggregationStrategy() {

                    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                        Object body = newExchange.getIn().getBody();
                        if ("Damn".equals(body)) {
                            throw new IllegalArgumentException();
                        }
                        return newExchange;
                    }
                })
                .to("mock:result");

            from("direct:predicate")
                .aggregate(new Expression() {

                    public <T> T evaluate(Exchange exchange, Class<T> type) {
                        if (exchange.getIn().getBody().equals("Damn")) {
                            throw new IllegalArgumentException();
                        }
                        return ExpressionBuilder.headerExpression("id").evaluate(exchange, type);
                    }
                }, new UseLatestAggregationStrategy())
                .completionTimeout(500)
                .to("mock:result");
        }
    };
}

项目：Camel 文件：AggregateTimeoutWithNoExecutorServiceTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            for (int i = 0; i < AggregateTimeoutWithExecutorServiceTest.NUM_AGGREGATORS; ++i) {
                from("direct:start" + i)
                    // aggregate timeout after 3th seconds
                    .aggregate(header("id"), new UseLatestAggregationStrategy()).completionTimeout(3000)
                    .to("mock:result" + i);
            }
        }
    };
}

项目：Camel 文件：AggregateTimeoutOnlyTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // START SNIPPET: e1
            from("direct:start")
                // aggregate timeout after 3th seconds
                .aggregate(header("id"), new UseLatestAggregationStrategy()).completionTimeout(3000)
                .to("mock:result");
            // END SNIPPET: e1
        }
    };
}

项目：Camel 文件：PredicateAggregatorCollectionTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        public void configure() throws Exception {
            // START SNIPPET: e1
            // our route is aggregating from the direct queue and sending the response to the mock
            from("direct:start")
                // we use the collection based aggregator we already have configured
                .aggregate(header("id"), new UseLatestAggregationStrategy())
                .completionSize(3)
                .to("mock:result");
            // END SNIPPET: e1
        }
    };
}

项目：Camel 文件：ManagedRemoveRouteAggregateThreadPoolTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:foo").routeId("foo")
                .aggregate(constant(true), new UseLatestAggregationStrategy()).completionTimeout(1000)
                    .to("mock:result");
        }
    };
}

项目：Camel 文件：LevelDBBigPayloadTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("timer:foo")
                .bean(BigPayload.class)
                .aggregate(method(LevelDBBigPayloadTest.class, "number"), new UseLatestAggregationStrategy())
                    .aggregationRepository(repo)
                    .completionSize(2).completionTimeout(5000)
                    .log("Aggregated key ${header.CamelAggregatedCorrelationKey}");
        }
    };
}

项目：Camel 文件：HawtDBBigPayloadTest.java

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("timer:foo")
                .bean(BigPayload.class)
                .aggregate(method(HawtDBBigPayloadTest.class, "number"), new UseLatestAggregationStrategy())
                    .aggregationRepository(repo)
                    .completionSize(2).completionTimeout(5000)
                    .log("Aggregated key ${header.CamelAggregatedCorrelationKey}");
        }
    };
}

项目：Camel 文件：AggregationStrategies.java

/**
 * Use the latest incoming exchange.
 *
 * @see org.apache.camel.processor.aggregate.UseLatestAggregationStrategy
 */
public static AggregationStrategy useLatest() {
    return new UseLatestAggregationStrategy();
}