Add isPending() method in RxManager to find if there is any call pending

alfonz-rx/README.md

@@ -64,6 +64,8 @@ public void onDestroy()
 }
 ```
 
+If you would call `isRunning()` right after `runHelloCall()` in the previous example, it might return false. It's because the call is considered as running after you actually subscribe which could happen a few milliseconds after calling the `isRunning()`. Keep in mind that you are usually subscribing on another thread. Use `isPending()` to find if there is any call pending. Use `!mRxManager.isPending() && !mRxManager.isRunning()` to make sure that a specific call has not been executed yet.
+
 
 How to use RxBus
 ----------------

alfonz-rx/src/main/java/org/alfonz/rx/RxManager.java

@@ -21,14 +21,17 @@ public class RxManager
 	private static final String TAG = "ALFONZ";
 
 	private CompositeDisposable mCompositeDisposable = new CompositeDisposable();
+	private Map<String, Short> mPendingCalls = new ArrayMap<>();
 	private Map<String, Short> mRunningCalls = new ArrayMap<>();
 
 
 	public <T> Observable<T> setupObservable(@NonNull Observable<T> observable, @NonNull String callType)
 	{
+		addPendingCall(callType);
 		return observable
 				.doOnSubscribe(disposable ->
 				{
+					removePendingCall(callType);
 					addRunningCall(callType);
 					registerDisposable(disposable);
 				})
@@ -44,9 +47,11 @@ public <T> Observable<T> setupObservableWithSchedulers(@NonNull Observable<T> ob
 
 	public <T> Single<T> setupSingle(@NonNull Single<T> single, @NonNull String callType)
 	{
+		addPendingCall(callType);
 		return single
 				.doOnSubscribe(disposable ->
 				{
+					removePendingCall(callType);
 					addRunningCall(callType);
 					registerDisposable(disposable);
 				})
@@ -62,9 +67,11 @@ public <T> Single<T> setupSingleWithSchedulers(@NonNull Single<T> single, @NonNu
 
 	public Completable setupCompletable(@NonNull Completable completable, @NonNull String callType)
 	{
+		addPendingCall(callType);
 		return completable
 				.doOnSubscribe(disposable ->
 				{
+					removePendingCall(callType);
 					addRunningCall(callType);
 					registerDisposable(disposable);
 				})
@@ -80,9 +87,11 @@ public Completable setupCompletableWithSchedulers(@NonNull Completable completab
 
 	public <T> Maybe<T> setupMaybe(@NonNull Maybe<T> maybe, @NonNull String callType)
 	{
+		addPendingCall(callType);
 		return maybe
 				.doOnSubscribe(disposable ->
 				{
+					removePendingCall(callType);
 					addRunningCall(callType);
 					registerDisposable(disposable);
 				})
@@ -99,29 +108,49 @@ public <T> Maybe<T> setupMaybeWithSchedulers(@NonNull Maybe<T> maybe, @NonNull S
 	public void disposeAll()
 	{
 		mCompositeDisposable.clear();
+		mPendingCalls.clear();
 		mRunningCalls.clear();
 	}
 
 
+	public boolean isPending(@NonNull String callType)
+	{
+		return mPendingCalls.containsKey(callType);
+	}
+
+
 	public boolean isRunning(@NonNull String callType)
 	{
 		return mRunningCalls.containsKey(callType);
 	}
 
 
-	public void printAll()
+	public synchronized void printAll()
 	{
 		String codeLocation = "[" + RxManager.class.getSimpleName() + ".printAll] ";
 
-		if(mRunningCalls.isEmpty())
+		if(mPendingCalls.isEmpty())
+		{
+			Log.d(TAG, codeLocation + "no pending calls");
+		}
+		else
 		{
-			Log.d(TAG, codeLocation + "empty");
-			return;
+			for(Map.Entry<String, Short> entry : mPendingCalls.entrySet())
+			{
+				Log.d(TAG, codeLocation + entry.getKey() + ": " + entry.getValue() + " pending");
+			}
 		}
 
-		for(Map.Entry<String, Short> entry : mRunningCalls.entrySet())
+		if(mRunningCalls.isEmpty())
+		{
+			Log.d(TAG, codeLocation + "no running calls");
+		}
+		else
 		{
-			Log.d(TAG, codeLocation + entry.getKey() + ": " + entry.getValue());
+			for(Map.Entry<String, Short> entry : mRunningCalls.entrySet())
+			{
+				Log.d(TAG, codeLocation + entry.getKey() + ": " + entry.getValue() + " running");
+			}
 		}
 	}
 
@@ -132,6 +161,33 @@ private void registerDisposable(@NonNull Disposable disposable)
 	}
 
 
+	private synchronized void addPendingCall(@NonNull String callType)
+	{
+		short count = 0;
+		if(mPendingCalls.containsKey(callType))
+		{
+			count = mPendingCalls.get(callType);
+		}
+		mPendingCalls.put(callType, ++count);
+	}
+
+
+	private synchronized void removePendingCall(@NonNull String callType)
+	{
+		Short count = mPendingCalls.get(callType);
+		if(count == null) return;
+
+		if(count > 1)
+		{
+			mPendingCalls.put(callType, --count);
+		}
+		else
+		{
+			mPendingCalls.remove(callType);
+		}
+	}
+
+
 	private synchronized void addRunningCall(@NonNull String callType)
 	{
 		short count = 0;