397398399400401402403404405406407
ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); Exception e = e4.getException(); assertNotNull(e); assertEquals("The correlation key [123] has been closed. Exchange[Message: C]", e.getMessage()); assertMockEndpointsSatisfied();
457458459460461462463464465466467
e5.getIn().setHeader("id", 123); e5.setProperty(Exchange.BATCH_INDEX, 2); e5.setProperty(Exchange.BATCH_SIZE, 3); e5.setProperty(Exchange.BATCH_COMPLETE, true); ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); ap.process(e5);
458459460461462463464465466467468
e5.setProperty(Exchange.BATCH_INDEX, 2); e5.setProperty(Exchange.BATCH_SIZE, 3); e5.setProperty(Exchange.BATCH_COMPLETE, true); ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); ap.process(e5); assertMockEndpointsSatisfied();
459460461462463464465466467468469
e5.setProperty(Exchange.BATCH_SIZE, 3); e5.setProperty(Exchange.BATCH_COMPLETE, true); ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); ap.process(e5); assertMockEndpointsSatisfied();
460461462463464465466467468469470
e5.setProperty(Exchange.BATCH_COMPLETE, true); ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); ap.process(e5); assertMockEndpointsSatisfied(); ap.stop();
461462463464465466467468469470471
ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); ap.process(e5); assertMockEndpointsSatisfied(); ap.stop(); }
8182838485868788899091
Exchange e4 = new DefaultExchange(context); e4.getIn().setBody("D"); e4.getIn().setHeader("id", 123); ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); assertMockEndpointsSatisfied();
8283848586878889909192
8384858687888990919293
e4.getIn().setBody("D"); e4.getIn().setHeader("id", 123); ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); assertMockEndpointsSatisfied(); ap.stop();
8485868788899091929394
e4.getIn().setHeader("id", 123); ap.process(e1); ap.process(e2); ap.process(e3); ap.process(e4); assertMockEndpointsSatisfied(); ap.stop(); }