Merge pull request #76 from dpsm/feature/cursor

Adding support to create self closing Observables from Cursors

Author: JakeWharton

rxandroid/src/main/java/rx/android/content/ContentObservable.java

@@ -6,6 +6,7 @@
 import android.content.Intent;
 import android.content.IntentFilter;
 import android.content.SharedPreferences;
+import android.database.Cursor;
 import android.os.Build;
 import android.os.Handler;
 
@@ -138,6 +139,15 @@ public static Observable<String> fromSharedPreferencesChanges(SharedPreferences
         return Observable.create(new OnSubscribeSharedPreferenceChange(sharedPreferences));
     }
 
+    /**
+     * Create Observable that emits the specified {@link android.database.Cursor} for each available position
+     * of the cursor moving to the next position before each call and closing the cursor whether the
+     * Observable completes or an error occurs.
+     */
+    public static Observable<Cursor> fromCursor(final Cursor cursor) {
+        return Observable.create(new OnSubscribeCursor(cursor));
+    }
+
     private ContentObservable() {
     }
 }

rxandroid/src/main/java/rx/android/content/OnSubscribeCursor.java

@@ -0,0 +1,46 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.android.content;
+
+import android.database.Cursor;
+
+import rx.Observable;
+import rx.Subscriber;
+
+/**
+ * Emits a {@link android.database.Cursor} for every available position.
+ */
+final class OnSubscribeCursor implements Observable.OnSubscribe<Cursor> {
+
+    private final Cursor cursor;
+
+    OnSubscribeCursor(final Cursor cursor) {
+        this.cursor = cursor;
+    }
+
+    @Override
+    public void call(final Subscriber<? super Cursor> subscriber) {
+        try {
+            while (!subscriber.isUnsubscribed() && cursor.moveToNext()) {
+                subscriber.onNext(cursor);
+            }
+        } finally {
+            if (!cursor.isClosed()) {
+                cursor.close();
+            }
+        }
+        subscriber.onCompleted();
+    }
+
+}

rxandroid/src/test/java/rx/android/content/ContentObservableTest.java

@@ -13,34 +13,43 @@
  */
 package rx.android.content;
 
-import static org.mockito.Mockito.verify;
+import android.app.Activity;
+import android.app.Fragment;
+import android.database.Cursor;
+import android.support.v4.app.FragmentActivity;
 
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.Matchers;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.robolectric.Robolectric;
 import org.robolectric.RobolectricTestRunner;
 import org.robolectric.annotation.Config;
 
-import rx.Observable;
-import rx.Observer;
-import rx.android.content.ContentObservable;
-import rx.android.TestUtil;
-import rx.observers.TestObserver;
-
-import android.app.Activity;
-import android.app.Fragment;
-import android.support.v4.app.FragmentActivity;
-
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
+import rx.Observable;
+import rx.Observer;
+import rx.Subscriber;
+import rx.android.TestUtil;
+import rx.observers.TestObserver;
+import rx.observers.TestSubscriber;
+
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
 
 @RunWith(RobolectricTestRunner.class)
 @Config(manifest = Config.NONE)
@@ -157,4 +166,70 @@ public void bindActivityToSourceFromDifferentThread() throws InterruptedExceptio
     }
 
 
+    public void givenCursorWhenFromCursorInvokedThenObservableCallsOnNextWhileHasNext() {
+        final Subscriber<Cursor> subscriber = spy(new TestSubscriber<Cursor>());
+        final Cursor cursor = mock(Cursor.class);
+
+        when(cursor.isAfterLast()).thenReturn(false, false, true);
+        when(cursor.moveToNext()).thenReturn(true, true, false);
+        when(cursor.getCount()).thenReturn(2);
+
+        Observable<Cursor> observable = ContentObservable.fromCursor(cursor);
+        observable.subscribe(subscriber);
+
+        verify(subscriber, times(2)).onNext(cursor);
+        verify(subscriber, never()).onError(Matchers.any(Throwable.class));
+        verify(subscriber).onCompleted();
+    }
+
+    @Test
+    public void givenEmptyCursorWhenFromCursorInvokedThenObservableCompletesWithoutCallingOnNext() {
+        final Subscriber<Cursor> subscriber = spy(new TestSubscriber<Cursor>());
+        final Cursor cursor = mock(Cursor.class);
+
+        Observable<Cursor> observable = ContentObservable.fromCursor(cursor);
+        observable.subscribe(subscriber);
+
+        verify(subscriber, never()).onNext(cursor);
+        verify(subscriber, never()).onError(Matchers.any(Throwable.class));
+        verify(subscriber).onCompleted();
+    }
+
+    @Test
+    public void givenCursorWhenFromCursorCalledThenEmitsAndClosesCursorAfterCompletion() {
+        final Subscriber<Cursor> subscriber = spy(new TestSubscriber<Cursor>());
+        final Cursor cursor = mock(Cursor.class);
+
+        when(cursor.isAfterLast()).thenReturn(false, true);
+        when(cursor.moveToNext()).thenReturn(true, false);
+        when(cursor.getCount()).thenReturn(1);
+
+        Observable<Cursor> observable = ContentObservable.fromCursor(cursor);
+        observable.subscribe(subscriber);
+
+        verify(subscriber, never()).onError(Matchers.any(Throwable.class));
+        verify(subscriber).onNext(cursor);
+        verify(cursor).close();
+        verify(subscriber).onCompleted();
+    }
+
+    @Test
+    public void givenCursorWhenFromCursorCalledThenEmitsAndClosesCursorAfterError() {
+        final Subscriber<Cursor> subscriber = spy(new TestSubscriber<Cursor>());
+        final Cursor cursor = mock(Cursor.class);
+        final RuntimeException throwable = mock(RuntimeException.class);
+        doThrow(throwable).when(subscriber).onNext(cursor);
+
+        when(cursor.isAfterLast()).thenReturn(false, true);
+        when(cursor.moveToNext()).thenReturn(true, false);
+        when(cursor.getCount()).thenReturn(1);
+
+        Observable<Cursor> observable = ContentObservable.fromCursor(cursor);
+        observable.subscribe(subscriber);
+
+        verify(subscriber, never()).onCompleted();
+        verify(subscriber).onNext(cursor);
+        verify(subscriber).onError(throwable);
+        verify(cursor).close();
+    }
 }