@Test public void testTail() throws Exception { EventTailCommand tail = new EventTailCommand(); tail.session = mock(Session.class); tail.collector = new EventCollector(); PrintStream out = System.out; expect(tail.session.getConsole()).andReturn(out); exception = null; replay(tail.session); ExecutorService executor = Executors.newSingleThreadExecutor(); executor.execute(() -> { try { tail.execute(); } catch (Exception e) { exception = e; } }); tail.collector.handleEvent(event()); Thread.sleep(200); executor.shutdownNow(); executor.awaitTermination(10, TimeUnit.SECONDS); if (exception != null) { throw exception; } verify(tail.session); }
@Override public Object execute() throws Exception { EventPrinter printer = new EventPrinter(session.getConsole(), verbose); Consumer<Event> filteredPrinter = executeIf(matchTopic(topicFilter), printer); collector.addConsumer(filteredPrinter); try { waitTillInterrupted(); } catch (InterruptedException e) { collector.removeConsumer(filteredPrinter); } return null; }
EventTailCommand implements Action { @Override public Object execute() throws Exception { EventPrinter printer = new EventPrinter(session.getConsole(), verbose); Consumer<Event> filteredPrinter = executeIf(matchTopic(topicFilter), printer); collector.addConsumer(filteredPrinter); try { waitTillInterrupted(); } catch (InterruptedException e) { collector.removeConsumer(filteredPrinter); } return null; } }
EventTailCommand implements Action { @Override public Object execute() throws Exception { EventPrinter printer = new EventPrinter(session.getConsole(), verbose); Consumer<Event> filteredPrinter = executeIf(matchTopic(topicFilter), printer); collector.addConsumer(filteredPrinter); try { waitTillInterrupted(); } catch (InterruptedException e) { collector.removeConsumer(filteredPrinter); } return null; }  }
EventTailCommand implements Action { @Override public Object execute() throws Exception { EventPrinter printer = new EventPrinter(session.getConsole(), verbose); Consumer<Event> filteredPrinter = executeIf(matchTopic(topicFilter), printer); collector.addConsumer(filteredPrinter); try { waitTillInterrupted(); } catch (InterruptedException e) { collector.removeConsumer(filteredPrinter); } return null; }  @Override Object execute(); }
EventTailCommand implements Action { @Override public Object execute() throws Exception { EventPrinter printer = new EventPrinter(session.getConsole(), verbose); Consumer<Event> filteredPrinter = executeIf(matchTopic(topicFilter), printer); collector.addConsumer(filteredPrinter); try { waitTillInterrupted(); } catch (InterruptedException e) { collector.removeConsumer(filteredPrinter); } return null; }  @Override Object execute();  }