Add operators to create Observables from BroadcastReceiver

it allows to listen global and local (with support LocalBroadcastManager) broadcasts

Author: Yarikx

rxjava-contrib/rxjava-android/src/main/java/rx/android/observables/AndroidObservable.java

@@ -15,16 +15,22 @@
  */
 package rx.android.observables;
 
-import static rx.android.schedulers.AndroidSchedulers.mainThread;
+import android.app.Activity;
+import android.app.Fragment;
+import android.content.Context;
+import android.content.Intent;
+import android.content.IntentFilter;
+import android.os.Build;
+import android.os.Handler;
 
 import rx.Observable;
 import rx.functions.Func1;
-import rx.operators.OperatorObserveFromAndroidComponent;
+import rx.operators.OperatorBroadcastRegister;
 import rx.operators.OperatorConditionalBinding;
+import rx.operators.OperatorLocalBroadcastRegister;
+import rx.operators.OperatorObserveFromAndroidComponent;
 
-import android.app.Activity;
-import android.app.Fragment;
-import android.os.Build;
+import static rx.android.schedulers.AndroidSchedulers.mainThread;
 
 
 public final class AndroidObservable {
@@ -176,4 +182,37 @@ public static <T> Observable<T> bindFragment(Object fragment, Observable<T> sour
             throw new IllegalArgumentException("Target fragment is neither a native nor support library Fragment");
         }
     }
+
+    /**
+     * Create Observable that wraps BroadcastReceiver and emmit received intents.
+     *
+     * @param filter Selects the Intent broadcasts to be received.
+     */
+    public static Observable<Intent> fromBroadcast(Context context, IntentFilter filter){
+        return Observable.create(new OperatorBroadcastRegister(context, filter, null, null));
+    }
+
+    /**
+     * Create Observable that wraps BroadcastReceiver and emmit received intents.
+     *
+     * @param filter Selects the Intent broadcasts to be received.
+     * @param broadcastPermission String naming a permissions that a
+     *      broadcaster must hold in order to send an Intent to you.  If null,
+     *      no permission is required.
+     * @param schedulerHandler Handler identifying the thread that will receive
+     *      the Intent.  If null, the main thread of the process will be used.
+     */
+    public static Observable<Intent> fromBroadcast(Context context, IntentFilter filter, String broadcastPermission, Handler schedulerHandler){
+        return Observable.create(new OperatorBroadcastRegister(context, filter, broadcastPermission, schedulerHandler));
+    }
+
+    /**
+     * Create Observable that wraps BroadcastReceiver and connects to LocalBroadcastManager
+     * to emmit received intents.
+     *
+     * @param filter Selects the Intent broadcasts to be received.
+     */
+    public static Observable<Intent> fromLocalBroadcast(Context context, IntentFilter filter){
+        return Observable.create(new OperatorLocalBroadcastRegister(context, filter));
+    }
 }

rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorBroadcastRegister.java

@@ -0,0 +1,67 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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.operators;
+
+import android.content.BroadcastReceiver;
+import android.content.Context;
+import android.content.Intent;
+import android.content.IntentFilter;
+import android.os.Handler;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.Subscription;
+import rx.android.subscriptions.AndroidSubscriptions;
+import rx.functions.Action0;
+
+public class OperatorBroadcastRegister implements Observable.OnSubscribe<Intent> {
+
+    private final Context context;
+    private final IntentFilter intentFilter;
+    private final String broadcastPermission;
+    private final Handler schedulerHandler;
+
+    public OperatorBroadcastRegister(Context context, IntentFilter intentFilter, String broadcastPermission, Handler schedulerHandler) {
+        this.context = context;
+        this.intentFilter = intentFilter;
+        this.broadcastPermission = broadcastPermission;
+        this.schedulerHandler = schedulerHandler;
+    }
+
+    @Override
+    public void call(final Subscriber<? super Intent> subscriber) {
+        final BroadcastReceiver broadcastReceiver = new BroadcastReceiver() {
+            @Override
+            public void onReceive(Context context, Intent intent) {
+                subscriber.onNext(intent);
+            }
+        };
+
+        final Subscription subscription = AndroidSubscriptions.unsubscribeInUiThread(new Action0() {
+            @Override
+            public void call() {
+                context.unregisterReceiver(broadcastReceiver);
+            }
+        });
+
+        subscriber.add(subscription);
+        Intent stickyIntent = context.registerReceiver(broadcastReceiver, intentFilter, broadcastPermission, schedulerHandler);
+        if (stickyIntent != null) {
+            subscriber.onNext(stickyIntent);
+        }
+
+    }
+}

rxjava-contrib/rxjava-android/src/main/java/rx/operators/OperatorLocalBroadcastRegister.java

@@ -0,0 +1,60 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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.operators;
+
+import android.content.BroadcastReceiver;
+import android.content.Context;
+import android.content.Intent;
+import android.content.IntentFilter;
+import android.support.v4.content.LocalBroadcastManager;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.Subscription;
+import rx.android.subscriptions.AndroidSubscriptions;
+import rx.functions.Action0;
+
+public class OperatorLocalBroadcastRegister implements Observable.OnSubscribe<Intent> {
+
+    private final Context context;
+    private final IntentFilter intentFilter;
+
+    public OperatorLocalBroadcastRegister(Context context, IntentFilter intentFilter) {
+        this.context = context;
+        this.intentFilter = intentFilter;
+    }
+
+    @Override
+    public void call(final Subscriber<? super Intent> subscriber) {
+        final LocalBroadcastManager localBroadcastManager = LocalBroadcastManager.getInstance(context);
+        final BroadcastReceiver broadcastReceiver = new BroadcastReceiver() {
+            @Override
+            public void onReceive(Context context, Intent intent) {
+                subscriber.onNext(intent);
+            }
+        };
+
+        final Subscription subscription = AndroidSubscriptions.unsubscribeInUiThread(new Action0() {
+            @Override
+            public void call() {
+                localBroadcastManager.unregisterReceiver(broadcastReceiver);
+            }
+        });
+
+        subscriber.add(subscription);
+        localBroadcastManager.registerReceiver(broadcastReceiver, intentFilter);
+    }
+}

rxjava-contrib/rxjava-android/src/test/java/rx/android/operators/OperatorLocalBroadcastRegisterTest.java

@@ -0,0 +1,73 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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.operators;
+
+import android.app.Application;
+import android.content.Intent;
+import android.content.IntentFilter;
+import android.support.v4.content.LocalBroadcastManager;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InOrder;
+import org.robolectric.Robolectric;
+import org.robolectric.RobolectricTestRunner;
+
+import rx.Observable;
+import rx.Observer;
+import rx.Subscription;
+import rx.android.observables.AndroidObservable;
+import rx.observers.TestObserver;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+
+@RunWith(RobolectricTestRunner.class)
+public class OperatorLocalBroadcastRegisterTest {
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testLocalBroadcast() {
+        String action = "TEST_ACTION";
+        IntentFilter intentFilter = new IntentFilter(action);
+        Application application = Robolectric.application;
+        Observable<Intent> observable = AndroidObservable.fromLocalBroadcast(application, intentFilter);
+        final Observer<Intent> observer = mock(Observer.class);
+        final Subscription subscription = observable.subscribe(new TestObserver<Intent>(observer));
+
+        final InOrder inOrder = inOrder(observer);
+
+        inOrder.verify(observer, never()).onNext(any(Intent.class));
+
+        Intent intent = new Intent(action);
+        LocalBroadcastManager localBroadcastManager = LocalBroadcastManager.getInstance(application);
+        localBroadcastManager.sendBroadcast(intent);
+        inOrder.verify(observer, times(1)).onNext(intent);
+
+        localBroadcastManager.sendBroadcast(intent);
+        inOrder.verify(observer, times(1)).onNext(intent);
+
+        subscription.unsubscribe();
+        inOrder.verify(observer, never()).onNext(any(Intent.class));
+
+        inOrder.verify(observer, never()).onError(any(Throwable.class));
+        inOrder.verify(observer, never()).onCompleted();
+    }
+
+}