RomeSyndicationOperator extends SimpleSinglePortInputOperator<RomeFeedEntry> implements Runnable { @Override public void run() { try { while (true) { InputStreamReader isr = null; try { isr = new InputStreamReader(getFeedInputStream()); SyndFeedInput feedInput = new SyndFeedInput(); SyndFeed feed = feedInput.build(isr); List entries = feed.getEntries(); List<RomeFeedEntry> nfeedItems = new ArrayList<RomeFeedEntry>(); boolean oldEntries = false; for (int i = 0; i < entries.size(); ++i) { SyndEntry syndEntry = (SyndEntry)entries.get(i); RomeFeedEntry romeFeedEntry = getSerializableEntry(syndEntry); if (!oldEntries) { if (!feedItems.contains(romeFeedEntry)) { outputPort.emit(romeFeedEntry); } else if (orderedUpdate) { oldEntries = true; } } nfeedItems.add(romeFeedEntry); } feedItems = nfeedItems; } catch (Exception e) { logger.error(e.getMessage()); } finally { if (isr != null) { try { isr.close(); } catch (Exception ce) { logger.error(ce.getMessage()); } } } Thread.sleep(interval); } } catch (InterruptedException ie) { logger.error("Interrupted: " + ie.getMessage()); } } RomeSyndicationOperator(); void setLocation(String location); String getLocation(); void setInterval(int interval); int getInterval(); void setStreamProvider(RomeStreamProvider streamProvider); RomeStreamProvider getStreamProvider(); void setOrderedUpdate(boolean orderedUpdate); boolean isOrderedUpdate(); @Override void run();  }
@Test public void testRun() { LocalMode lma = LocalMode.newInstance(); DAG dag = lma.getDAG(); RomeSyndicationOperator rop = dag.addOperator("romesyndication", RomeSyndicationOperator.class); FeedCollector fc = dag.addOperator("feedcollector", FeedCollector.class); dag.addStream("ss", rop.outputPort, fc.input); rop.setInterval(2000); rop.setStreamProvider(new CNNSyndicationTestProvider()); try { LocalMode.Controller lc = lma.getController(); lc.setHeartbeatMonitoringEnabled(false); lc.run(10000); Assert.assertEquals("Entries size", entries.size(), 10); Assert.assertEquals("First entry title", entries.get(0).getSyndEntry().getTitle(), "Dimensions Computation (Aggregate Navigator) Part 1: Intro"); Assert.assertEquals("First entry URI", entries.get(0).getSyndEntry().getUri(), "https: Assert.assertEquals("Second run first entry title", entries.get(7).getSyndEntry().getTitle(), "Building Applications with Apache Apex and Malhar"); Assert.assertEquals("Second run first entry URI", entries.get(7).getSyndEntry().getUri(), "https: Assert.assertEquals("Last entry title", entries.get(9).getSyndEntry().getTitle(), "Dimensions Computation (Aggregate Navigator) Part 2: Implementation"); Assert.assertEquals("Last entry URI", entries.get(9).getSyndEntry().getUri(), "https: } catch (Exception ex) { logger.error(ex.getMessage()); Assert.assertFalse(true); } }