@Test public void shouldLogOnlyStreamData() throws Throwable { loggableObservable.get(observableRule.stringType()).subscribe(subscriber); verify(messageManager).printObservableOnNextWithValue(any(ObservableInfo.class), anyString()); verify(messageManager).printObservableItemTimeInfo(any(ObservableInfo.class)); verifyNoMoreInteractions(messageManager); }
@Override <T> Observable<T> get(T type) throws Throwable { final StopWatch stopWatch = new StopWatch(); final Counter emittedItems = new Counter(joinPoint.getMethodName()); return ((Observable<T>) joinPoint.proceed()) .doOnSubscribe(new Action0() { @Override public void call() { stopWatch.start(); } }) .doOnNext(new Action1<T>() { @Override public void call(T value) { emittedItems.increment(); messageManager.printObservableOnNextWithValue(observableInfo, value); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { messageManager.printObservableOnError(observableInfo, throwable); } }) .doOnTerminate(new Action0() { @Override public void call() { stopWatch.stop(); observableInfo.setTotalExecutionTime(stopWatch.getTotalTimeMillis()); observableInfo.setTotalEmittedItems(emittedItems.tally()); messageManager.printObservableItemTimeInfo(observableInfo); } }); }
LogStreamObservable extends LoggableObservable { @Override <T> Observable<T> get(T type) throws Throwable { final StopWatch stopWatch = new StopWatch(); final Counter emittedItems = new Counter(joinPoint.getMethodName()); return ((Observable<T>) joinPoint.proceed()) .doOnSubscribe(new Action0() { @Override public void call() { stopWatch.start(); } }) .doOnNext(new Action1<T>() { @Override public void call(T value) { emittedItems.increment(); messageManager.printObservableOnNextWithValue(observableInfo, value); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { messageManager.printObservableOnError(observableInfo, throwable); } }) .doOnTerminate(new Action0() { @Override public void call() { stopWatch.stop(); observableInfo.setTotalExecutionTime(stopWatch.getTotalTimeMillis()); observableInfo.setTotalEmittedItems(emittedItems.tally()); messageManager.printObservableItemTimeInfo(observableInfo); } }); } }
LogStreamObservable extends LoggableObservable { @Override <T> Observable<T> get(T type) throws Throwable { final StopWatch stopWatch = new StopWatch(); final Counter emittedItems = new Counter(joinPoint.getMethodName()); return ((Observable<T>) joinPoint.proceed()) .doOnSubscribe(new Action0() { @Override public void call() { stopWatch.start(); } }) .doOnNext(new Action1<T>() { @Override public void call(T value) { emittedItems.increment(); messageManager.printObservableOnNextWithValue(observableInfo, value); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { messageManager.printObservableOnError(observableInfo, throwable); } }) .doOnTerminate(new Action0() { @Override public void call() { stopWatch.stop(); observableInfo.setTotalExecutionTime(stopWatch.getTotalTimeMillis()); observableInfo.setTotalEmittedItems(emittedItems.tally()); messageManager.printObservableItemTimeInfo(observableInfo); } }); } LogStreamObservable(FrodoProceedingJoinPoint joinPoint,
      MessageManager messageManager, ObservableInfo observableInfo); }
LogStreamObservable extends LoggableObservable { @Override <T> Observable<T> get(T type) throws Throwable { final StopWatch stopWatch = new StopWatch(); final Counter emittedItems = new Counter(joinPoint.getMethodName()); return ((Observable<T>) joinPoint.proceed()) .doOnSubscribe(new Action0() { @Override public void call() { stopWatch.start(); } }) .doOnNext(new Action1<T>() { @Override public void call(T value) { emittedItems.increment(); messageManager.printObservableOnNextWithValue(observableInfo, value); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { messageManager.printObservableOnError(observableInfo, throwable); } }) .doOnTerminate(new Action0() { @Override public void call() { stopWatch.stop(); observableInfo.setTotalExecutionTime(stopWatch.getTotalTimeMillis()); observableInfo.setTotalEmittedItems(emittedItems.tally()); messageManager.printObservableItemTimeInfo(observableInfo); } }); } LogStreamObservable(FrodoProceedingJoinPoint joinPoint,
      MessageManager messageManager, ObservableInfo observableInfo);  }
LogStreamObservable extends LoggableObservable { @Override <T> Observable<T> get(T type) throws Throwable { final StopWatch stopWatch = new StopWatch(); final Counter emittedItems = new Counter(joinPoint.getMethodName()); return ((Observable<T>) joinPoint.proceed()) .doOnSubscribe(new Action0() { @Override public void call() { stopWatch.start(); } }) .doOnNext(new Action1<T>() { @Override public void call(T value) { emittedItems.increment(); messageManager.printObservableOnNextWithValue(observableInfo, value); } }) .doOnError(new Action1<Throwable>() { @Override public void call(Throwable throwable) { messageManager.printObservableOnError(observableInfo, throwable); } }) .doOnTerminate(new Action0() { @Override public void call() { stopWatch.stop(); observableInfo.setTotalExecutionTime(stopWatch.getTotalTimeMillis()); observableInfo.setTotalEmittedItems(emittedItems.tally()); messageManager.printObservableItemTimeInfo(observableInfo); } }); } LogStreamObservable(FrodoProceedingJoinPoint joinPoint,
      MessageManager messageManager, ObservableInfo observableInfo);   }