Bugfixes in download manager and other minor changes

app/src/main/AndroidManifest.xml

@@ -5,6 +5,7 @@
     <uses-permission android:name="android.permission.INTERNET" />
     <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
     <uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" />
+    <uses-permission android:name="android.permission.WAKE_LOCK" />
 
     <application
         android:name=".App"

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

@@ -7,12 +7,10 @@ import com.google.gson.reflect.TypeToken;
 import com.google.gson.stream.JsonReader;
 
 import java.io.File;
-import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.FileReader;
 import java.io.IOException;
 import java.lang.reflect.Type;
-import java.util.ArrayList;
 import java.util.List;
 
 import eu.kanade.mangafeed.data.models.Chapter;
@@ -33,16 +31,18 @@ import timber.log.Timber;
 
 public class DownloadManager {
 
-    private PublishSubject<DownloadChaptersEvent> downloadsSubject;
-    private Subscription downloadSubscription;
-    private Subscription threadNumberSubscription;
-
     private Context context;
     private SourceManager sourceManager;
     private PreferencesHelper preferences;
     private Gson gson;
 
+    private PublishSubject<Download> downloadsQueueSubject;
+    private BehaviorSubject<Integer> threadsNumber;
+    private Subscription downloadsSubscription;
+    private Subscription threadNumberSubscription;
+
     private DownloadQueue queue;
+    private transient boolean isQueuePaused;
 
     public static final String PAGE_LIST_FILE = "index.json";
 
@@ -54,78 +54,63 @@ public class DownloadManager {
 
         queue = new DownloadQueue();
 
-        initializeDownloadSubscription();
+        initializeDownloadsSubscription();
     }
 
-    public PublishSubject<DownloadChaptersEvent> getDownloadsSubject() {
+    private void initializeDownloadsSubscription() {
-        return downloadsSubject;
+        if (downloadsSubscription != null && !downloadsSubscription.isUnsubscribed())
-    }
+            downloadsSubscription.unsubscribe();
-
-    private void initializeDownloadSubscription() {
-        if (downloadSubscription != null && !downloadSubscription.isUnsubscribed()) {
-            downloadSubscription.unsubscribe();
-        }
 
         if (threadNumberSubscription != null && !threadNumberSubscription.isUnsubscribed())
             threadNumberSubscription.unsubscribe();
 
-        downloadsSubject = PublishSubject.create();
+        downloadsQueueSubject = PublishSubject.create();
-        BehaviorSubject<Integer> threads = BehaviorSubject.create();
+        threadsNumber = BehaviorSubject.create();
 
-        threadNumberSubscription = preferences.getDownloadTheadsObs()
+        threadNumberSubscription = preferences.getDownloadTheadsObservable()
-                .subscribe(threads::onNext);
+                .filter(n -> !isQueuePaused)
+                .doOnNext(n -> isQueuePaused = (n == 0))
+                .subscribe(threadsNumber::onNext);
 
-        // Listen for download events, add them to queue and download
+        downloadsSubscription = downloadsQueueSubject
-        downloadSubscription = downloadsSubject
+                .observeOn(Schedulers.newThread())
-                .subscribeOn(Schedulers.io())
+                .lift(new DynamicConcurrentMergeOperator<>(this::downloadChapter, threadsNumber))
-                .flatMap(this::prepareDownloads)
-                .lift(new DynamicConcurrentMergeOperator<>(this::downloadChapter, threads))
                 .onBackpressureBuffer()
                 .subscribe(page -> {},
                         e -> Timber.e(e.fillInStackTrace(), e.getMessage()));
     }
 
-    // Create a download object for every chapter and add it to the downloads queue
+    // Create a download object for every chapter in the event and add them to the downloads queue
-    private Observable<Download> prepareDownloads(DownloadChaptersEvent event) {
+    public void onDownloadChaptersEvent(DownloadChaptersEvent event) {
         final Manga manga = event.getManga();
         final Source source = sourceManager.get(manga.source);
-        List<Download> downloads = new ArrayList<>();
 
         for (Chapter chapter : event.getChapters()) {
             Download download = new Download(source, manga, chapter);
 
             if (!isChapterDownloaded(download)) {
                 queue.add(download);
-                downloads.add(download);
+                downloadsQueueSubject.onNext(download);
             }
         }
-
-        return Observable.from(downloads);
     }
 
     // Check if a chapter is already downloaded
     private boolean isChapterDownloaded(Download download) {
         // If the chapter is already queued, don't add it again
         for (Download queuedDownload : queue.get()) {
-            if (download.chapter.id == queuedDownload.chapter.id)
+            if (download.chapter.id.equals(queuedDownload.chapter.id))
                 return true;
         }
 
         // Add the directory to the download object for future access
         download.directory = getAbsoluteChapterDirectory(download);
 
-        // If the directory doesn't exist, the chapter isn't downloaded. Create it in this case
+        // If the directory doesn't exist, the chapter isn't downloaded.
         if (!download.directory.exists()) {
-            // FIXME Sometimes it's failing to create the directory... My fault?
-            try {
-                DiskUtils.createDirectory(download.directory);
-            } catch (IOException e) {
-                Timber.e("Unable to create directory for chapter");
-            }
             return false;
         }
 
-
         // If the page list doesn't exist, the chapter isn't download (or maybe it's,
         // but we consider it's not)
         List<Page> savedPages = getSavedPageList(download);
@@ -142,6 +127,12 @@ public class DownloadManager {
 
     // Download the entire chapter
     private Observable<Page> downloadChapter(Download download) {
+        try {
+            DiskUtils.createDirectory(download.directory);
+        } catch (IOException e) {
+            Timber.e(e.getMessage());
+        }
+
         Observable<List<Page>> pageListObservable = download.pages == null ?
                 // Pull page list from network and add them to download object
                 download.source
@@ -152,7 +143,6 @@ public class DownloadManager {
                 Observable.just(download.pages);
 
         return pageListObservable
-                .subscribeOn(Schedulers.io())
                 .doOnNext(pages -> download.setStatus(Download.DOWNLOADING))
                 // Get all the URLs to the source images, fetch pages if necessary
                 .flatMap(pageList -> Observable.merge(
@@ -161,31 +151,29 @@ public class DownloadManager {
                 // Start downloading images, consider we can have downloaded images already
                 .concatMap(page -> getDownloadedImage(page, download.source, download.directory))
                 // Do after download completes
-                .doOnCompleted(() -> onChapterDownloaded(download));
+                .doOnCompleted(() -> onDownloadCompleted(download));
     }
 
     // Get downloaded image if exists, otherwise download it with the method below
     public Observable<Page> getDownloadedImage(final Page page, Source source, File chapterDir) {
-        Observable<Page> obs = Observable.just(page);
+        Observable<Page> pageObservable = Observable.just(page);
         if (page.getImageUrl() == null)
-            return obs;
+            return pageObservable;
 
         String imageFilename = getImageFilename(page);
         File imagePath = new File(chapterDir, imageFilename);
 
         if (!isImageDownloaded(imagePath)) {
             page.setStatus(Page.DOWNLOAD_IMAGE);
-            obs = downloadImage(page, source, chapterDir, imageFilename);
+            pageObservable = downloadImage(page, source, chapterDir, imageFilename);
         }
 
-        return obs.flatMap(p -> {
+        return pageObservable
-            page.setImagePath(imagePath.getAbsolutePath());
+                .doOnNext(p -> p.setImagePath(imagePath.getAbsolutePath()))
-            page.setStatus(Page.READY);
+                .doOnNext(p -> p.setStatus(Page.READY))
-            return Observable.just(page);
+                .doOnError(e -> page.setStatus(Page.ERROR))
-        }).onErrorResumeNext(e -> {
+                // Allow to download the remaining images
-            page.setStatus(Page.ERROR);
+                .onErrorResumeNext(e -> Observable.just(page));
-            return Observable.just(page);
-        });
     }
 
     // Download the image
@@ -210,31 +198,46 @@ public class DownloadManager {
     }
 
     private boolean isImageDownloaded(File imagePath) {
-        return imagePath.exists() && !imagePath.isDirectory();
+        return imagePath.exists();
     }
 
-    private void onChapterDownloaded(final Download download) {
+    // Called when a download finishes. This doesn't mean the download was successful, so we check it
-        download.setStatus(Download.DOWNLOADED);
+    private void onDownloadCompleted(final Download download) {
-        download.totalProgress = download.pages.size() * 100;
+        checkDownloadIsSuccessful(download);
-        savePageList(download.source, download.manga, download.chapter, download.pages);
+        savePageList(download);
+    }
+
+    private void checkDownloadIsSuccessful(final Download download) {
+        int expectedProgress = download.pages.size() * 100;
+        int actualProgress = 0;
+        int status = Download.DOWNLOADED;
+        // If any page has an error, the download result will be error
+        for (Page page : download.pages) {
+            actualProgress += page.getProgress();
+            if (page.getStatus() == Page.ERROR) status = Download.ERROR;
+        }
+        // If the download is successful, it's safer to use the expected progress
+        download.totalProgress = (status == Download.DOWNLOADED) ? expectedProgress : actualProgress;
+        download.setStatus(status);
     }
 
     // Return the page list from the chapter's directory if it exists, null otherwise
     public List<Page> getSavedPageList(Source source, Manga manga, Chapter chapter) {
+        List<Page> pages = null;
         File chapterDir = getAbsoluteChapterDirectory(source, manga, chapter);
         File pagesFile = new File(chapterDir, PAGE_LIST_FILE);
 
         try {
             if (pagesFile.exists()) {
                 JsonReader reader = new JsonReader(new FileReader(pagesFile.getAbsolutePath()));
-
                 Type collectionType = new TypeToken<List<Page>>() {}.getType();
-                return gson.fromJson(reader, collectionType);
+                pages = gson.fromJson(reader, collectionType);
+                reader.close();
             }
-        } catch (FileNotFoundException e) {
+        } catch (Exception e) {
             Timber.e(e.fillInStackTrace(), e.getMessage());
         }
-        return null;
+        return pages;
     }
 
     // Shortcut for the method above
@@ -287,4 +290,13 @@ public class DownloadManager {
     public DownloadQueue getQueue() {
         return queue;
     }
+
+    public void pauseDownloads() {
+        threadsNumber.onNext(0);
+    }
+
+    public void resumeDownloads() {
+        isQueuePaused = false;
+        threadsNumber.onNext(preferences.getDownloadThreads());
+    }
 }

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

@@ -68,7 +68,7 @@ public class PreferencesHelper {
         return Integer.parseInt(prefs.getString(getKey(R.string.pref_download_threads_key), "1"));
     }
 
-    public Observable<Integer> getDownloadTheadsObs() {
+    public Observable<Integer> getDownloadTheadsObservable() {
         return rxPrefs.getString(getKey(R.string.pref_download_threads_key), "1")
                 .asObservable().map(Integer::parseInt);
     }

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

@@ -3,6 +3,8 @@ package eu.kanade.mangafeed.data.services;
 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 javax.inject.Inject;
@@ -12,14 +14,18 @@ import eu.kanade.mangafeed.App;
 import eu.kanade.mangafeed.data.helpers.DownloadManager;
 import eu.kanade.mangafeed.events.DownloadChaptersEvent;
 import eu.kanade.mangafeed.util.AndroidComponentUtil;
+import eu.kanade.mangafeed.util.ContentObservable;
 import eu.kanade.mangafeed.util.EventBusHook;
+import rx.Subscription;
 
 public class DownloadService extends Service {
 
     @Inject DownloadManager downloadManager;
 
-    public static Intent getStartIntent(Context context) {
+    private Subscription networkChangeSubscription;
-        return new Intent(context, DownloadService.class);
+
+    public static void start(Context context) {
+        context.startService(new Intent(context, DownloadService.class));
     }
 
     public static boolean isRunning(Context context) {
@@ -30,6 +36,7 @@ public class DownloadService extends Service {
     public void onCreate() {
         super.onCreate();
         App.get(this).getComponent().inject(this);
+        listenNetworkChanges();
 
         EventBus.getDefault().registerSticky(this);
     }
@@ -39,6 +46,13 @@ public class DownloadService extends Service {
         return START_STICKY;
     }
 
+    @Override
+    public void onDestroy() {
+        EventBus.getDefault().unregister(this);
+        networkChangeSubscription.unsubscribe();
+        super.onDestroy();
+    }
+
     @Override
     public IBinder onBind(Intent intent) {
         return null;
@@ -46,14 +60,16 @@ public class DownloadService extends Service {
 
     @EventBusHook
     public void onEvent(DownloadChaptersEvent event) {
-        downloadManager.getDownloadsSubject().onNext(event);
         EventBus.getDefault().removeStickyEvent(event);
+        downloadManager.onDownloadChaptersEvent(event);
     }
 
-    @Override
+    private void listenNetworkChanges() {
-    public void onDestroy() {
+        IntentFilter intentFilter = new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION);
-        EventBus.getDefault().unregister(this);
+        networkChangeSubscription = ContentObservable.fromBroadcast(this, intentFilter)
-        super.onDestroy();
+                .subscribe(state -> {
+                    // TODO
+                });
     }
 
 }

app/src/main/java/eu/kanade/mangafeed/ui/fragment/LibraryFragment.java

@@ -21,7 +21,6 @@ import eu.kanade.mangafeed.R;
 import eu.kanade.mangafeed.data.models.Manga;
 import eu.kanade.mangafeed.data.services.LibraryUpdateService;
 import eu.kanade.mangafeed.presenter.LibraryPresenter;
-import eu.kanade.mangafeed.ui.activity.MainActivity;
 import eu.kanade.mangafeed.ui.activity.MangaDetailActivity;
 import eu.kanade.mangafeed.ui.adapter.LibraryAdapter;
 import eu.kanade.mangafeed.ui.fragment.base.BaseRxFragment;
@@ -31,7 +30,6 @@ import nucleus.factory.RequiresPresenter;
 public class LibraryFragment extends BaseRxFragment<LibraryPresenter> {
 
     @Bind(R.id.gridView) GridView grid;
-    private MainActivity activity;
     private LibraryAdapter adapter;
 
     public static LibraryFragment newInstance() {
@@ -45,8 +43,6 @@ public class LibraryFragment extends BaseRxFragment<LibraryPresenter> {
     public void onCreate(Bundle savedInstanceState) {
         super.onCreate(savedInstanceState);
         setHasOptionsMenu(true);
-
-        activity = (MainActivity)getActivity();
     }
 
     @Override
@@ -54,7 +50,7 @@ public class LibraryFragment extends BaseRxFragment<LibraryPresenter> {
                              Bundle savedInstanceState) {
         // Inflate the layout for this fragment
         View view = inflater.inflate(R.layout.fragment_library, container, false);
-        activity.setToolbarTitle(getString(R.string.library_title));
+        setToolbarTitle(getString(R.string.library_title));
         ButterKnife.bind(this, view);
 
         createAdapter();
@@ -73,9 +69,9 @@ public class LibraryFragment extends BaseRxFragment<LibraryPresenter> {
     public boolean onOptionsItemSelected(MenuItem item) {
         switch (item.getItemId()) {
             case R.id.action_refresh:
-                if (!LibraryUpdateService.isRunning(activity)) {
+                if (!LibraryUpdateService.isRunning(getActivity())) {
-                    Intent intent = LibraryUpdateService.getStartIntent(activity);
+                    Intent intent = LibraryUpdateService.getStartIntent(getActivity());
-                    activity.startService(intent);
+                    getActivity().startService(intent);
                 }
 
                 return true;

app/src/main/java/eu/kanade/mangafeed/ui/fragment/MangaChaptersFragment.java

@@ -131,8 +131,7 @@ public class MangaChaptersFragment extends BaseRxFragment<MangaChaptersPresenter
                 getPresenter().markChaptersRead(getSelectedChapters(), false);
                 return true;
             case R.id.action_download:
-                Intent intent = DownloadService.getStartIntent(getActivity());
+                DownloadService.start(getActivity());
-                getActivity().startService(intent);
                 getPresenter().downloadChapters(getSelectedChapters());
                 closeActionMode();
                 return true;

app/src/main/java/eu/kanade/mangafeed/ui/fragment/SourceFragment.java

@@ -26,19 +26,12 @@ public class SourceFragment extends BaseRxFragment<SourcePresenter> {
 
     @Bind(R.id.catalogue_list) ListView source_list;
 
-    private MainActivity activity;
     private EasyAdapter<Source> adapter;
 
     public static SourceFragment newInstance() {
         return new SourceFragment();
     }
 
-    @Override
-    public void onCreate(Bundle savedInstanceState) {
-        super.onCreate(savedInstanceState);
-        activity = (MainActivity)getActivity();
-    }
-
     @Override
     public View onCreateView(LayoutInflater inflater, ViewGroup container,
                              Bundle savedInstanceState) {
@@ -56,6 +49,7 @@ public class SourceFragment extends BaseRxFragment<SourcePresenter> {
     @OnItemClick(R.id.catalogue_list)
     public void onSourceClick(int position) {
         Source source = adapter.getItem(position);
+        MainActivity activity = (MainActivity) getActivity();
 
         if (getPresenter().isValidSource(source)) {
             CatalogueFragment fragment = CatalogueFragment.newInstance(source.getSourceId());
@@ -66,7 +60,7 @@ public class SourceFragment extends BaseRxFragment<SourcePresenter> {
     }
 
     private void createAdapter() {
-        adapter = new EasyAdapter<>(activity, SourceHolder.class);
+        adapter = new EasyAdapter<>(getActivity(), SourceHolder.class);
         source_list.setAdapter(adapter);
     }
 

app/src/main/java/eu/kanade/mangafeed/ui/fragment/base/BaseFragment.java

@@ -7,11 +7,15 @@ import eu.kanade.mangafeed.ui.activity.base.BaseActivity;
 public class BaseFragment extends Fragment {
 
     public void setToolbarTitle(String title) {
-        ((BaseActivity)getActivity()).setToolbarTitle(title);
+        getBaseActivity().setToolbarTitle(title);
     }
 
     public void setToolbarTitle(int resourceId) {
-        ((BaseActivity)getActivity()).setToolbarTitle(getString(resourceId));
+        getBaseActivity().setToolbarTitle(getString(resourceId));
+    }
+
+    public BaseActivity getBaseActivity() {
+        return (BaseActivity) getActivity();
     }
 
 }

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

@@ -0,0 +1,93 @@
+/**
+ * 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/java/eu/kanade/mangafeed/util/DiskUtils.java

@@ -115,9 +115,7 @@ public final class DiskUtils {
     }
 
     public static File saveBufferedSourceToDirectory(BufferedSource bufferedSource, File directory, String name) throws IOException {
-        if (!directory.exists() && !directory.mkdirs()) {
+        createDirectory(directory);
-            throw new IOException("Failed Creating Directory");
-        }
 
         File writeFile = new File(directory, name);
         if (writeFile.exists()) {