Examples.java

//Example 1...
//I want to get data from the server-side
mRepository.getDogsFromServerSide()
	//If there's no internet connection I want to plug the database cache
	.onErrorResumeNext(ObservableUtils.whenExceptionIs(UnknownHostException.class, mDogsCache.getAllDogs()))
	//If something goes wrong on the server-side I want to plug the memory cache
	.onErrorResumeNext(ObservableUtils.whenExceptionIs(ServerSideException.class, mMemoryDogsCache.getAllDogs()))
	//I want to turn each Dog into a Cat
	.map(new Func1<DataSource<List<Dog>>, DataSource<List<Cat>>>() {
		@Override
		public DataSource<List<Cat>> call(DataSource<List<Dog>> origin) {
			List<Cat> result = new ArrayList<>();
			for (Dog dog : origin.getModel()) {
				result.add(dog.turnIntoCat());
			}
			return new DataSource<>(result, origin.getState());
		}
	})
	//Now we subscribe
	.subscribe(new SourceSubscriber<List<Cat>>() {
		
		@Override
		protected void onSqlError(SQLException e) {
			//If something goes wrong with the database query I want to show a message
			callback.showError()
		}
		
		@Override
		protected void onResultNext(List<Cat> cats) {
			//IDK from where and I don't care, we just got a result
			callback.showCats(cats);
		}
		
		 @Override
		protected void onResultError(Throwable e) {
			//NEVER hook to this method if you're plugin another stream in the #onErrorResumeNext. If you do you are
			//going to display some error message here, then plug another stream, then some #onNext hook will be triggered.
			//You are going to display data and 
		}
		
	});
//...

//Example 2...
//I want to get data from the server-side
mRepository.getDogsFromServerSide()
	//We don't want to hook any other stream
	.subscribe(new SourceSubscriber<List<Dog>>() {
		
		@Override
		protected void onSqlError(SQLException e) {
			//This hook is NEVER going to be called because we don't have any stream from the Cache here
		}
		
		@Override
		protected void onDatabaseNext(List<Team> teamList) {
			//Same as above. This hooks are useless in this case.
		}
		
		@Override
		protected void onResultNext(List<Dog> cats) {
			//IDK from where and I don't care, we just got a result
			callback.showCats(cats);
		}
		
		 @Override
		protected void onResultError(Throwable e) {
			//Something went wrong, no matther what, we have to display the error UID
			callback.showError();
		}
		
	});
//...

//Example 3...
//I want to get data from the server-side
mRepository.getDogsFromServerSide()
	//We don't want to hook any other stream
	.subscribe(new SourceSubscriber<List<Dog>>() {
		
		@Override
		protected void onSuccessNext(List<Dog> dogs) {
			//If its 200 we want to update the Dog database
			mDogCache.saveAllDogs(dogs);
		}

		@Override
		protected void onResultNext(List<Dog> dogs) {
			//No matter if it was 200 or 304, we want to show this in the UI
			callback.showDogs(dogs);
		}
		
		@Override
		protected void onResultError(Throwable e) {
			//Something went wrong, no matther what, we have to display the error UID
			callback.showError();
		}
		
	});
//...

//Example 4...
//We want to get a dog collection from this three sources, no matter which goes first
Observable.merge(mRepository.getDogsFromServerSide(), mDbCache.getAllDogs(), mMemoryCache.getAllDogs())
	//We don't want to hook any other stream
	.subscribe(new SourceSubscriber<List<Dog>>() {
		
		@Override
		protected void onSuccessNext(List<Dog> dogs) {
			//This method is going to be called only once from mRepository.getDogsFromServerSide() if the response is 200
			callback.showLiveDogs(dogs);
		}
		
		@Override
		protected void onNotModifiedNext(List<Dog> dogs) {
			//This method is going to be called only once from mRepository.getDogsFromServerSide() if the response is 304
			callback.showLiveDogs(dogs);
		}
		
		@Override
		protected void onDatabaseNext(List<Dog> dogs) {
			//This method is going to be called only once from mDbCache.getAllDogs()
			callback.showOutdatedDatabaseDogs(dogs);
		}
		
		//NOTE if we would handle MemoryCaches as another component I'd have another hook.

		@Override
		protected void onResultNext(List<Dog> dogs) {
			//This method is going to be called three times one for each source
			callback.showDogs(dogs);
		}
		
		@Override
		protected void onSqlError(SQLException e) {
			//If something goes wrong with mDbCache.getAllDogs() this method is going to be triggered
		}

		@Override
		protected void onServerSideError(Throwable e) {
			//If something goes wrong with the mRepository.getDogsFromServerSide() this method is going to be triggered
		}

		@Override
		protected void onResultError(Throwable e) {
			//This method is going to be triggered once per each of the above errors. Works the same way that #onResultNext does.
		}
	
		
	});
//...

//Example 5...
//We want to get a dog collection from this three sources, no matter which goes first
Observable.zip(mRepository.getDogsFromServerSide(), mDbCache.getAllWolfs(),  
	new Func2<DataSource<List<Dog>>, DataSource<List<Wolf>>, DataSource<List<Animal>>>() {
            @Override
            public DataSource<List<Animal>> call(DataSource<List<Dog>> serverDogs, DataSource<List<Wolf>> cachedWolfs) {
				List<Animal> result = new ArrayList<>();
				
				result.addAll(serverDogs.getModel());
				result.addAll(cachedWolfs.getModel());
				
                return new DataSource<>(result, DataSource.NONE); //Only #onResultNext will be triggered.
				
				//NOTE We can set whatever source we want also.
            }
        })
	//We don't want to hook any other stream
	.subscribe(new SourceSubscriber<List<Animal>>() {
		
		@Override
		protected void onResultNext(List<Animal> animals) {
			//We are zipping them toghether, when zipping only one stream is return and only one stream is executed an emits items.
			callback.showAnimals(animals);
		}
		
		@Override
		protected void onResultError(Throwable e) {
			//Same as above, #onError and #onNExt will be executed only once because we are zipping items toghether
			callback.showError();
		}
		
	});
//...