Extract actual feed loading code into separate method

Increase readability

app/src/main/java/org/schabi/newpipe/local/feed/service/FeedLoadManager.kt

@@ -1,6 +1,7 @@
 package org.schabi.newpipe.local.feed.service
 
 import android.content.Context
+import android.content.SharedPreferences
 import androidx.preference.PreferenceManager
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.Completable
@@ -13,6 +14,7 @@ import io.reactivex.rxjava3.schedulers.Schedulers
 import org.schabi.newpipe.R
 import org.schabi.newpipe.database.feed.model.FeedGroupEntity
 import org.schabi.newpipe.database.subscription.NotificationMode
+import org.schabi.newpipe.database.subscription.SubscriptionEntity
 import org.schabi.newpipe.extractor.Info
 import org.schabi.newpipe.extractor.NewPipe
 import org.schabi.newpipe.extractor.feed.FeedInfo
@@ -108,6 +110,38 @@ class FeedLoadManager(private val context: Context) {
             .runOn(Schedulers.io(), PARALLEL_EXTRACTIONS * 2)
             .filter { !cancelSignal.get() }
             .map { subscriptionEntity ->
+                loadStreams(subscriptionEntity, useFeedExtractor, defaultSharedPreferences)
+            }
+            .sequential()
+            .observeOn(AndroidSchedulers.mainThread())
+            .doOnNext(NotificationConsumer())
+            .observeOn(Schedulers.io())
+            .buffer(BUFFER_COUNT_BEFORE_INSERT)
+            .doOnNext(DatabaseConsumer())
+            .subscribeOn(Schedulers.io())
+            .toList()
+            .flatMap { x -> postProcessFeed().toSingleDefault(x.flatten()) }
+    }
+
+    fun cancel() {
+        cancelSignal.set(true)
+    }
+
+    private fun broadcastProgress() {
+        FeedEventManager.postEvent(
+            FeedEventManager.Event.ProgressEvent(
+                currentProgress.get(),
+                maxProgress.get()
+            )
+        )
+    }
+
+    private fun loadStreams(
+        subscriptionEntity: SubscriptionEntity,
+        useFeedExtractor: Boolean,
+        defaultSharedPreferences: SharedPreferences
+    ):
+        Notification<FeedUpdateInfo> {
         var error: Throwable? = null
         val storeOriginalErrorAndRethrow = { e: Throwable ->
             // keep original to prevent blockingGet() from wrapping it into RuntimeException
@@ -183,7 +217,7 @@ class FeedLoadManager(private val context: Context) {
                     .filterIsInstance<StreamInfoItem>()
             }
 
-                    return@map Notification.createOnNext(
+            return Notification.createOnNext(
                 FeedUpdateInfo(
                     subscriptionEntity,
                     originalInfo!!,
@@ -199,32 +233,9 @@ class FeedLoadManager(private val context: Context) {
                 // do this to prevent blockingGet() from wrapping into RuntimeException
                 error ?: e
             )
-                    return@map Notification.createOnError<FeedUpdateInfo>(wrapper)
+            return Notification.createOnError(wrapper)
         }
     }
-            .sequential()
-            .observeOn(AndroidSchedulers.mainThread())
-            .doOnNext(NotificationConsumer())
-            .observeOn(Schedulers.io())
-            .buffer(BUFFER_COUNT_BEFORE_INSERT)
-            .doOnNext(DatabaseConsumer())
-            .subscribeOn(Schedulers.io())
-            .toList()
-            .flatMap { x -> postProcessFeed().toSingleDefault(x.flatten()) }
-    }
-
-    fun cancel() {
-        cancelSignal.set(true)
-    }
-
-    private fun broadcastProgress() {
-        FeedEventManager.postEvent(
-            FeedEventManager.Event.ProgressEvent(
-                currentProgress.get(),
-                maxProgress.get()
-            )
-        )
-    }
 
     /**
      * Keep the feed and the stream tables small