Java 类org.apache.camel.builder.NotifyBuilder 实例源码

@Test
public void testRouteFileToFile() throws Exception {
    deleteDirectory("target/file2file");
    NotifyBuilder notify = new NotifyBuilder(context).from("activemq:queue:hello").whenDone(1).create();

    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMessageCount(1);

    template.sendBodyAndHeader("file://target/file2file/in", "Hello World", Exchange.FILE_NAME, "hello.txt");

    assertMockEndpointsSatisfied();

    notify.matchesMockWaitTime();

    File file = new File("target/file2file/out/hello.txt");
    assertTrue("The file should exists", file.exists());
}

@Test
public void testNotifyFrom() throws Exception {
    // use from to indicate it should only be messages originating from the given endpoint
    NotifyBuilder notify = new NotifyBuilder(context)
            .from("seda:order").whenDone(1).create();

    template.sendBody("seda:quote", "Camel rocks");
    template.sendBody("seda:order", "123,2010-04-20'T'15:47:59,4444,5555");

    boolean matches = notify.matches(1, TimeUnit.SECONDS);
    assertTrue(matches);

    SedaEndpoint confirm = context.getEndpoint("seda:confirm", SedaEndpoint.class);
    assertEquals(1, confirm.getExchanges().size());
    assertEquals("OK,123,2010-04-20'T'15:47:59,4444,5555", confirm.getExchanges().get(0).getIn().getBody());
}

public void testBurstySampling() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(5).create();

    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMinimumMessageCount(2);
    mock.setResultWaitTime(3000);

    List<Exchange> sentExchanges = new ArrayList<Exchange>();

    // send a burst of 5 exchanges, expecting only one to get through
    sendExchangesThroughDroppingThrottler(sentExchanges, 5);
    // sleep through a complete period
    Thread.sleep(1100);
    // send another 5 now
    sendExchangesThroughDroppingThrottler(sentExchanges, 5);

    notify.matchesMockWaitTime();
    mock.assertIsSatisfied();

    validateDroppedExchanges(sentExchanges, mock.getReceivedCounter());
}

public void testZipFileUnmarshalDelete() throws Exception {
    // there are 2 exchanges
    NotifyBuilder event = event().whenDone(2).create();

    getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
    template.sendBodyAndHeader("file:target/zip", "Hello World", Exchange.FILE_NAME, "hello.txt");
    assertMockEndpointsSatisfied();

    event.matchesMockWaitTime();

    File in = new File("target/zip/hello.txt");
    assertFalse("Should have been deleted " + in, in.exists());

    File out = new File("target/zip/out/hello.txt.zip");
    assertFalse("Should have been deleted " + out, out.exists());
}

public void xxxtestTokenPairPerformanceRoute() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(size).create();

    boolean matches = notify.matches(5, TimeUnit.MINUTES);
    log.info("Processed file with " + size + " elements in: " + TimeUtils.printDuration(watch.stop()));

    log.info("Processed " + tiny.get() + " tiny messages");
    log.info("Processed " + small.get() + " small messages");
    log.info("Processed " + med.get() + " medium messages");
    log.info("Processed " + large.get() + " large messages");

    assertEquals((size / 10) * 4, tiny.get());
    assertEquals((size / 10) * 2, small.get());
    assertEquals((size / 10) * 3, med.get());
    assertEquals((size / 10) * 1, large.get());

    assertTrue("Should complete route", matches);
}

public void xxTestXPatPerformanceRoute() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(size).create();

    boolean matches = notify.matches(60, TimeUnit.SECONDS);
    log.info("Processed file with " + size + " elements in: " + TimeUtils.printDuration(watch.stop()));

    log.info("Processed " + tiny.get() + " tiny messages");
    log.info("Processed " + small.get() + " small messages");
    log.info("Processed " + med.get() + " medium messages");
    log.info("Processed " + large.get() + " large messages");

    assertEquals((size / 10) * 4, tiny.get());
    assertEquals((size / 10) * 2, small.get());
    assertEquals((size / 10) * 3, med.get());
    assertEquals((size / 10) * 1, large.get());

    assertTrue("Should complete route", matches);
}

public void testInOnly() throws Exception {
    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start").to("seda:foo?timeout=1000");
        }
    });

    context.start();
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    // no problem for in only as we do not expect a reply
    template.sendBody("direct:start", "Hello World");
    notify.matches(2, TimeUnit.SECONDS);

}

public void testFileConsumeDoneFileIssue() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(5).create();

    template.sendBodyAndHeader("file:target/done", "A", Exchange.FILE_NAME, "foo-a.txt");
    template.sendBodyAndHeader("file:target/done", "B", Exchange.FILE_NAME, "foo-b.txt");
    template.sendBodyAndHeader("file:target/done", "C", Exchange.FILE_NAME, "foo-c.txt");
    template.sendBodyAndHeader("file:target/done", "D", Exchange.FILE_NAME, "foo-d.txt");
    template.sendBodyAndHeader("file:target/done", "E", Exchange.FILE_NAME, "foo-e.txt");
    template.sendBodyAndHeader("file:target/done", "E", Exchange.FILE_NAME, "foo.done");

    assertTrue("Done file should exists", new File("target/done/foo.done").exists());

    getMockEndpoint("mock:result").expectedBodiesReceivedInAnyOrder("A", "B", "C", "D", "E");

    context.startRoute("foo");

    assertMockEndpointsSatisfied();
    assertTrue(notify.matchesMockWaitTime());

    Thread.sleep(250);

    // the done file should be deleted
    assertFalse("Done file should be deleted", new File("target/done/foo.done").exists());
}

public void testPollFileAndShouldBeDeletedAtThirdPoll() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(3).create();

    template.sendBodyAndHeader("file://target/deletefile", body, Exchange.FILE_NAME, "hello.txt");
    context.startRoute("FromFilePollThirdTimeOkTest");

    getMockEndpoint("mock:result").expectedBodiesReceived(body);

    assertMockEndpointsSatisfied();
    assertTrue(notify.matchesMockWaitTime());
    assertEquals(3, counter);

    // assert the file is deleted
    File file = new File("target/deletefile/hello.txt");
    assertFalse("The file should have been deleted", file.exists());
}

@Test
public void testJmsToJdbcJmsCommit() throws Exception {
    checkInitialState();

    // use a notify to know when the message is done
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    // use mock during testing as well
    getMockEndpoint("mock:a").expectedMessageCount(1);
    getMockEndpoint("mock:b").expectedMessageCount(1);

    template.sendBodyAndHeader("activemq:queue:inbox", "A", "uid", 123);

    // assert mock and wait for the message to be done
    assertMockEndpointsSatisfied();
    assertTrue("Should complete 1 message", notify.matchesMockWaitTime());

    // check that there is a message in the database and JMS queue
    assertEquals(new Integer(1), jdbcTemplate.queryForObject("select count(*) from CAMEL_MESSAGEPROCESSED", Integer.class));
    Object out = consumer.receiveBody("activemq:queue:outbox", 3000);
    assertEquals("DONE-A", out);
}

@Test
public void testFilterIdempotent() throws Exception {
    checkInitialState();

    // use a notify to know when the message is done
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(3).create();

    // use mock during testing as well
    getMockEndpoint("mock:a").expectedMessageCount(3);
    // there should be 1 duplicate
    getMockEndpoint("mock:b").expectedMessageCount(2);

    template.sendBodyAndHeader("activemq:queue:inbox", "D", "uid", 111);
    template.sendBodyAndHeader("activemq:queue:inbox", "E", "uid", 222);
    template.sendBodyAndHeader("activemq:queue:inbox", "D", "uid", 111);

    // assert mock and wait for the message to be done
    assertMockEndpointsSatisfied();
    assertTrue("Should complete 3 messages", notify.matchesMockWaitTime());

    // check that there is two messages in the database and JMS queue
    assertEquals(new Integer(2), jdbcTemplate.queryForObject("select count(*) from CAMEL_MESSAGEPROCESSED", Integer.class));
    assertEquals("DONE-D", consumer.receiveBody("activemq:queue:outbox", 3000));
    assertEquals("DONE-E", consumer.receiveBody("activemq:queue:outbox", 3000));
}

@Test
public void testFromFileToFtpDelete() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMessageCount(1);

    template.sendBodyAndHeader("file:target/delete", "Hello World", Exchange.FILE_NAME, "hello.txt");

    assertMockEndpointsSatisfied();
    assertTrue(notify.matchesMockWaitTime());

    // file should be deleted
    File file = new File("target/delete/hello.txt");
    assertFalse("File should be deleted", file.exists());

    // file should exists on ftp server
    file = new File(FTP_ROOT_DIR + "/hello.txt");
    assertTrue("File should exist on ftp server", file.exists());
}

@Test
public void testZipkinRoute() throws Exception {
    // JMX tests dont work well on AIX CI servers (hangs them)
    if (isPlatform("aix")) {
        return;
    }

    MBeanServer mbeanServer = getMBeanServer();
    ObjectName on = new ObjectName("org.apache.camel:context=camel-1,type=services,name=ZipkinTracer");
    assertNotNull(on);
    assertTrue(mbeanServer.isRegistered(on));

    Float rate = (Float) mbeanServer.getAttribute(on, "Rate");
    assertEquals("Should be 1.0f", 1.0f, rate.floatValue(), 0.1f);

    NotifyBuilder notify = new NotifyBuilder(context).whenDone(5).create();

    for (int i = 0; i < 5; i++) {
        template.sendBody("seda:dude", "Hello World");
    }

    assertTrue(notify.matches(30, TimeUnit.SECONDS));
}

@Test
public void testXPathSTaXPerformanceRoute() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(size).create();

    boolean matches = notify.matches(60, TimeUnit.SECONDS);
    log.info("Processed file with " + size + " elements in: " + TimeUtils.printDuration(watch.stop()));

    log.info("Processed " + tiny.get() + " tiny messages");
    log.info("Processed " + small.get() + " small messages");
    log.info("Processed " + med.get() + " medium messages");
    log.info("Processed " + large.get() + " large messages");

    assertEquals((size / 10) * 4, tiny.get());
    assertEquals((size / 10) * 2, small.get());
    assertEquals((size / 10) * 3, med.get());
    assertEquals((size / 10) * 1, large.get());

    assertTrue("Should complete route", matches);
}

@Test
public void testReuse() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    getMockEndpoint("mock:a").expectedBodiesReceived("Hello World");
    getMockEndpoint("mock:b").expectedBodiesReceived("Hello Hello World");
    getMockEndpoint("mock:result").expectedBodiesReceived("Hello World", "Hello Hello World");

    template.sendBody("direct:start", "World\n");

    assertMockEndpointsSatisfied();

    assertTrue(notify.matchesMockWaitTime());

    assertEquals(2, channels.size());
    assertSame("Should reuse channel", channels.get(0), channels.get(1));
    assertFalse("And closed when routing done", channels.get(0).isOpen());
    assertFalse("And closed when routing done", channels.get(1).isOpen());
}

@Test
public void testSendMultipleMessages() throws Exception {
    int messageCount = 5;
    acknowledged.setExpectedMessageCount(messageCount);
    timeout.setExpectedMessageCount(0);
    frame.setExpectedMessageCount(0);

    NotifyBuilder[] complete = new NotifyBuilder[messageCount];
    for (int i = 0; i < messageCount; ++i) {
        complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
    }

    for (int i = 0; i < messageCount; ++i) {
        source.sendBody(generateMessage(i + 1));
        assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
    }

    assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
}

@Test
public void testNoResponseOnFirstMessage() throws Exception {
    int sendMessageCount = 5;
    acknowledged.setExpectedMessageCount(sendMessageCount - 1);
    timeout.expectedMessageCount(1);
    frame.setExpectedMessageCount(0);

    NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
    for (int i = 0; i < sendMessageCount; ++i) {
        complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
    }

    mllpServer.disableResponse();

    source.sendBody(generateMessage(1));
    assertTrue("Messege 1 not completed", complete[0].matches(1, TimeUnit.SECONDS));

    mllpServer.enableResponse();

    for (int i = 1; i < sendMessageCount; ++i) {
        source.sendBody(generateMessage(i + 1));
        assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
    }

    assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
}

@Test
public void testNoResponseOnNthMessage() throws Exception {
    int sendMessageCount = 3;
    acknowledged.setExpectedMessageCount(sendMessageCount - 1);
    timeout.expectedMessageCount(1);
    frame.setExpectedMessageCount(0);

    NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
    for (int i = 0; i < sendMessageCount; ++i) {
        complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
    }

    mllpServer.disableResponse(sendMessageCount);

    for (int i = 0; i < sendMessageCount; ++i) {
        source.sendBody(generateMessage(i + 1));
        assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
    }

    assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
}

@Test
public void testMissingEndOfDataByte() throws Exception {
    int sendMessageCount = 3;
    acknowledged.setExpectedMessageCount(sendMessageCount - 1);

    NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
    for (int i = 0; i < sendMessageCount; ++i) {
        complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
    }

    mllpServer.setExcludeEndOfDataModulus(sendMessageCount);

    for (int i = 0; i < sendMessageCount; ++i) {
        source.sendBody(generateMessage(i + 1));
        assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
    }

    assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
}

@Test
public void testMissingEndOfBlockByte() throws Exception {
    int sendMessageCount = 3;
    acknowledged.setExpectedMessageCount(sendMessageCount - 1);

    NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
    for (int i = 0; i < sendMessageCount; ++i) {
        complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
    }

    mllpServer.setExcludeEndOfBlockModulus(sendMessageCount);

    for (int i = 0; i < sendMessageCount; ++i) {
        source.sendBody(generateMessage(i + 1));
        assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
    }

    assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
}

@Test
public void testApplicationAcceptAcknowledgement() throws Exception {
    int sendMessageCount = 5;
    acknowledged.setExpectedMessageCount(sendMessageCount);

    NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
    for (int i = 0; i < sendMessageCount; ++i) {
        complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
    }

    for (int i = 0; i < sendMessageCount; ++i) {
        source.sendBody(generateMessage(i + 1));
        assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
    }

    assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
}

@Test
public void testTarToFileWithoutFileName() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    String[] files = TEST_DIR.list();
    assertTrue(files == null || files.length == 0);

    MockEndpoint mock = getMockEndpoint("mock:intercepted");
    mock.expectedMessageCount(1);

    template.sendBody("direct:tarToFile", TEXT);

    assertMockEndpointsSatisfied();

    // use builder to ensure the exchange is fully done before we check for file exists
    assertTrue(notify.matches(5, TimeUnit.SECONDS));

    Exchange exchange = mock.getReceivedExchanges().get(0);
    File file = new File(TEST_DIR, exchange.getIn().getMessageId() + ".tar");
    assertTrue(file.exists());
    assertArrayEquals(getTaredText(exchange.getIn().getMessageId()), getBytes(file));
}

@Test
public void testTarToFileWithFileName() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    MockEndpoint mock = getMockEndpoint("mock:tarToFile");
    mock.expectedMessageCount(1);

    File file = new File(TEST_DIR, "poem.txt.tar");
    assertFalse(file.exists());

    template.sendBodyAndHeader("direct:tarToFile", TEXT, FILE_NAME, "poem.txt");

    // just make sure the file is created
    mock.assertIsSatisfied();

    // use builder to ensure the exchange is fully done before we check for file exists
    assertTrue(notify.matches(5, TimeUnit.SECONDS));

    assertTrue(file.exists());
    assertArrayEquals(getTaredText("poem.txt"), getBytes(file));
}

@Test
public void testTarToFileWithoutFileName() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    String[] files = TEST_DIR.list();
    assertTrue(files == null || files.length == 0);

    MockEndpoint mock = getMockEndpoint("mock:intercepted");
    mock.expectedMessageCount(1);

    template.sendBody("direct:tarToFile", TEXT);

    assertMockEndpointsSatisfied();

    // use builder to ensure the exchange is fully done before we check for file exists
    assertTrue(notify.matches(5, TimeUnit.SECONDS));

    Exchange exchange = mock.getReceivedExchanges().get(0);
    File file = new File(TEST_DIR, exchange.getIn().getMessageId() + ".tar");
    assertTrue(file.exists());
    assertTrue(ObjectHelper.equalByteArray(getTaredText(exchange.getIn().getMessageId()), getBytes(file)));
}

@Test
public void testTarToFileWithFileName() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    MockEndpoint mock = getMockEndpoint("mock:tarToFile");
    mock.expectedMessageCount(1);

    File file = new File(TEST_DIR, "poem.txt.tar");
    assertFalse(file.exists());

    template.sendBodyAndHeader("direct:tarToFile", TEXT, FILE_NAME, "poem.txt");

    // just make sure the file is created
    mock.assertIsSatisfied();

    // use builder to ensure the exchange is fully done before we check for file exists
    assertTrue(notify.matches(5, TimeUnit.SECONDS));

    assertTrue(file.exists());
    assertTrue(ObjectHelper.equalByteArray(getTaredText("poem.txt"), getBytes(file)));
}

@Test
public void testAddingDataToCacheWithNonStringCacheKey() throws Exception {
    context.addRoutes(new RouteBuilder() {
        public void configure() {
            from("direct:a").
                    setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_ADD)).
                    setHeader(CacheConstants.CACHE_KEY, constant(10L)).
                    to("cache://TestCache1");
        }
    });
    context.start();
    NotifyBuilder notify = new NotifyBuilder(context).whenExactlyDone(1).create();

    log.debug("------------Beginning CacheProducer Add Test---------------");
    sendOriginalFile();

    notify.matches(10, TimeUnit.SECONDS);
    assertNotNull(fetchElement("10"));
}

@Test
@Ignore("Manual test")
public void testXPathPerformanceRoute() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(size).create();

    boolean matches = notify.matches(60, TimeUnit.SECONDS);
    log.info("Processed file with " + size + " elements in: " + TimeUtils.printDuration(watch.stop()));

    log.info("Processed " + tiny.get() + " tiny messages");
    log.info("Processed " + small.get() + " small messages");
    log.info("Processed " + med.get() + " medium messages");
    log.info("Processed " + large.get() + " large messages");

    assertEquals((size / 10) * 4, tiny.get());
    assertEquals((size / 10) * 2, small.get());
    assertEquals((size / 10) * 3, med.get());
    assertEquals((size / 10) * 1, large.get());

    assertTrue("Should complete route", matches);
}

@Test
public void consumeMessagesFromQueue() throws Exception {
    // simple test to make sure that concurrent consumers were used in the test

    NotifyBuilder notifier = new NotifyBuilder(context).whenCompleted(NUM_MESSAGES).create();
    assertTrue("We didn't process " + NUM_MESSAGES + " messages as we expected!", notifier.matches(5, TimeUnit.SECONDS));

    if (isPlatform("windows")) {
        // threading is different on windows
    } else {
        // usually we use all threads evenly but sometimes threads are reused so just test that 50%+ was used
        if (threadNumbers.size() < (NUM_CONCURRENT / 2)) {
            fail(String.format("We were expecting to have about half of %d numbers of concurrent consumers, but only found %d",
                    NUM_CONCURRENT, threadNumbers.size()));
        }
    }
}

@Test
public void testZipToFileWithoutFileName() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    String[] files = TEST_DIR.list();
    assertTrue(files == null || files.length == 0);

    MockEndpoint mock = getMockEndpoint("mock:intercepted");
    mock.expectedMessageCount(1);

    template.sendBody("direct:zipToFile", TEXT);

    assertMockEndpointsSatisfied();

    // use builder to ensure the exchange is fully done before we check for file exists
    assertTrue("The exchange is not done in time.", notify.matches(5, TimeUnit.SECONDS));

    Exchange exchange = mock.getReceivedExchanges().get(0);
    File file = new File(TEST_DIR, exchange.getIn().getMessageId() + ".zip");
    assertTrue("The file should exist.", file.exists());
    assertArrayEquals("Get a wrong message content.", getZippedText(exchange.getIn().getMessageId()), getBytes(file));
}

@Test
public void testZipToFileWithFileName() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    MockEndpoint mock = getMockEndpoint("mock:zipToFile");
    mock.expectedMessageCount(1);

    File file = new File(TEST_DIR, "poem.txt.zip");
    assertFalse("The zip should not exit.", file.exists());

    template.sendBodyAndHeader("direct:zipToFile", TEXT, FILE_NAME, "poem.txt");

    // just make sure the file is created
    mock.assertIsSatisfied();

    // use builder to ensure the exchange is fully done before we check for file exists
    assertTrue("The exchange is not done in time.", notify.matches(5, TimeUnit.SECONDS));

    assertTrue("The file should exist.", file.exists());
    assertArrayEquals("Get a wrong message content.", getZippedText("poem.txt"), getBytes(file));
}

@Test
public void testFileThreads() throws Exception {
    log.info("Creating {} files...", files);

    // create many files
    for (int i = 0; i < files; i++) {
        template.sendBodyAndHeader("file:target/inbox", "Message " + i, Exchange.FILE_NAME, "file-" + i + ".txt");
    }

    log.info("Starting route");

    // keep track of completed files
    NotifyBuilder notify = new NotifyBuilder(context).whenCompleted(files).create();

    // start route
    context.startRoute("myRoute");

    // wait for all files to be processed
    assertTrue("Should complete all files", notify.matches(60, TimeUnit.SECONDS));
}

@Test
public void testFileThreads() throws Exception {
    log.info("Creating {} files...", files);

    // create many files
    for (int i = 0; i < files; i++) {
        template.sendBodyAndHeader("file:target/inbox", "Message " + i, Exchange.FILE_NAME, "file-" + i + ".txt");
    }

    log.info("Starting route");

    // keep track of completed files
    NotifyBuilder notify = new NotifyBuilder(context).whenCompleted(files).create();

    // start route
    context.startRoute("myRoute");

    // wait for all files to be processed
    assertTrue("Should complete all files", notify.matches(60, TimeUnit.SECONDS));
}

@Test
public void testThrottlerReporter() throws Exception {
    // use notifier to known when we have processed all the messages
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(size).create();

    // create the reporter using injector so Camel can do dependency injection
    RiderThrottlerReporter reporter = context.getInjector().newInstance(RiderThrottlerReporter.class);

    // schedule a background task that logs the current throttle count
    scheduler = context.getExecutorServiceManager().newSingleThreadScheduledExecutor(this, "ThrottleReporter");
    scheduler.scheduleAtFixedRate(new ThrottleTask(reporter), 1, 1, TimeUnit.SECONDS);

    // send some orders
    for (int i = 0; i < size; i++) {
        template.sendBody("seda:orders", "Order " + size);
    }

    // wait for all messages to be done
    log.info("Waiting to process all the messages...");
    assertTrue("Should process all messages", notify.matches(1, TimeUnit.MINUTES));

    // shutdown thread pool
    context.getExecutorServiceManager().shutdown(scheduler);
}

@Test
public void testSendPartnerReportIntoDatabase() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    // start Camel manually as we use advice-with in this unit tests class
    context.start();

    // there should be 0 row in the database when we start
    int rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(0, rows);

    String xml = "<?xml version=\"1.0\"?><partner id=\"123\"><date>201702250815</date><code>200</code><time>4387</time></partner>";
    template.sendBody("activemq:queue:partners", xml);

    // wait for the route to complete one message
    assertTrue(notify.matches(10, TimeUnit.SECONDS));

    // there should be 1 row in the database
    rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(1, rows);
}

@Test
public void testSendPartnerReportIntoDatabase() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    // there should be 0 row in the database when we start
    int rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(0, rows);

    String xml = "<?xml version=\"1.0\"?><partner id=\"123\"><date>201702250815</date><code>200</code><time>4387</time></partner>";
    template.sendBody("activemq:queue:partners", xml);

    // wait for the route to complete one message
    assertTrue(notify.matches(10, TimeUnit.SECONDS));

    // there should be 1 row in the database
    rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(1, rows);
}

@Test
public void testSendPartnerReportIntoDatabase() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    // start Camel manually as we use advice-with in this unit tests class
    context.start();

    // there should be 0 row in the database when we start
    int rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(0, rows);

    String xml = "<?xml version=\"1.0\"?><partner id=\"123\"><date>201702250815</date><code>200</code><time>4387</time></partner>";
    template.sendBody("activemq:queue:partners", xml);

    // wait for the route to complete one message
    assertTrue(notify.matches(10, TimeUnit.SECONDS));

    // there should be 1 row in the database
    rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(1, rows);
}

@Test
public void testRollbackBeforeActiveMQ() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenReceived(10).create();

    jdbc.execute("insert into partner_metric (partner_id, time_occurred, status_code, perf_time) values ('123', '20170315183457', '200', '1503')");

    assertTrue(notify.matches(15, TimeUnit.SECONDS));

    // and there should be 1 row in the database as it keep rolling back
    int rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(1, rows);

    String order = consumer.receiveBody("activemq:queue:order", 2000, String.class);
    assertNull("Should NOT be in order queue", order);

    context.stop();
}

@Test
public void testCommit() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    jdbc.execute("insert into partner_metric (partner_id, time_occurred, status_code, perf_time) values ('123', '20170315183457', '200', '1503')");

    assertTrue(notify.matches(10, TimeUnit.SECONDS));

    // give time for database
    Thread.sleep(1000);

    // and there should be 0 rows in the database
    int rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(0, rows);

    String order = consumer.receiveBody("activemq:queue:order", 2000, String.class);
    assertNotNull("Should be in order queue", order);

    context.stop();
}

@Test
public void testRollbackAfterActiveMQ() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenReceived(10).create();

    jdbc.execute("insert into partner_metric (partner_id, time_occurred, status_code, perf_time) values ('123', '20170315183457', '200', '1503')");

    assertTrue(notify.matches(15, TimeUnit.SECONDS));

    // and there should be 1 row in the database as it keep rolling back
    int rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(1, rows);

    String order = consumer.receiveBody("activemq:queue:order", 2000, String.class);
    assertNull("Should NOT be in order queue", order);

    context.stop();
}

@Test
public void testMoveFile() throws Exception {
    // use NotifyBuilder to wait for the file to be routed
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    // create a new file in the inbox folder with the name hello.txt and containing Hello World as body
    template.sendBodyAndHeader("file://target/inbox", "Hello World", Exchange.FILE_NAME, "hello.txt");

    // notifier will wait for the file to be processed
    // and if that never happen it will time out after 10 seconds (default mock timeout)
    assertTrue(notify.matchesMockWaitTime());

    // test the file was moved
    File target = new File("target/outbox/hello.txt");
    assertTrue("File should have been moved", target.exists());

    // test that its content is correct as well
    String content = context.getTypeConverter().convertTo(String.class, target);
    assertEquals("Hello World", content);
}