TransactionAnalyzer extends AbstractMessageAnalyzer<TransactionReport> implements LogEnabled { @Override public void process(MessageTree tree) { String domain = tree.getDomain(); TransactionReport report = m_reportManager.getHourlyReport(getStartTime(), domain, true); List<Transaction> transactions = tree.findOrCreateTransactions(); for (Transaction t : transactions) { String data = String.valueOf(t.getData()); if (data.length() > 0 && data.charAt(0) == CatConstants.BATCH_FLAG) { processBatchTransaction(tree, report, t, data); } else { processTransaction(report, tree, t); } } if (System.currentTimeMillis() > m_nextClearTime) { m_nextClearTime = m_nextClearTime + TimeHelper.ONE_MINUTE; Threads.forGroup("cat").start(new Runnable() { @Override public void run() { cleanUpReports(); } }); } }  @Override synchronized void doCheckpoint(boolean atEnd); @Override void enableLogging(Logger logger); Set<String> getDomains(); @Override TransactionReport getReport(String domain); @Override ReportManager<TransactionReport> getReportManager(); @Override void initialize(long startTime, long duration, long extraTime); @Override boolean isEligable(MessageTree tree); @Override void process(MessageTree tree); static final String ID; }
@Test public void testProcess() throws Exception { for (int i = 1; i <= 1000; i++) { MessageTree tree = ((DefaultMessageTree) generateMessageTree(i)).copyForTest(); m_analyzer.process(tree); } TransactionReport report = m_analyzer.getReport(m_domain); report.accept(new TransactionStatisticsComputer()); String expected = Files.forIO().readFrom(getClass().getResourceAsStream("transaction_analyzer.xml"), "utf-8"); Assert.assertTrue( TestHelper.isEquals(com.dianping.cat.consumer.transaction.model.transform.DefaultSaxParser.parse(expected), report)); }