Add an option to allow downloads on mobile connections.

app/build.gradle

@@ -113,6 +113,7 @@ dependencies {
     compile 'eu.davidea:flexible-adapter:4.2.0@aar'
     compile 'com.nononsenseapps:filepicker:2.5.0'
     compile 'com.amulyakhare:com.amulyakhare.textdrawable:1.0.1'
+    compile 'com.github.pwittchen:reactivenetwork:0.1.5'
 
     compile "com.google.dagger:dagger:$DAGGER_VERSION"
     apt "com.google.dagger:dagger-compiler:$DAGGER_VERSION"

app/src/main/java/eu/kanade/mangafeed/data/download/DownloadManager.java

@@ -393,7 +393,7 @@ public class DownloadManager {
             return false;
 
         boolean hasPendingDownloads = false;
-        if (downloadsSubscription == null || threadsNumberSubscription == null)
+        if (downloadsSubscription == null)
             initializeSubscriptions();
 
         for (Download download : queue) {

app/src/main/java/eu/kanade/mangafeed/data/download/DownloadService.java

@@ -3,28 +3,33 @@ package eu.kanade.mangafeed.data.download;
 import android.app.Service;
 import android.content.Context;
 import android.content.Intent;
-import android.content.IntentFilter;
-import android.net.ConnectivityManager;
 import android.os.IBinder;
 import android.os.PowerManager;
 
+import com.github.pwittchen.reactivenetwork.library.ReactiveNetwork;
+
 import javax.inject.Inject;
 
 import de.greenrobot.event.EventBus;
 import eu.kanade.mangafeed.App;
+import eu.kanade.mangafeed.R;
+import eu.kanade.mangafeed.data.preference.PreferencesHelper;
 import eu.kanade.mangafeed.event.DownloadChaptersEvent;
-import eu.kanade.mangafeed.util.ContentObservable;
 import eu.kanade.mangafeed.util.EventBusHook;
-import eu.kanade.mangafeed.util.NetworkUtil;
+import eu.kanade.mangafeed.util.ToastUtil;
 import rx.Subscription;
+import rx.android.schedulers.AndroidSchedulers;
+import rx.schedulers.Schedulers;
 
 public class DownloadService extends Service {
 
     @Inject DownloadManager downloadManager;
+    @Inject PreferencesHelper preferences;
 
     private PowerManager.WakeLock wakeLock;
     private Subscription networkChangeSubscription;
     private Subscription queueRunningSubscription;
+    private boolean isRunning;
 
     public static void start(Context context) {
         context.startService(new Intent(context, DownloadService.class));
@@ -73,22 +78,43 @@ public class DownloadService extends Service {
     }
 
     private void listenNetworkChanges() {
-        IntentFilter intentFilter = new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION);
+        networkChangeSubscription = new ReactiveNetwork().enableInternetCheck()
-        networkChangeSubscription = ContentObservable.fromBroadcast(this, intentFilter)
+                .observeConnectivity(getApplicationContext())
+                .subscribeOn(Schedulers.io())
+                .observeOn(AndroidSchedulers.mainThread())
                 .subscribe(state -> {
-                    if (NetworkUtil.isNetworkConnected(this)) {
+                    switch (state) {
+                        case WIFI_CONNECTED_HAS_INTERNET:
                             // If there are no remaining downloads, destroy the service
-                        if (!downloadManager.startDownloads())
+                            if (!isRunning && !downloadManager.startDownloads()) {
                                 stopSelf();
-                    } else {
+                            }
+                            break;
+                        case MOBILE_CONNECTED:
+                            if (!preferences.downloadOnlyOverWifi()) {
+                                if (!isRunning && !downloadManager.startDownloads()) {
+                                    stopSelf();
+                                }
+                            } else if (isRunning) {
                                 downloadManager.stopDownloads();
                             }
+                            break;
+                        default:
+                            if (isRunning) {
+                                downloadManager.stopDownloads();
+                            }
+                            break;
+                    }
+                }, error -> {
+                    ToastUtil.showShort(this, R.string.download_queue_error);
+                    stopSelf();
                 });
     }
 
     private void listenQueueRunningChanges() {
         queueRunningSubscription = downloadManager.getRunningSubject()
                 .subscribe(running -> {
+                    isRunning = running;
                     if (running)
                         acquireWakeLock();
                     else

app/src/main/java/eu/kanade/mangafeed/data/preference/PreferencesHelper.java

@@ -160,6 +160,10 @@ public class PreferencesHelper {
         return rxPrefs.getInteger(getKey(R.string.pref_download_slots_key), 1);
     }
 
+    public boolean downloadOnlyOverWifi() {
+        return prefs.getBoolean(getKey(R.string.pref_download_only_over_wifi_key), true);
+    }
+
     public static int getLibraryUpdateInterval(Context context) {
         return PreferenceManager.getDefaultSharedPreferences(context).getInt(
                 context.getString(R.string.pref_library_update_interval_key), 0);

app/src/main/java/eu/kanade/mangafeed/util/ContentObservable.java

@@ -1,93 +0,0 @@
-/**
- * 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 eu.kanade.mangafeed.util;
-
-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.functions.Action0;
-import rx.subscriptions.Subscriptions;
-
-public final class ContentObservable {
-    private ContentObservable() {
-        throw new AssertionError("No instances");
-    }
-
-    /**
-     * Create Observable that wraps BroadcastReceiver and emits received intents.
-     *
-     * @param filter Selects the Intent broadcasts to be received.
-     */
-    public static Observable<Intent> fromBroadcast(Context context, IntentFilter filter){
-        return Observable.create(new OnSubscribeBroadcastRegister(context, filter, null, null));
-    }
-
-    /**
-     * Create Observable that wraps BroadcastReceiver and emits 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 OnSubscribeBroadcastRegister(context, filter, broadcastPermission, schedulerHandler));
-    }
-
-
-    static class OnSubscribeBroadcastRegister implements Observable.OnSubscribe<Intent> {
-
-        private final Context context;
-        private final IntentFilter intentFilter;
-        private final String broadcastPermission;
-        private final Handler schedulerHandler;
-
-        public OnSubscribeBroadcastRegister(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 = Subscriptions.create(new Action0() {
-                @Override
-                public void call() {
-                    context.unregisterReceiver(broadcastReceiver);
-                }
-            });
-
-            subscriber.add(subscription);
-            context.registerReceiver(broadcastReceiver, intentFilter, broadcastPermission, schedulerHandler);
-
-        }
-    }
-
-}

app/src/main/res/values/keys.xml

@@ -28,6 +28,7 @@
 
     <string name="pref_download_directory_key">pref_download_directory_key</string>
     <string name="pref_download_slots_key">pref_download_slots_key</string>
+    <string name="pref_download_only_over_wifi_key">pref_download_only_over_wifi_key</string>
 
     <string name="pref_clear_chapter_cache_key">pref_clear_chapter_cache_key</string>
     <string name="pref_clear_database_key">pref_clear_database_key</string>

app/src/main/res/values/strings.xml

@@ -99,6 +99,7 @@
       <!-- Downloads section -->
     <string name="pref_download_directory">Downloads directory</string>
     <string name="pref_download_slots">Simultaneous downloads</string>
+    <string name="pref_download_only_over_wifi">Download only over Wi-Fi</string>
 
       <!-- Advanced section -->
     <string name="pref_clear_chapter_cache">Clear chapter cache</string>
@@ -180,6 +181,9 @@
     <string name="decode_image_error">The image could not be loaded.\nTry to change the image decoder</string>
     <string name="confirm_update_manga_sync">Update last chapter read in enabled services to %1$d?</string>
 
+    <!-- Downloads activity and service -->
+    <string name="download_queue_error">An error occurred while downloading chapters. Restart it from the downloads section</string>
+
     <!-- Library update service notifications -->
     <string name="notification_update_progress">Update progress: %1$d/%2$d</string>
     <string name="notification_update_completed">Update completed</string>

app/src/main/res/xml/pref_downloads.xml

@@ -5,6 +5,11 @@
         android:title="@string/pref_download_directory"
         android:key="@string/pref_download_directory_key"/>
 
+    <CheckBoxPreference
+        android:title="@string/pref_download_only_over_wifi"
+        android:key="@string/pref_download_only_over_wifi_key"
+        android:defaultValue="true"/>
+
     <eu.kanade.mangafeed.widget.preference.IntListPreference
         android:title="@string/pref_download_slots"
         android:key="@string/pref_download_slots_key"