@Test public void testStop() throws Exception { PipedInputStream pi = new PipedInputStream(); PipedOutputStream po = new PipedOutputStream(pi); PrintWriter prw = new PrintWriter(new OutputStreamWriter(po)); PirasoEntryWriter writer = new PirasoEntryWriter("1", "1", prw); final PirasoEntryReader reader = new PirasoEntryReader(pi); final List<Entry> entriesRead = new ArrayList<Entry>(); reader.addListener(new EntryReadAdapter() { @Override public void readEntry(EntryReadEvent evt) { synchronized (monitor) { entriesRead.add(evt.getEntry()); monitor.notifyAll(); } } }); final AtomicBoolean fail = new AtomicBoolean(false); ExecutorService executor = Executors.newFixedThreadPool(2); executor.submit(new Runnable() { public void run() { try { reader.start(); } catch (SAXException e) { } catch (Exception e) { e.printStackTrace(); fail.set(true); } } }); writer.write(new MessageEntry(1l, "hello")); writer.write(new MessageEntry(1l, "hello2")); while(entriesRead.size() != 2) { synchronized (monitor) { monitor.wait(1000l); } } executor.shutdown(); reader.stop(); writer.write(new MessageEntry(1l, "hello")); writer.write(new MessageEntry(1l, "hello")); writer.close(); if(fail.get()) { fail("failure see exception trace."); } assertEquals(2, entriesRead.size()); }
public void stop() { stopped = true; owningThread.interrupt(); }
PirasoEntryReader extends DefaultHandler { public void stop() { stopped = true; owningThread.interrupt(); } }
PirasoEntryReader extends DefaultHandler { public void stop() { stopped = true; owningThread.interrupt(); } PirasoEntryReader(InputStream in); }
PirasoEntryReader extends DefaultHandler { public void stop() { stopped = true; owningThread.interrupt(); } PirasoEntryReader(InputStream in); void start(); void stop(); @Override @SuppressWarnings("unchecked") void endElement(String uri, String localName, String qName); @Override void startElement(String uri, String localName, String qName, Attributes attributes); @Override void characters(char[] ch, int start, int length); String getId(); String getWatchedAddr(); void fireEntryReadStartedEvent(EntryReadEvent evt); void fireEntryReadEvent(EntryReadEvent evt); List<EntryReadListener> getListeners(); void addListener(EntryReadListener listener); void removeListener(EntryReadListener listener); void clearListeners(); }
PirasoEntryReader extends DefaultHandler { public void stop() { stopped = true; owningThread.interrupt(); } PirasoEntryReader(InputStream in); void start(); void stop(); @Override @SuppressWarnings("unchecked") void endElement(String uri, String localName, String qName); @Override void startElement(String uri, String localName, String qName, Attributes attributes); @Override void characters(char[] ch, int start, int length); String getId(); String getWatchedAddr(); void fireEntryReadStartedEvent(EntryReadEvent evt); void fireEntryReadEvent(EntryReadEvent evt); List<EntryReadListener> getListeners(); void addListener(EntryReadListener listener); void removeListener(EntryReadListener listener); void clearListeners();  }