### Question:
ImageFetcher { public Observable<Image> toObservable() { Observable<Image> observable = Observable .create(new ObservableOnSubscribe<Image>() { @Override public void subscribe(ObservableEmitter<Image> observableEmitter) throws Exception { hasNext=true; while (hasNext && !isClosed()) { final Image image = ImageFetcher.this.fetch(); if (hasNext) { if (debug && frameIndex % Math.round(getFps()) == 0) { String msg = String.format("->%6d:%4.0fx%4.0f", frameIndex, size.width, size.height); LOG.info(msg); } observableEmitter.onNext(image); } } observableEmitter.onComplete(); } }); return observable; } ImageFetcher(String source); double getFps(); void setFps(double fps); int getFrameIndex(); boolean isStaticImage(); void setStaticImage(boolean staticImage); boolean open(); void close(); long waitForNextFrame(); Image fetch(); Image createNextImage(Mat frame, long currentMillis); Observable<Image> toObservable(); boolean hasNext(); boolean isClosed(); public boolean debug; static double DEFAULT_FPS; }### Answer:
@Test public void testImageFetcherObservable() { ImageFetcher imageFetcher = getTestImageFetcher(); Observable<Image> imageObservable = imageFetcher.toObservable(); ImageObserver imageObserver = new ImageObserver(); imageObserver.debug=debug; imageObservable.subscribe(imageObserver); assertNull(imageObserver.error); assertTrue(imageObserver.completed); assertEquals(768,imageObserver.cols); assertEquals(576,imageObserver.rows); }
@Test public void testImageFetcherObservableSample() { ImageFetcher imageFetcher = getTestImageFetcher(); Observable<Image> imageObservable = imageFetcher.toObservable(); ImageObserver imageObserver = new ImageObserver(); imageObserver.debug=debug; imageObservable.sample(40, TimeUnit.MILLISECONDS).subscribe(imageObserver); assertNull(imageObserver.error); assertTrue(imageObserver.completed); assertEquals(768,imageObserver.cols); assertEquals(576,imageObserver.rows); }