Added CombinedSearcher for podcasts

app/src/main/java/de/danoeh/antennapod/discovery/CombinedSearcher.java

@@ -0,0 +1,96 @@
+package de.danoeh.antennapod.discovery;
+
+import android.content.Context;
+import android.util.Log;
+import android.util.Pair;
+import io.reactivex.Single;
+import io.reactivex.SingleOnSubscribe;
+import io.reactivex.android.schedulers.AndroidSchedulers;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.schedulers.Schedulers;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+public class CombinedSearcher implements PodcastSearcher {
+    private static final String TAG = "CombinedSearcher";
+
+    private final List<Pair<PodcastSearcher, Float>> searchProviders = new ArrayList<>();
+
+    public CombinedSearcher(Context context) {
+        addProvider(new FyydPodcastSearcher(), 1.f);
+        addProvider(new ItunesPodcastSearcher(context), 1.f);
+        addProvider(new GpodnetPodcastSearcher(), 0.6f);
+    }
+
+    private void addProvider(PodcastSearcher provider, float priority) {
+        searchProviders.add(new Pair<>(provider, priority));
+    }
+
+    public Single<List<PodcastSearchResult>> search(String query) {
+        ArrayList<Disposable> disposables = new ArrayList<>();
+        List<List<PodcastSearchResult>> singleResults = new ArrayList<>(Collections.nCopies(searchProviders.size(), null));
+        CountDownLatch latch = new CountDownLatch(searchProviders.size());
+        for (int i = 0; i < searchProviders.size(); i++) {
+            Pair<PodcastSearcher, Float> searchProviderInfo = searchProviders.get(i);
+            PodcastSearcher searcher = searchProviderInfo.first;
+            final int index = i;
+            disposables.add(searcher.search(query).subscribe(e -> {
+                        singleResults.set(index, e);
+                        latch.countDown();
+                    }, throwable -> {
+                        Log.d(TAG, Log.getStackTraceString(throwable));
+                        latch.countDown();
+                    }
+            ));
+        }
+
+        return Single.create((SingleOnSubscribe<List<PodcastSearchResult>>) subscriber -> {
+            latch.await();
+            List<PodcastSearchResult> results = weightSearchResults(singleResults);
+            subscriber.onSuccess(results);
+        })
+                .doOnDispose(() -> {
+                    for (Disposable disposable : disposables) {
+                        disposable.dispose();
+                    }
+                })
+                .subscribeOn(Schedulers.io())
+                .observeOn(AndroidSchedulers.mainThread());
+    }
+
+    private List<PodcastSearchResult> weightSearchResults(List<List<PodcastSearchResult>> singleResults) {
+        HashMap<String, Float> resultRanking = new HashMap<>();
+        HashMap<String, PodcastSearchResult> urlToResult = new HashMap<>();
+        for (int i = 0; i < singleResults.size(); i++) {
+            float providerPriority = searchProviders.get(i).second;
+            List<PodcastSearchResult> providerResults = singleResults.get(i);
+            if (providerResults == null) {
+                continue;
+            }
+            for (int position = 0; position < providerResults.size(); position++) {
+                PodcastSearchResult result = providerResults.get(position);
+                urlToResult.put(result.feedUrl, result);
+
+                float ranking = 0;
+                if (resultRanking.containsKey(result.feedUrl)) {
+                    ranking = resultRanking.get(result.feedUrl);
+                }
+                ranking += 1.f / (position + 1.f);
+                resultRanking.put(result.feedUrl, ranking * providerPriority);
+            }
+        }
+        List<Map.Entry<String, Float>> sortedResults = new ArrayList<>(resultRanking.entrySet());
+        Collections.sort(sortedResults, (o1, o2) -> Double.compare(o2.getValue(), o1.getValue()));
+
+        List<PodcastSearchResult> results = new ArrayList<>();
+        for (Map.Entry<String, Float> res : sortedResults) {
+            results.add(urlToResult.get(res.getKey()));
+        }
+        return results;
+    }
+}

app/src/main/java/de/danoeh/antennapod/discovery/FyydPodcastSearcher.java

@@ -4,42 +4,35 @@ import de.danoeh.antennapod.core.service.download.AntennapodHttpClient;
 import de.mfietz.fyydlin.FyydClient;
 import de.mfietz.fyydlin.FyydResponse;
 import de.mfietz.fyydlin.SearchHit;
+import io.reactivex.Single;
+import io.reactivex.SingleOnSubscribe;
 import io.reactivex.android.schedulers.AndroidSchedulers;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.functions.Consumer;
 import io.reactivex.schedulers.Schedulers;
 
 import java.util.ArrayList;
 import java.util.List;
 
 public class FyydPodcastSearcher implements PodcastSearcher {
-    private final String query;
     private final FyydClient client = new FyydClient(AntennapodHttpClient.getHttpClient());
 
-    public FyydPodcastSearcher(String query) {
+    public Single<List<PodcastSearchResult>> search(String query) {
-        this.query = query;
+        return Single.create((SingleOnSubscribe<List<PodcastSearchResult>>) subscriber -> {
-    }
+            FyydResponse response = client.searchPodcasts(query, 10)
+                    .subscribeOn(Schedulers.io())
+                    .blockingGet();
 
-    public Disposable search(Consumer<? super List<PodcastSearchResult>> successHandler, Consumer<? super Throwable> errorHandler) {
+            ArrayList<PodcastSearchResult> searchResults = new ArrayList<>();
-        return client.searchPodcasts(query, 10)
-                .subscribeOn(Schedulers.io())
-                .observeOn(AndroidSchedulers.mainThread())
-                .subscribe(result -> {
-                    ArrayList<PodcastSearchResult> results = processSearchResult(result);
-                    successHandler.accept(results);
-                }, errorHandler);
-    }
 
-    private ArrayList<PodcastSearchResult> processSearchResult(FyydResponse response) {
+            if (!response.getData().isEmpty()) {
-        ArrayList<PodcastSearchResult> searchResults = new ArrayList<>();
+                for (SearchHit searchHit : response.getData()) {
-
+                    PodcastSearchResult podcast = PodcastSearchResult.fromFyyd(searchHit);
-        if (!response.getData().isEmpty()) {
+                    searchResults.add(podcast);
-            for (SearchHit searchHit : response.getData()) {
+                }
-                PodcastSearchResult podcast = PodcastSearchResult.fromFyyd(searchHit);
-                searchResults.add(podcast);
             }
-        }
 
-        return searchResults;
+            subscriber.onSuccess(searchResults);
+        })
+                .subscribeOn(Schedulers.io())
+                .observeOn(AndroidSchedulers.mainThread());
     }
 }

app/src/main/java/de/danoeh/antennapod/discovery/GpodnetPodcastSearcher.java

@@ -6,21 +6,13 @@ import de.danoeh.antennapod.core.gpoddernet.model.GpodnetPodcast;
 import io.reactivex.Single;
 import io.reactivex.SingleOnSubscribe;
 import io.reactivex.android.schedulers.AndroidSchedulers;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.functions.Consumer;
 import io.reactivex.schedulers.Schedulers;
 
 import java.util.ArrayList;
 import java.util.List;
 
 public class GpodnetPodcastSearcher implements PodcastSearcher {
-    private final String query;
+    public Single<List<PodcastSearchResult>> search(String query) {
-
-    public GpodnetPodcastSearcher(String query) {
-        this.query = query;
-    }
-
-    public Disposable search(Consumer<? super List<PodcastSearchResult>> successHandler, Consumer<? super Throwable> errorHandler) {
         return Single.create((SingleOnSubscribe<List<PodcastSearchResult>>) subscriber -> {
             GpodnetService service = null;
             try {
@@ -41,7 +33,6 @@ public class GpodnetPodcastSearcher implements PodcastSearcher {
             }
         })
         .subscribeOn(Schedulers.io())
-        .observeOn(AndroidSchedulers.mainThread())
+        .observeOn(AndroidSchedulers.mainThread());
-        .subscribe(successHandler, errorHandler);
     }
 }

app/src/main/java/de/danoeh/antennapod/discovery/ItunesPodcastSearcher.java

@@ -7,8 +7,6 @@ import de.danoeh.antennapod.core.service.download.AntennapodHttpClient;
 import io.reactivex.Single;
 import io.reactivex.SingleOnSubscribe;
 import io.reactivex.android.schedulers.AndroidSchedulers;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.functions.Consumer;
 import io.reactivex.schedulers.Schedulers;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
@@ -26,14 +24,12 @@ import java.util.List;
 public class ItunesPodcastSearcher implements PodcastSearcher {
     private static final String ITUNES_API_URL = "https://itunes.apple.com/search?media=podcast&term=%s";
     private final Context context;
-    private final String query;
 
-    public ItunesPodcastSearcher(Context context, String query) {
+    public ItunesPodcastSearcher(Context context) {
         this.context = context;
-        this.query = query;
     }
 
-    public Disposable search(Consumer<? super List<PodcastSearchResult>> successHandler, Consumer<? super Throwable> errorHandler) {
+    public Single<List<PodcastSearchResult>> search(String query) {
         return Single.create((SingleOnSubscribe<List<PodcastSearchResult>>) subscriber -> {
             String encodedQuery = null;
             try {
@@ -75,7 +71,6 @@ public class ItunesPodcastSearcher implements PodcastSearcher {
             subscriber.onSuccess(podcasts);
         })
                 .subscribeOn(Schedulers.io())
-                .observeOn(AndroidSchedulers.mainThread())
+                .observeOn(AndroidSchedulers.mainThread());
-                .subscribe(successHandler, errorHandler);
     }
 }

app/src/main/java/de/danoeh/antennapod/discovery/PodcastSearcher.java

@@ -1,9 +1,10 @@
 package de.danoeh.antennapod.discovery;
 
+import io.reactivex.Single;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.Consumer;
 import java.util.List;
 
 public interface PodcastSearcher {
-    Disposable search(Consumer<? super List<PodcastSearchResult>> successHandler, Consumer<? super Throwable> errorHandler);
+    Single<List<PodcastSearchResult>> search(String query);
 }

app/src/main/java/de/danoeh/antennapod/fragment/CombinedSearchFragment.java

@@ -16,36 +16,16 @@ import android.widget.Button;
 import android.widget.GridView;
 import android.widget.ProgressBar;
 import android.widget.TextView;
-import android.widget.Toast;
 import de.danoeh.antennapod.R;
 import de.danoeh.antennapod.activity.OnlineFeedViewActivity;
 import de.danoeh.antennapod.adapter.itunes.ItunesAdapter;
-import de.danoeh.antennapod.core.gpoddernet.GpodnetService;
+import de.danoeh.antennapod.discovery.CombinedSearcher;
-import de.danoeh.antennapod.core.gpoddernet.GpodnetServiceException;
-import de.danoeh.antennapod.core.gpoddernet.model.GpodnetPodcast;
-import de.danoeh.antennapod.discovery.FyydPodcastSearcher;
-import de.danoeh.antennapod.discovery.GpodnetPodcastSearcher;
-import de.danoeh.antennapod.discovery.ItunesPodcastSearcher;
 import de.danoeh.antennapod.discovery.PodcastSearchResult;
-import de.danoeh.antennapod.discovery.PodcastSearcher;
 import de.danoeh.antennapod.menuhandler.MenuItemUtils;
-import io.reactivex.Single;
-import io.reactivex.SingleOnSubscribe;
-import io.reactivex.android.schedulers.AndroidSchedulers;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.schedulers.Schedulers;
 
-import java.io.Serializable;
-import java.lang.reflect.Array;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
 
 public class CombinedSearchFragment extends Fragment {
 
@@ -65,7 +45,7 @@ public class CombinedSearchFragment extends Fragment {
      * List of podcasts retreived from the search
      */
     private List<PodcastSearchResult> searchResults = new ArrayList<>();
-    private List<Disposable> disposables = new ArrayList<>();
+    private Disposable disposable;
 
     /**
      * Constructor
@@ -108,7 +88,9 @@ public class CombinedSearchFragment extends Fragment {
     @Override
     public void onDestroy() {
         super.onDestroy();
-        disposeAll();
+        if (disposable != null) {
+            disposable.dispose();
+        }
         adapter = null;
     }
 
@@ -149,58 +131,14 @@ public class CombinedSearchFragment extends Fragment {
     }
 
     private void search(String query) {
-        disposeAll();
+        if (disposable != null) {
+            disposable.dispose();
+        }
 
         showOnlyProgressBar();
 
-        List<PodcastSearcher> searchProviders = new ArrayList<>();
+        CombinedSearcher searcher = new CombinedSearcher(getContext());
-        searchProviders.add(new FyydPodcastSearcher(query));
+        disposable = searcher.search(query).subscribe(result -> {
-        searchProviders.add(new ItunesPodcastSearcher(getContext(), query));
-        searchProviders.add(new GpodnetPodcastSearcher(query));
-
-        List<List<PodcastSearchResult>> singleResults = new ArrayList<>();
-        CountDownLatch latch = new CountDownLatch(searchProviders.size());
-        for (PodcastSearcher searchProvider : searchProviders) {
-            disposables.add(searchProvider.search(e -> {
-                    singleResults.add(e);
-                    latch.countDown();
-                }, throwable -> {
-                    Toast.makeText(getContext(), throwable.getLocalizedMessage(), Toast.LENGTH_LONG).show();
-                    latch.countDown();
-                }
-            ));
-        }
-
-        disposables.add(Single.create((SingleOnSubscribe<List<PodcastSearchResult>>) subscriber -> {
-            latch.await();
-
-            HashMap<String, Float> resultRanking = new HashMap<>();
-            HashMap<String, PodcastSearchResult> urlToResult = new HashMap<>();
-            for (List<PodcastSearchResult> providerResults : singleResults) {
-                for (int position = 0; position < providerResults.size(); position++) {
-                    PodcastSearchResult result = providerResults.get(position);
-                    urlToResult.put(result.feedUrl, result);
-
-                    float ranking = 0;
-                    if (resultRanking.containsKey(result.feedUrl)) {
-                        ranking = resultRanking.get(result.feedUrl);
-                    }
-                    ranking += 1.f / (position + 1.f);
-                    resultRanking.put(result.feedUrl, ranking);
-                }
-            }
-            List<Map.Entry<String, Float>> sortedResults = new ArrayList<>(resultRanking.entrySet());
-            Collections.sort(sortedResults, (o1, o2) -> Double.compare(o2.getValue(), o1.getValue()));
-
-            List<PodcastSearchResult> results = new ArrayList<>();
-            for (Map.Entry<String, Float> res : sortedResults) {
-                results.add(urlToResult.get(res.getKey()));
-            }
-            subscriber.onSuccess(results);
-        })
-            .subscribeOn(Schedulers.io())
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribe(result -> {
                 searchResults = result;
                 progressBar.setVisibility(View.GONE);
 
@@ -217,14 +155,7 @@ public class CombinedSearchFragment extends Fragment {
                 txtvError.setVisibility(View.VISIBLE);
                 butRetry.setOnClickListener(v -> search(query));
                 butRetry.setVisibility(View.VISIBLE);
-            }));
+            });
-    }
-
-    private void disposeAll() {
-        for (Disposable d : disposables) {
-            d.dispose();
-        }
-        disposables.clear();
     }
 
     private void showOnlyProgressBar() {
@@ -234,9 +165,4 @@ public class CombinedSearchFragment extends Fragment {
         txtvEmpty.setVisibility(View.GONE);
         progressBar.setVisibility(View.VISIBLE);
     }
-
-    public static <K, V extends Comparable<? super V>> Comparator<Map.Entry<K, V>> comparingByValue() {
-        return (Comparator<Map.Entry<K, V>> & Serializable)
-                (c1, c2) -> c1.getValue().compareTo(c2.getValue());
-    }
 }

app/src/main/java/de/danoeh/antennapod/fragment/FyydSearchFragment.java

@@ -136,8 +136,8 @@ public class FyydSearchFragment extends Fragment {
         }
         showOnlyProgressBar();
 
-        FyydPodcastSearcher searcher = new FyydPodcastSearcher(query);
+        FyydPodcastSearcher searcher = new FyydPodcastSearcher();
-        disposable = searcher.search(result -> {
+        disposable = searcher.search(query).subscribe(result -> {
             searchResults = result;
             progressBar.setVisibility(View.GONE);
 

app/src/main/java/de/danoeh/antennapod/fragment/ItunesSearchFragment.java

@@ -308,8 +308,8 @@ public class ItunesSearchFragment extends Fragment {
         txtvEmpty.setVisibility(View.GONE);
         progressBar.setVisibility(View.VISIBLE);
 
-        ItunesPodcastSearcher searcher = new ItunesPodcastSearcher(getContext(), query);
+        ItunesPodcastSearcher searcher = new ItunesPodcastSearcher(getContext());
-        disposable = searcher.search(podcasts -> {
+        disposable = searcher.search(query).subscribe(podcasts -> {
             progressBar.setVisibility(View.GONE);
             updateData(podcasts);
         }, error -> {