Refactor FeedLoadService to use it within the notification worker

app/build.gradle

@@ -214,8 +214,8 @@ dependencies {
     implementation 'androidx.swiperefreshlayout:swiperefreshlayout:1.1.0'
     implementation 'androidx.webkit:webkit:1.4.0'
     implementation 'com.google.android.material:material:1.2.1'
-    implementation "androidx.work:work-runtime:${workVersion}"
+    implementation "androidx.work:work-runtime-ktx:${workVersion}"
-    implementation "androidx.work:work-rxjava2:${workVersion}"
+    implementation "androidx.work:work-rxjava3:${workVersion}"
 
 /** Third-party libraries **/
     // Instance state boilerplate elimination

app/src/debug/java/org/schabi/newpipe/settings/DebugSettingsFragment.java

@@ -6,6 +6,7 @@ import androidx.preference.Preference;
 
 import org.schabi.newpipe.R;
 import org.schabi.newpipe.util.PicassoHelper;
+import org.schabi.newpipe.local.feed.notifications.NotificationWorker;
 
 import leakcanary.LeakCanary;
 
@@ -20,10 +21,13 @@ public class DebugSettingsFragment extends BasePreferenceFragment {
                 = findPreference(getString(R.string.show_image_indicators_key));
         final Preference crashTheAppPreference
                 = findPreference(getString(R.string.crash_the_app_key));
+        final Preference checkNewStreamsPreference
+                = findPreference(getString(R.string.check_new_streams_key));
 
         assert showMemoryLeaksPreference != null;
         assert showImageIndicatorsPreference != null;
         assert crashTheAppPreference != null;
+        assert checkNewStreamsPreference != null;
 
         showMemoryLeaksPreference.setOnPreferenceClickListener(preference -> {
             startActivity(LeakCanary.INSTANCE.newLeakDisplayActivityIntent());
@@ -38,5 +42,10 @@ public class DebugSettingsFragment extends BasePreferenceFragment {
         crashTheAppPreference.setOnPreferenceClickListener(preference -> {
             throw new RuntimeException();
         });
+
+        checkNewStreamsPreference.setOnPreferenceClickListener(preference -> {
+            NotificationWorker.runNow(preference.getContext());
+            return true;
+        });
     }
 }

app/src/main/java/org/schabi/newpipe/App.java

@@ -1,7 +1,5 @@
 package org.schabi.newpipe;
 
-import android.app.NotificationChannel;
-import android.app.NotificationManager;
 import android.content.Context;
 import android.content.SharedPreferences;
 import android.util.Log;
@@ -250,18 +248,15 @@ public class App extends MultiDexApplication {
                 .setDescription(getString(R.string.hash_channel_description))
                 .build();
 
-        final NotificationChannel newStreamsChannel = new NotificationChannel(
+        final NotificationChannelCompat newStreamsChannel = new NotificationChannelCompat
-                getString(R.string.streams_notification_channel_id),
+                .Builder(getString(R.string.streams_notification_channel_id),
-                getString(R.string.streams_notification_channel_name),
+                NotificationManagerCompat.IMPORTANCE_DEFAULT)
-                NotificationManager.IMPORTANCE_DEFAULT
+                .setName(getString(R.string.streams_notification_channel_name))
-        );
+                .setDescription(getString(R.string.streams_notification_channel_description))
-        newStreamsChannel.setDescription(
+                .build();
-                getString(R.string.streams_notification_channel_description)
-        );
-        newStreamsChannel.enableVibration(false);
 
         final NotificationManagerCompat notificationManager = NotificationManagerCompat.from(this);
-        notificationManager.createNotificationChannels(
+        notificationManager.createNotificationChannelsCompat(
                 Arrays.asList(mainChannel, appUpdateChannel, hashChannel, newStreamsChannel)
         );
     }

app/src/main/java/org/schabi/newpipe/MainActivity.java

@@ -69,7 +69,7 @@ import org.schabi.newpipe.fragments.BackPressable;
 import org.schabi.newpipe.fragments.MainFragment;
 import org.schabi.newpipe.fragments.detail.VideoDetailFragment;
 import org.schabi.newpipe.fragments.list.search.SearchFragment;
-import org.schabi.newpipe.notifications.NotificationWorker;
+import org.schabi.newpipe.local.feed.notifications.NotificationWorker;
 import org.schabi.newpipe.player.Player;
 import org.schabi.newpipe.player.event.OnKeyDownListener;
 import org.schabi.newpipe.player.helper.PlayerHolder;

app/src/main/java/org/schabi/newpipe/database/stream/dao/StreamDAO.kt

@@ -40,7 +40,7 @@ abstract class StreamDAO : BasicDAO<StreamEntity> {
     internal abstract fun silentInsertAllInternal(streams: List<StreamEntity>): List<Long>
 
     @Query("SELECT COUNT(*) != 0 FROM streams WHERE url = :url AND service_id = :serviceId")
-    internal abstract fun exists(serviceId: Long, url: String?): Boolean
+    internal abstract fun exists(serviceId: Int, url: String): Boolean
 
     @Query(
         """

app/src/main/java/org/schabi/newpipe/fragments/list/channel/ChannelFragment.java

@@ -44,7 +44,7 @@ import org.schabi.newpipe.extractor.stream.StreamInfoItem;
 import org.schabi.newpipe.fragments.list.BaseListInfoFragment;
 import org.schabi.newpipe.ktx.AnimationType;
 import org.schabi.newpipe.local.subscription.SubscriptionManager;
-import org.schabi.newpipe.notifications.NotificationHelper;
+import org.schabi.newpipe.local.feed.notifications.NotificationHelper;
 import org.schabi.newpipe.player.playqueue.ChannelPlayQueue;
 import org.schabi.newpipe.player.playqueue.PlayQueue;
 import org.schabi.newpipe.util.ExtractorHelper;
@@ -252,13 +252,13 @@ public class ChannelFragment extends BaseListInfoFragment<ChannelInfo>
                 .map(List::isEmpty)
                 .distinctUntilChanged()
                 .observeOn(AndroidSchedulers.mainThread())
-                .subscribe((Boolean isEmpty) -> updateSubscribeButton(!isEmpty), onError));
+                .subscribe(isEmpty -> updateSubscribeButton(!isEmpty), onError));
 
         disposables.add(observable
                 .map(List::isEmpty)
-                .filter(x -> NotificationHelper.isNewStreamsNotificationsEnabled(requireContext()))
                 .distinctUntilChanged()
-                .skip(1)
+                .skip(1) // channel has just been opened
+                .filter(x -> NotificationHelper.isNewStreamsNotificationsEnabled(requireContext()))
                 .observeOn(AndroidSchedulers.mainThread())
                 .subscribe(isEmpty -> {
                     if (!isEmpty) {

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

@@ -72,6 +72,10 @@ class FeedDatabaseManager(context: Context) {
     fun markAsOutdated(subscriptionId: Long) = feedTable
         .setLastUpdatedForSubscription(FeedLastUpdatedEntity(subscriptionId, null))
 
+    fun isStreamExist(stream: StreamInfoItem): Boolean {
+        return streamTable.exists(stream.serviceId, stream.url)
+    }
+
     fun upsertAll(
         subscriptionId: Long,
         items: List<StreamInfoItem>,

app/src/main/java/org/schabi/newpipe/local/feed/notifications/NotificationHelper.kt

@@ -0,0 +1,135 @@
+package org.schabi.newpipe.local.feed.notifications
+
+import android.app.NotificationManager
+import android.app.PendingIntent
+import android.content.Context
+import android.content.Intent
+import android.graphics.BitmapFactory
+import android.net.Uri
+import android.os.Build
+import android.provider.Settings
+import androidx.core.app.NotificationCompat
+import androidx.core.app.NotificationManagerCompat
+import androidx.core.content.ContextCompat
+import androidx.preference.PreferenceManager
+import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
+import io.reactivex.rxjava3.core.Completable
+import io.reactivex.rxjava3.core.Single
+import io.reactivex.rxjava3.schedulers.Schedulers
+import org.schabi.newpipe.R
+import org.schabi.newpipe.extractor.stream.StreamInfoItem
+import org.schabi.newpipe.local.feed.service.FeedUpdateInfo
+import org.schabi.newpipe.util.NavigationHelper
+
+class NotificationHelper(val context: Context) {
+
+    private val manager = context.getSystemService(
+        Context.NOTIFICATION_SERVICE
+    ) as NotificationManager
+
+    fun notify(data: FeedUpdateInfo): Completable {
+        val newStreams: List<StreamInfoItem> = data.newStreams
+        val summary = context.resources.getQuantityString(
+            R.plurals.new_streams, newStreams.size, newStreams.size
+        )
+        val builder = NotificationCompat.Builder(
+            context,
+            context.getString(R.string.streams_notification_channel_id)
+        )
+            .setContentTitle(
+                context.getString(
+                    R.string.notification_title_pattern,
+                    data.name,
+                    summary
+                )
+            )
+            .setContentText(
+                data.listInfo.relatedItems.joinToString(
+                    context.getString(R.string.enumeration_comma)
+                ) { x -> x.name }
+            )
+            .setNumber(newStreams.size)
+            .setBadgeIconType(NotificationCompat.BADGE_ICON_LARGE)
+            .setPriority(NotificationCompat.PRIORITY_DEFAULT)
+            .setSmallIcon(R.drawable.ic_newpipe_triangle_white)
+            .setLargeIcon(
+                BitmapFactory.decodeResource(
+                    context.resources,
+                    R.drawable.ic_newpipe_triangle_white
+                )
+            )
+            .setColor(ContextCompat.getColor(context, R.color.ic_launcher_background))
+            .setColorized(true)
+            .setAutoCancel(true)
+            .setCategory(NotificationCompat.CATEGORY_SOCIAL)
+        val style = NotificationCompat.InboxStyle()
+        for (stream in newStreams) {
+            style.addLine(stream.name)
+        }
+        style.setSummaryText(summary)
+        style.setBigContentTitle(data.name)
+        builder.setStyle(style)
+        builder.setContentIntent(
+            PendingIntent.getActivity(
+                context,
+                data.pseudoId,
+                NavigationHelper.getChannelIntent(context, data.listInfo.serviceId, data.listInfo.url)
+                    .setFlags(Intent.FLAG_ACTIVITY_NEW_TASK),
+                0
+            )
+        )
+        return Single.create(NotificationIcon(context, data.avatarUrl))
+            .subscribeOn(Schedulers.io())
+            .observeOn(AndroidSchedulers.mainThread())
+            .doOnSuccess { icon ->
+                builder.setLargeIcon(icon)
+            }
+            .ignoreElement()
+            .onErrorComplete()
+            .doOnComplete { manager.notify(data.pseudoId, builder.build()) }
+    }
+
+    companion object {
+        /**
+         * Check whether notifications are not disabled by user via system settings.
+         *
+         * @param context Context
+         * @return true if notifications are allowed, false otherwise
+         */
+        fun isNotificationsEnabledNative(context: Context): Boolean {
+            return if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
+                val channelId = context.getString(R.string.streams_notification_channel_id)
+                val manager = context.getSystemService(
+                    Context.NOTIFICATION_SERVICE
+                ) as NotificationManager
+                val channel = manager.getNotificationChannel(channelId)
+                channel != null && channel.importance != NotificationManager.IMPORTANCE_NONE
+            } else {
+                NotificationManagerCompat.from(context).areNotificationsEnabled()
+            }
+        }
+
+        @JvmStatic
+        fun isNewStreamsNotificationsEnabled(context: Context): Boolean {
+            return (
+                PreferenceManager.getDefaultSharedPreferences(context)
+                    .getBoolean(context.getString(R.string.enable_streams_notifications), false) &&
+                    isNotificationsEnabledNative(context)
+                )
+        }
+
+        fun openNativeSettingsScreen(context: Context) {
+            if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
+                val channelId = context.getString(R.string.streams_notification_channel_id)
+                val intent = Intent(Settings.ACTION_CHANNEL_NOTIFICATION_SETTINGS)
+                    .putExtra(Settings.EXTRA_APP_PACKAGE, context.packageName)
+                    .putExtra(Settings.EXTRA_CHANNEL_ID, channelId)
+                context.startActivity(intent)
+            } else {
+                val intent = Intent(Settings.ACTION_APPLICATION_DETAILS_SETTINGS)
+                intent.data = Uri.parse("package:" + context.packageName)
+                context.startActivity(intent)
+            }
+        }
+    }
+}

app/src/main/java/org/schabi/newpipe/local/feed/notifications/NotificationIcon.kt

@@ -0,0 +1,48 @@
+package org.schabi.newpipe.local.feed.notifications
+
+import android.app.ActivityManager
+import android.content.Context
+import android.graphics.Bitmap
+import android.view.View
+import com.nostra13.universalimageloader.core.ImageLoader
+import com.nostra13.universalimageloader.core.assist.FailReason
+import com.nostra13.universalimageloader.core.assist.ImageSize
+import com.nostra13.universalimageloader.core.listener.SimpleImageLoadingListener
+import io.reactivex.rxjava3.core.SingleEmitter
+import io.reactivex.rxjava3.core.SingleOnSubscribe
+
+internal class NotificationIcon(
+    context: Context,
+    private val url: String
+) : SingleOnSubscribe<Bitmap> {
+
+    private val size = getIconSize(context)
+
+    override fun subscribe(emitter: SingleEmitter<Bitmap>) {
+        ImageLoader.getInstance().loadImage(
+            url,
+            ImageSize(size, size),
+            object : SimpleImageLoadingListener() {
+                override fun onLoadingFailed(imageUri: String?, view: View?, failReason: FailReason) {
+                    emitter.onError(failReason.cause)
+                }
+
+                override fun onLoadingComplete(imageUri: String?, view: View?, loadedImage: Bitmap) {
+                    emitter.onSuccess(loadedImage)
+                }
+            }
+        )
+    }
+
+    private companion object {
+
+        fun getIconSize(context: Context): Int {
+            val activityManager = context.getSystemService(
+                Context.ACTIVITY_SERVICE
+            ) as ActivityManager?
+            val size1 = activityManager?.launcherLargeIconSize ?: 0
+            val size2 = context.resources.getDimensionPixelSize(android.R.dimen.app_icon_size)
+            return maxOf(size2, size1)
+        }
+    }
+}

app/src/main/java/org/schabi/newpipe/local/feed/notifications/NotificationWorker.kt

@@ -1,4 +1,4 @@
-package org.schabi.newpipe.notifications
+package org.schabi.newpipe.local.feed.notifications
 
 import android.content.Context
 import androidx.preference.PreferenceManager
@@ -6,14 +6,16 @@ import androidx.work.BackoffPolicy
 import androidx.work.Constraints
 import androidx.work.ExistingPeriodicWorkPolicy
 import androidx.work.NetworkType
+import androidx.work.OneTimeWorkRequestBuilder
 import androidx.work.PeriodicWorkRequest
-import androidx.work.RxWorker
 import androidx.work.WorkManager
 import androidx.work.WorkerParameters
-import io.reactivex.BackpressureStrategy
+import androidx.work.rxjava3.RxWorker
-import io.reactivex.Flowable
+import io.reactivex.rxjava3.core.Observable
-import io.reactivex.Single
+import io.reactivex.rxjava3.core.Single
 import org.schabi.newpipe.R
+import org.schabi.newpipe.database.subscription.NotificationMode
+import org.schabi.newpipe.local.feed.service.FeedLoadManager
 import java.util.concurrent.TimeUnit
 
 class NotificationWorker(
@@ -24,20 +26,27 @@ class NotificationWorker(
     private val notificationHelper by lazy {
         NotificationHelper(appContext)
     }
+    private val feedLoadManager = FeedLoadManager(appContext)
 
-    override fun createWork() = if (isEnabled(applicationContext)) {
+    override fun createWork(): Single<Result> = if (isEnabled(applicationContext)) {
-        Flowable.create(
+        feedLoadManager.startLoading()
-            SubscriptionUpdates(applicationContext),
+            .map { feed ->
-            BackpressureStrategy.BUFFER
+                feed.mapNotNull { x ->
-        ).doOnNext { notificationHelper.notify(it) }
+                    x.value?.takeIf {
-            .toList()
+                        it.notificationMode == NotificationMode.ENABLED_DEFAULT &&
-            .map { Result.success() }
+                            it.newStreamsCount > 0
+                    }
+                }
+            }
+            .flatMapObservable { Observable.fromIterable(it) }
+            .flatMapCompletable { x -> notificationHelper.notify(x) }
+            .toSingleDefault(Result.success())
             .onErrorReturnItem(Result.failure())
     } else Single.just(Result.success())
 
     companion object {
 
-        private const val TAG = "notifications"
+        private const val TAG = "streams_notifications"
 
         private fun isEnabled(context: Context): Boolean {
             return PreferenceManager.getDefaultSharedPreferences(context)
@@ -78,5 +87,13 @@ class NotificationWorker(
 
         @JvmStatic
         fun schedule(context: Context) = schedule(context, ScheduleOptions.from(context))
+
+        @JvmStatic
+        fun runNow(context: Context) {
+            val request = OneTimeWorkRequestBuilder<NotificationWorker>()
+                .addTag(TAG)
+                .build()
+            WorkManager.getInstance(context).enqueue(request)
+        }
     }
 }

app/src/main/java/org/schabi/newpipe/local/feed/notifications/ScheduleOptions.kt

@@ -1,4 +1,4 @@
-package org.schabi.newpipe.notifications
+package org.schabi.newpipe.local.feed.notifications
 
 import android.content.Context
 import androidx.preference.PreferenceManager

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

@@ -0,0 +1,217 @@
+package org.schabi.newpipe.local.feed.service
+
+import android.content.Context
+import androidx.preference.PreferenceManager
+import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
+import io.reactivex.rxjava3.core.Completable
+import io.reactivex.rxjava3.core.Flowable
+import io.reactivex.rxjava3.core.Notification
+import io.reactivex.rxjava3.core.Single
+import io.reactivex.rxjava3.functions.Consumer
+import io.reactivex.rxjava3.processors.PublishProcessor
+import io.reactivex.rxjava3.schedulers.Schedulers
+import org.schabi.newpipe.R
+import org.schabi.newpipe.database.feed.model.FeedGroupEntity
+import org.schabi.newpipe.extractor.ListInfo
+import org.schabi.newpipe.extractor.stream.StreamInfoItem
+import org.schabi.newpipe.local.feed.FeedDatabaseManager
+import org.schabi.newpipe.local.subscription.SubscriptionManager
+import org.schabi.newpipe.util.ExtractorHelper
+import java.time.OffsetDateTime
+import java.time.ZoneOffset
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
+
+class FeedLoadManager(private val context: Context) {
+
+    private val subscriptionManager = SubscriptionManager(context)
+    private val feedDatabaseManager = FeedDatabaseManager(context)
+
+    private val notificationUpdater = PublishProcessor.create<String>()
+    private val currentProgress = AtomicInteger(-1)
+    private val maxProgress = AtomicInteger(-1)
+    private val cancelSignal = AtomicBoolean()
+    private val feedResultsHolder = FeedResultsHolder()
+
+    val notification: Flowable<FeedLoadState> = notificationUpdater.map { description ->
+        FeedLoadState(description, maxProgress.get(), currentProgress.get())
+    }
+
+    fun startLoading(
+        groupId: Long = FeedGroupEntity.GROUP_ALL_ID
+    ): Single<List<Notification<FeedUpdateInfo>>> {
+        val defaultSharedPreferences = PreferenceManager.getDefaultSharedPreferences(context)
+        val useFeedExtractor = defaultSharedPreferences.getBoolean(
+            context.getString(R.string.feed_use_dedicated_fetch_method_key),
+            false
+        )
+        val thresholdOutdatedSeconds = defaultSharedPreferences.getString(
+            context.getString(R.string.feed_update_threshold_key),
+            context.getString(R.string.feed_update_threshold_default_value)
+        )!!.toInt()
+
+        val outdatedThreshold = OffsetDateTime.now(ZoneOffset.UTC).minusSeconds(thresholdOutdatedSeconds.toLong())
+
+        val subscriptions = when (groupId) {
+            FeedGroupEntity.GROUP_ALL_ID -> feedDatabaseManager.outdatedSubscriptions(outdatedThreshold)
+            else -> feedDatabaseManager.outdatedSubscriptionsForGroup(groupId, outdatedThreshold)
+        }
+
+        return subscriptions
+            .take(1)
+
+            .doOnNext {
+                currentProgress.set(0)
+                maxProgress.set(it.size)
+            }
+            .filter { it.isNotEmpty() }
+
+            .observeOn(AndroidSchedulers.mainThread())
+            .doOnNext {
+                notificationUpdater.onNext("")
+                broadcastProgress()
+            }
+
+            .observeOn(Schedulers.io())
+            .flatMap { Flowable.fromIterable(it) }
+            .takeWhile { !cancelSignal.get() }
+
+            .parallel(PARALLEL_EXTRACTIONS, PARALLEL_EXTRACTIONS * 2)
+            .runOn(Schedulers.io(), PARALLEL_EXTRACTIONS * 2)
+            .filter { !cancelSignal.get() }
+
+            .map { subscriptionEntity ->
+                var error: Throwable? = null
+                try {
+                    val listInfo = if (useFeedExtractor) {
+                        ExtractorHelper
+                            .getFeedInfoFallbackToChannelInfo(subscriptionEntity.serviceId, subscriptionEntity.url)
+                            .onErrorReturn {
+                                error = it // store error, otherwise wrapped into RuntimeException
+                                throw it
+                            }
+                            .blockingGet()
+                    } else {
+                        ExtractorHelper
+                            .getChannelInfo(subscriptionEntity.serviceId, subscriptionEntity.url, true)
+                            .onErrorReturn {
+                                error = it // store error, otherwise wrapped into RuntimeException
+                                throw it
+                            }
+                            .blockingGet()
+                    } as ListInfo<StreamInfoItem>
+
+                    return@map Notification.createOnNext(FeedUpdateInfo(subscriptionEntity, listInfo))
+                } catch (e: Throwable) {
+                    if (error == null) {
+                        // do this to prevent blockingGet() from wrapping into RuntimeException
+                        error = e
+                    }
+
+                    val request = "${subscriptionEntity.serviceId}:${subscriptionEntity.url}"
+                    val wrapper = FeedLoadService.RequestException(subscriptionEntity.uid, request, error!!)
+                    return@map Notification.createOnError<FeedUpdateInfo>(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()))
+    }
+
+    private fun postProcessFeed() = Completable.fromRunnable {
+        FeedEventManager.postEvent(FeedEventManager.Event.ProgressEvent(R.string.feed_processing_message))
+        feedDatabaseManager.removeOrphansOrOlderStreams()
+
+        FeedEventManager.postEvent(FeedEventManager.Event.SuccessResultEvent(feedResultsHolder.itemsErrors))
+    }.doOnSubscribe {
+        currentProgress.set(-1)
+        maxProgress.set(-1)
+
+        notificationUpdater.onNext(context.getString(R.string.feed_processing_message))
+        FeedEventManager.postEvent(FeedEventManager.Event.ProgressEvent(R.string.feed_processing_message))
+    }.subscribeOn(Schedulers.io())
+
+    private inner class NotificationConsumer : Consumer<Notification<FeedUpdateInfo>> {
+        override fun accept(item: Notification<FeedUpdateInfo>) {
+            currentProgress.incrementAndGet()
+            notificationUpdater.onNext(item.value?.name.orEmpty())
+
+            broadcastProgress()
+        }
+    }
+
+    private inner class DatabaseConsumer : Consumer<List<Notification<FeedUpdateInfo>>> {
+
+        override fun accept(list: List<Notification<FeedUpdateInfo>>) {
+            feedDatabaseManager.database().runInTransaction {
+                for (notification in list) {
+                    when {
+                        notification.isOnNext -> {
+                            val subscriptionId = notification.value.uid
+                            val info = notification.value.listInfo
+
+                            notification.value.newStreamsCount = countNewStreams(info.relatedItems)
+                            feedDatabaseManager.upsertAll(subscriptionId, info.relatedItems)
+                            subscriptionManager.updateFromInfo(subscriptionId, info)
+
+                            if (info.errors.isNotEmpty()) {
+                                feedResultsHolder.addErrors(FeedLoadService.RequestException.wrapList(subscriptionId, info))
+                                feedDatabaseManager.markAsOutdated(subscriptionId)
+                            }
+                        }
+                        notification.isOnError -> {
+                            val error = notification.error
+                            feedResultsHolder.addError(error)
+
+                            if (error is FeedLoadService.RequestException) {
+                                feedDatabaseManager.markAsOutdated(error.subscriptionId)
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        private fun countNewStreams(list: List<StreamInfoItem>): Int {
+            var count = 0
+            for (item in list) {
+                if (feedDatabaseManager.isStreamExist(item)) {
+                    return count
+                } else {
+                    count++
+                }
+            }
+            return 0
+        }
+    }
+
+    private companion object {
+
+        /**
+         * How many extractions will be running in parallel.
+         */
+        const val PARALLEL_EXTRACTIONS = 6
+
+        /**
+         * Number of items to buffer to mass-insert in the database.
+         */
+        const val BUFFER_COUNT_BEFORE_INSERT = 20
+    }
+}

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

@@ -31,36 +31,19 @@ import android.util.Log
 import androidx.core.app.NotificationCompat
 import androidx.core.app.NotificationManagerCompat
 import androidx.core.app.ServiceCompat
-import androidx.preference.PreferenceManager
 import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.Flowable
-import io.reactivex.rxjava3.core.Notification
+import io.reactivex.rxjava3.disposables.Disposable
-import io.reactivex.rxjava3.core.Single
-import io.reactivex.rxjava3.disposables.CompositeDisposable
-import io.reactivex.rxjava3.functions.Consumer
 import io.reactivex.rxjava3.functions.Function
-import io.reactivex.rxjava3.processors.PublishProcessor
-import io.reactivex.rxjava3.schedulers.Schedulers
-import org.reactivestreams.Subscriber
-import org.reactivestreams.Subscription
 import org.schabi.newpipe.App
 import org.schabi.newpipe.MainActivity.DEBUG
 import org.schabi.newpipe.R
 import org.schabi.newpipe.database.feed.model.FeedGroupEntity
 import org.schabi.newpipe.extractor.ListInfo
 import org.schabi.newpipe.extractor.stream.StreamInfoItem
-import org.schabi.newpipe.local.feed.FeedDatabaseManager
 import org.schabi.newpipe.local.feed.service.FeedEventManager.Event.ErrorResultEvent
-import org.schabi.newpipe.local.feed.service.FeedEventManager.Event.ProgressEvent
-import org.schabi.newpipe.local.feed.service.FeedEventManager.Event.SuccessResultEvent
 import org.schabi.newpipe.local.feed.service.FeedEventManager.postEvent
-import org.schabi.newpipe.local.subscription.SubscriptionManager
-import org.schabi.newpipe.util.ExtractorHelper
-import java.time.OffsetDateTime
-import java.time.ZoneOffset
 import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.atomic.AtomicInteger
 
 class FeedLoadService : Service() {
     companion object {
@@ -73,27 +56,13 @@ class FeedLoadService : Service() {
          */
         private const val NOTIFICATION_SAMPLING_PERIOD = 1500
 
-        /**
-         * How many extractions will be running in parallel.
-         */
-        private const val PARALLEL_EXTRACTIONS = 6
-
-        /**
-         * Number of items to buffer to mass-insert in the database.
-         */
-        private const val BUFFER_COUNT_BEFORE_INSERT = 20
-
         const val EXTRA_GROUP_ID: String = "FeedLoadService.EXTRA_GROUP_ID"
     }
 
-    private var loadingSubscription: Subscription? = null
+    private var loadingDisposable: Disposable? = null
-    private lateinit var subscriptionManager: SubscriptionManager
+    private var notificationDisposable: Disposable? = null
 
-    private lateinit var feedDatabaseManager: FeedDatabaseManager
+    private lateinit var feedLoadManager: FeedLoadManager
-    private lateinit var feedResultsHolder: ResultsHolder
-
-    private var disposables = CompositeDisposable()
-    private var notificationUpdater = PublishProcessor.create<String>()
 
     // /////////////////////////////////////////////////////////////////////////
     // Lifecycle
@@ -101,8 +70,7 @@ class FeedLoadService : Service() {
 
     override fun onCreate() {
         super.onCreate()
-        subscriptionManager = SubscriptionManager(this)
+        feedLoadManager = FeedLoadManager(this)
-        feedDatabaseManager = FeedDatabaseManager(this)
     }
 
     override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
@@ -114,40 +82,45 @@ class FeedLoadService : Service() {
             )
         }
 
-        if (intent == null || loadingSubscription != null) {
+        if (intent == null || loadingDisposable != null) {
             return START_NOT_STICKY
         }
 
         setupNotification()
         setupBroadcastReceiver()
-        val defaultSharedPreferences = PreferenceManager.getDefaultSharedPreferences(this)
 
         val groupId = intent.getLongExtra(EXTRA_GROUP_ID, FeedGroupEntity.GROUP_ALL_ID)
-        val useFeedExtractor = defaultSharedPreferences
+        loadingDisposable = feedLoadManager.startLoading(groupId)
-            .getBoolean(getString(R.string.feed_use_dedicated_fetch_method_key), false)
+            .observeOn(AndroidSchedulers.mainThread())
-
+            .doOnSubscribe {
-        val thresholdOutdatedSecondsString = defaultSharedPreferences
+                startForeground(NOTIFICATION_ID, notificationBuilder.build())
-            .getString(getString(R.string.feed_update_threshold_key), getString(R.string.feed_update_threshold_default_value))
+            }
-        val thresholdOutdatedSeconds = thresholdOutdatedSecondsString!!.toInt()
+            .subscribe { _, error ->
-
+                // There seems to be a bug in the kotlin plugin as it tells you when
-        startLoading(groupId, useFeedExtractor, thresholdOutdatedSeconds)
+                // building that this can't be null:
-
+                // "Condition 'error != null' is always 'true'"
+                // However it can indeed be null
+                // The suppression may be removed in further versions
+                @Suppress("SENSELESS_COMPARISON")
+                if (error != null) {
+                    Log.e(TAG, "Error while storing result", error)
+                    handleError(error)
+                    return@subscribe
+                }
+                stopService()
+            }
         return START_NOT_STICKY
     }
 
     private fun disposeAll() {
         unregisterReceiver(broadcastReceiver)
-
+        loadingDisposable?.dispose()
-        loadingSubscription?.cancel()
+        notificationDisposable?.dispose()
-        loadingSubscription = null
-
-        disposables.dispose()
     }
 
     private fun stopService() {
         disposeAll()
         ServiceCompat.stopForeground(this, ServiceCompat.STOP_FOREGROUND_REMOVE)
-        notificationManager.cancel(NOTIFICATION_ID)
         stopSelf()
     }
 
@@ -171,190 +144,6 @@ class FeedLoadService : Service() {
         }
     }
 
-    private fun startLoading(groupId: Long = FeedGroupEntity.GROUP_ALL_ID, useFeedExtractor: Boolean, thresholdOutdatedSeconds: Int) {
-        feedResultsHolder = ResultsHolder()
-
-        val outdatedThreshold = OffsetDateTime.now(ZoneOffset.UTC).minusSeconds(thresholdOutdatedSeconds.toLong())
-
-        val subscriptions = when (groupId) {
-            FeedGroupEntity.GROUP_ALL_ID -> feedDatabaseManager.outdatedSubscriptions(outdatedThreshold)
-            else -> feedDatabaseManager.outdatedSubscriptionsForGroup(groupId, outdatedThreshold)
-        }
-
-        subscriptions
-            .take(1)
-
-            .doOnNext {
-                currentProgress.set(0)
-                maxProgress.set(it.size)
-            }
-            .filter { it.isNotEmpty() }
-
-            .observeOn(AndroidSchedulers.mainThread())
-            .doOnNext {
-                startForeground(NOTIFICATION_ID, notificationBuilder.build())
-                updateNotificationProgress(null)
-                broadcastProgress()
-            }
-
-            .observeOn(Schedulers.io())
-            .flatMap { Flowable.fromIterable(it) }
-            .takeWhile { !cancelSignal.get() }
-
-            .parallel(PARALLEL_EXTRACTIONS, PARALLEL_EXTRACTIONS * 2)
-            .runOn(Schedulers.io(), PARALLEL_EXTRACTIONS * 2)
-            .filter { !cancelSignal.get() }
-
-            .map { subscriptionEntity ->
-                var error: Throwable? = null
-                try {
-                    val listInfo = if (useFeedExtractor) {
-                        ExtractorHelper
-                            .getFeedInfoFallbackToChannelInfo(subscriptionEntity.serviceId, subscriptionEntity.url)
-                            .onErrorReturn {
-                                error = it // store error, otherwise wrapped into RuntimeException
-                                throw it
-                            }
-                            .blockingGet()
-                    } else {
-                        ExtractorHelper
-                            .getChannelInfo(subscriptionEntity.serviceId, subscriptionEntity.url, true)
-                            .onErrorReturn {
-                                error = it // store error, otherwise wrapped into RuntimeException
-                                throw it
-                            }
-                            .blockingGet()
-                    } as ListInfo<StreamInfoItem>
-
-                    return@map Notification.createOnNext(Pair(subscriptionEntity.uid, listInfo))
-                } catch (e: Throwable) {
-                    if (error == null) {
-                        // do this to prevent blockingGet() from wrapping into RuntimeException
-                        error = e
-                    }
-
-                    val request = "${subscriptionEntity.serviceId}:${subscriptionEntity.url}"
-                    val wrapper = RequestException(subscriptionEntity.uid, request, error!!)
-                    return@map Notification.createOnError<Pair<Long, ListInfo<StreamInfoItem>>>(wrapper)
-                }
-            }
-            .sequential()
-
-            .observeOn(AndroidSchedulers.mainThread())
-            .doOnNext(notificationsConsumer)
-
-            .observeOn(Schedulers.io())
-            .buffer(BUFFER_COUNT_BEFORE_INSERT)
-            .doOnNext(databaseConsumer)
-
-            .subscribeOn(Schedulers.io())
-            .observeOn(AndroidSchedulers.mainThread())
-            .subscribe(resultSubscriber)
-    }
-
-    private fun broadcastProgress() {
-        postEvent(ProgressEvent(currentProgress.get(), maxProgress.get()))
-    }
-
-    private val resultSubscriber
-        get() = object : Subscriber<List<Notification<Pair<Long, ListInfo<StreamInfoItem>>>>> {
-
-            override fun onSubscribe(s: Subscription) {
-                loadingSubscription = s
-                s.request(java.lang.Long.MAX_VALUE)
-            }
-
-            override fun onNext(notification: List<Notification<Pair<Long, ListInfo<StreamInfoItem>>>>) {
-                if (DEBUG) Log.v(TAG, "onNext() → $notification")
-            }
-
-            override fun onError(error: Throwable) {
-                handleError(error)
-            }
-
-            override fun onComplete() {
-                if (maxProgress.get() == 0) {
-                    postEvent(FeedEventManager.Event.IdleEvent)
-                    stopService()
-
-                    return
-                }
-
-                currentProgress.set(-1)
-                maxProgress.set(-1)
-
-                notificationUpdater.onNext(getString(R.string.feed_processing_message))
-                postEvent(ProgressEvent(R.string.feed_processing_message))
-
-                disposables.add(
-                    Single
-                        .fromCallable {
-                            feedResultsHolder.ready()
-
-                            postEvent(ProgressEvent(R.string.feed_processing_message))
-                            feedDatabaseManager.removeOrphansOrOlderStreams()
-
-                            postEvent(SuccessResultEvent(feedResultsHolder.itemsErrors))
-                            true
-                        }
-                        .subscribeOn(Schedulers.io())
-                        .observeOn(AndroidSchedulers.mainThread())
-                        .subscribe { _, throwable ->
-                            // There seems to be a bug in the kotlin plugin as it tells you when
-                            // building that this can't be null:
-                            // "Condition 'throwable != null' is always 'true'"
-                            // However it can indeed be null
-                            // The suppression may be removed in further versions
-                            @Suppress("SENSELESS_COMPARISON")
-                            if (throwable != null) {
-                                Log.e(TAG, "Error while storing result", throwable)
-                                handleError(throwable)
-                                return@subscribe
-                            }
-                            stopService()
-                        }
-                )
-            }
-        }
-
-    private val databaseConsumer: Consumer<List<Notification<Pair<Long, ListInfo<StreamInfoItem>>>>>
-        get() = Consumer {
-            feedDatabaseManager.database().runInTransaction {
-                for (notification in it) {
-
-                    if (notification.isOnNext) {
-                        val subscriptionId = notification.value!!.first
-                        val info = notification.value!!.second
-
-                        feedDatabaseManager.upsertAll(subscriptionId, info.relatedItems)
-                        subscriptionManager.updateFromInfo(subscriptionId, info)
-
-                        if (info.errors.isNotEmpty()) {
-                            feedResultsHolder.addErrors(RequestException.wrapList(subscriptionId, info))
-                            feedDatabaseManager.markAsOutdated(subscriptionId)
-                        }
-                    } else if (notification.isOnError) {
-                        val error = notification.error!!
-                        feedResultsHolder.addError(error)
-
-                        if (error is RequestException) {
-                            feedDatabaseManager.markAsOutdated(error.subscriptionId)
-                        }
-                    }
-                }
-            }
-        }
-
-    private val notificationsConsumer: Consumer<Notification<Pair<Long, ListInfo<StreamInfoItem>>>>
-        get() = Consumer { onItemCompleted(it.value?.second?.name) }
-
-    private fun onItemCompleted(updateDescription: String?) {
-        currentProgress.incrementAndGet()
-        notificationUpdater.onNext(updateDescription ?: "")
-
-        broadcastProgress()
-    }
-
     // /////////////////////////////////////////////////////////////////////////
     // Notification
     // /////////////////////////////////////////////////////////////////////////
@@ -362,9 +151,6 @@ class FeedLoadService : Service() {
     private lateinit var notificationManager: NotificationManagerCompat
     private lateinit var notificationBuilder: NotificationCompat.Builder
 
-    private var currentProgress = AtomicInteger(-1)
-    private var maxProgress = AtomicInteger(-1)
-
     private fun createNotification(): NotificationCompat.Builder {
         val cancelActionIntent = PendingIntent.getBroadcast(
             this,
@@ -384,33 +170,36 @@ class FeedLoadService : Service() {
         notificationManager = NotificationManagerCompat.from(this)
         notificationBuilder = createNotification()
 
-        val throttleAfterFirstEmission = Function { flow: Flowable<String> ->
+        val throttleAfterFirstEmission = Function { flow: Flowable<FeedLoadState> ->
             flow.take(1).concatWith(flow.skip(1).throttleLatest(NOTIFICATION_SAMPLING_PERIOD.toLong(), TimeUnit.MILLISECONDS))
         }
 
-        disposables.add(
+        notificationDisposable = feedLoadManager.notification
-            notificationUpdater
+            .publish(throttleAfterFirstEmission)
-                .publish(throttleAfterFirstEmission)
+            .observeOn(AndroidSchedulers.mainThread())
-                .observeOn(AndroidSchedulers.mainThread())
+            .doOnTerminate { notificationManager.cancel(NOTIFICATION_ID) }
-                .subscribe(this::updateNotificationProgress)
+            .subscribe(this::updateNotificationProgress)
-        )
     }
 
-    private fun updateNotificationProgress(updateDescription: String?) {
+    private fun updateNotificationProgress(state: FeedLoadState) {
-        notificationBuilder.setProgress(maxProgress.get(), currentProgress.get(), maxProgress.get() == -1)
+        notificationBuilder.setProgress(state.maxProgress, state.currentProgress, state.maxProgress == -1)
 
-        if (maxProgress.get() == -1) {
+        if (state.maxProgress == -1) {
             if (Build.VERSION.SDK_INT < Build.VERSION_CODES.N) notificationBuilder.setContentInfo(null)
-            if (!updateDescription.isNullOrEmpty()) notificationBuilder.setContentText(updateDescription)
+            if (state.updateDescription.isNotEmpty()) notificationBuilder.setContentText(state.updateDescription)
-            notificationBuilder.setContentText(updateDescription)
+            notificationBuilder.setContentText(state.updateDescription)
         } else {
-            val progressText = this.currentProgress.toString() + "/" + maxProgress
+            val progressText = state.currentProgress.toString() + "/" + state.maxProgress
 
             if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) {
-                if (!updateDescription.isNullOrEmpty()) notificationBuilder.setContentText("$updateDescription  ($progressText)")
+                if (state.updateDescription.isNotEmpty()) {
+                    notificationBuilder.setContentText("${state.updateDescription}  ($progressText)")
+                }
             } else {
                 notificationBuilder.setContentInfo(progressText)
-                if (!updateDescription.isNullOrEmpty()) notificationBuilder.setContentText(updateDescription)
+                if (state.updateDescription.isNotEmpty()) {
+                    notificationBuilder.setContentText(state.updateDescription)
+                }
             }
         }
 
@@ -422,13 +211,12 @@ class FeedLoadService : Service() {
     // /////////////////////////////////////////////////////////////////////////
 
     private lateinit var broadcastReceiver: BroadcastReceiver
-    private val cancelSignal = AtomicBoolean()
 
     private fun setupBroadcastReceiver() {
         broadcastReceiver = object : BroadcastReceiver() {
             override fun onReceive(context: Context?, intent: Intent?) {
                 if (intent?.action == ACTION_CANCEL) {
-                    cancelSignal.set(true)
+                    feedLoadManager.cancel()
                 }
             }
         }
@@ -443,29 +231,4 @@ class FeedLoadService : Service() {
         postEvent(ErrorResultEvent(error))
         stopService()
     }
-
-    // /////////////////////////////////////////////////////////////////////////
-    // Results Holder
-    // /////////////////////////////////////////////////////////////////////////
-
-    class ResultsHolder {
-        /**
-         * List of errors that may have happen during loading.
-         */
-        internal lateinit var itemsErrors: List<Throwable>
-
-        private val itemsErrorsHolder: MutableList<Throwable> = ArrayList()
-
-        fun addError(error: Throwable) {
-            itemsErrorsHolder.add(error)
-        }
-
-        fun addErrors(errors: List<Throwable>) {
-            itemsErrorsHolder.addAll(errors)
-        }
-
-        fun ready() {
-            itemsErrors = itemsErrorsHolder.toList()
-        }
-    }
 }

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

@@ -0,0 +1,7 @@
+package org.schabi.newpipe.local.feed.service
+
+data class FeedLoadState(
+    val updateDescription: String,
+    val maxProgress: Int,
+    val currentProgress: Int,
+)

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

@@ -0,0 +1,19 @@
+package org.schabi.newpipe.local.feed.service
+
+class FeedResultsHolder {
+    /**
+     * List of errors that may have happen during loading.
+     */
+    val itemsErrors: List<Throwable>
+        get() = itemsErrorsHolder
+
+    private val itemsErrorsHolder: MutableList<Throwable> = ArrayList()
+
+    fun addError(error: Throwable) {
+        itemsErrorsHolder.add(error)
+    }
+
+    fun addErrors(errors: List<Throwable>) {
+        itemsErrorsHolder.addAll(errors)
+    }
+}

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

@@ -0,0 +1,34 @@
+package org.schabi.newpipe.local.feed.service
+
+import org.schabi.newpipe.database.subscription.NotificationMode
+import org.schabi.newpipe.database.subscription.SubscriptionEntity
+import org.schabi.newpipe.extractor.ListInfo
+import org.schabi.newpipe.extractor.stream.StreamInfoItem
+
+data class FeedUpdateInfo(
+    val uid: Long,
+    @NotificationMode
+    val notificationMode: Int,
+    val name: String,
+    val avatarUrl: String,
+    val listInfo: ListInfo<StreamInfoItem>
+) {
+    constructor(subscription: SubscriptionEntity, listInfo: ListInfo<StreamInfoItem>) : this(
+        uid = subscription.uid,
+        notificationMode = subscription.notificationMode,
+        name = subscription.name,
+        avatarUrl = subscription.avatarUrl,
+        listInfo = listInfo
+    )
+
+    /**
+     * Integer id, can be used as notification id, etc.
+     */
+    val pseudoId: Int
+        get() = listInfo.url.hashCode()
+
+    var newStreamsCount: Int = 0
+
+    val newStreams: List<StreamInfoItem>
+        get() = listInfo.relatedItems.take(newStreamsCount)
+}

app/src/main/java/org/schabi/newpipe/notifications/ChannelUpdates.kt

@@ -1,46 +0,0 @@
-package org.schabi.newpipe.notifications
-
-import android.content.Context
-import android.content.Intent
-import org.schabi.newpipe.R
-import org.schabi.newpipe.extractor.channel.ChannelInfo
-import org.schabi.newpipe.extractor.stream.StreamInfoItem
-import org.schabi.newpipe.util.NavigationHelper
-
-data class ChannelUpdates(
-    val serviceId: Int,
-    val url: String,
-    val avatarUrl: String,
-    val name: String,
-    val streams: List<StreamInfoItem>
-) {
-
-    val id = url.hashCode()
-
-    val isNotEmpty: Boolean
-        get() = streams.isNotEmpty()
-
-    val size = streams.size
-
-    fun getText(context: Context): String {
-        val separator = context.resources.getString(R.string.enumeration_comma) + " "
-        return streams.joinToString(separator) { it.name }
-    }
-
-    fun createOpenChannelIntent(context: Context?): Intent {
-        return NavigationHelper.getChannelIntent(context, serviceId, url)
-            .setFlags(Intent.FLAG_ACTIVITY_NEW_TASK)
-    }
-
-    companion object {
-        fun from(channel: ChannelInfo, streams: List<StreamInfoItem>): ChannelUpdates {
-            return ChannelUpdates(
-                channel.serviceId,
-                channel.url,
-                channel.avatarUrl,
-                channel.name,
-                streams
-            )
-        }
-    }
-}

app/src/main/java/org/schabi/newpipe/notifications/NotificationHelper.java

@@ -1,137 +0,0 @@
-package org.schabi.newpipe.notifications;
-
-import android.app.NotificationChannel;
-import android.app.NotificationManager;
-import android.app.PendingIntent;
-import android.content.Context;
-import android.content.Intent;
-import android.graphics.BitmapFactory;
-import android.net.Uri;
-import android.os.Build;
-import android.provider.Settings;
-
-import androidx.annotation.NonNull;
-import androidx.core.app.NotificationCompat;
-import androidx.core.app.NotificationManagerCompat;
-import androidx.core.content.ContextCompat;
-import androidx.preference.PreferenceManager;
-
-import org.schabi.newpipe.BuildConfig;
-import org.schabi.newpipe.R;
-import org.schabi.newpipe.extractor.stream.StreamInfoItem;
-
-import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
-import io.reactivex.rxjava3.core.Single;
-import io.reactivex.rxjava3.disposables.CompositeDisposable;
-import io.reactivex.rxjava3.schedulers.Schedulers;
-
-public final class NotificationHelper {
-
-    private final Context context;
-    private final NotificationManager manager;
-    private final CompositeDisposable disposable;
-
-    public NotificationHelper(final Context context) {
-        this.context = context;
-        this.disposable = new CompositeDisposable();
-        this.manager = (NotificationManager) context.getSystemService(
-                Context.NOTIFICATION_SERVICE
-        );
-    }
-
-    public Context getContext() {
-        return context;
-    }
-
-    /**
-     * Check whether notifications are not disabled by user via system settings.
-     *
-     * @param context Context
-     * @return true if notifications are allowed, false otherwise
-     */
-    public static boolean isNotificationsEnabledNative(final Context context) {
-        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
-            final String channelId = context.getString(R.string.streams_notification_channel_id);
-            final NotificationManager manager = (NotificationManager) context
-                    .getSystemService(Context.NOTIFICATION_SERVICE);
-            if (manager != null) {
-                final NotificationChannel channel = manager.getNotificationChannel(channelId);
-                return channel != null
-                        && channel.getImportance() != NotificationManager.IMPORTANCE_NONE;
-            } else {
-                return false;
-            }
-        } else {
-            return NotificationManagerCompat.from(context).areNotificationsEnabled();
-        }
-    }
-
-    public static boolean isNewStreamsNotificationsEnabled(@NonNull final Context context) {
-        return PreferenceManager.getDefaultSharedPreferences(context)
-                .getBoolean(context.getString(R.string.enable_streams_notifications), false)
-                && isNotificationsEnabledNative(context);
-    }
-
-    public static void openNativeSettingsScreen(final Context context) {
-        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
-            final String channelId = context.getString(R.string.streams_notification_channel_id);
-            final Intent intent = new Intent(Settings.ACTION_CHANNEL_NOTIFICATION_SETTINGS)
-                    .putExtra(Settings.EXTRA_APP_PACKAGE, context.getPackageName())
-                    .putExtra(Settings.EXTRA_CHANNEL_ID, channelId);
-            context.startActivity(intent);
-        } else {
-            final Intent intent = new Intent(Settings.ACTION_APPLICATION_DETAILS_SETTINGS);
-            intent.setData(Uri.parse("package:" + context.getPackageName()));
-            context.startActivity(intent);
-        }
-    }
-
-    public void notify(final ChannelUpdates data) {
-        final String summary = context.getResources().getQuantityString(
-                R.plurals.new_streams, data.getSize(), data.getSize()
-        );
-        final NotificationCompat.Builder builder = new NotificationCompat.Builder(context,
-                context.getString(R.string.streams_notification_channel_id))
-                .setContentTitle(
-                        context.getString(R.string.notification_title_pattern,
-                                data.getName(),
-                                summary)
-                )
-                .setContentText(data.getText(context))
-                .setNumber(data.getSize())
-                .setBadgeIconType(NotificationCompat.BADGE_ICON_LARGE)
-                .setPriority(NotificationCompat.PRIORITY_DEFAULT)
-                .setSmallIcon(R.drawable.ic_stat_newpipe)
-                .setLargeIcon(BitmapFactory.decodeResource(context.getResources(),
-                        R.drawable.ic_newpipe_triangle_white))
-                .setColor(ContextCompat.getColor(context, R.color.ic_launcher_background))
-                .setColorized(true)
-                .setAutoCancel(true)
-                .setCategory(NotificationCompat.CATEGORY_SOCIAL);
-        final NotificationCompat.InboxStyle style = new NotificationCompat.InboxStyle();
-        for (final StreamInfoItem stream : data.getStreams()) {
-            style.addLine(stream.getName());
-        }
-        style.setSummaryText(summary);
-        style.setBigContentTitle(data.getName());
-        builder.setStyle(style);
-        builder.setContentIntent(PendingIntent.getActivity(
-                context,
-                data.getId(),
-                data.createOpenChannelIntent(context),
-                0
-        ));
-
-        disposable.add(
-                Single.create(new NotificationIcon(context, data.getAvatarUrl()))
-                        .subscribeOn(Schedulers.io())
-                        .observeOn(AndroidSchedulers.mainThread())
-                        .doAfterTerminate(() -> manager.notify(data.getId(), builder.build()))
-                        .subscribe(builder::setLargeIcon, throwable -> {
-                            if (BuildConfig.DEBUG) {
-                                throwable.printStackTrace();
-                            }
-                        })
-        );
-    }
-}

app/src/main/java/org/schabi/newpipe/notifications/NotificationIcon.java

@@ -1,60 +0,0 @@
-package org.schabi.newpipe.notifications;
-
-import android.app.ActivityManager;
-import android.content.Context;
-import android.graphics.Bitmap;
-import android.view.View;
-
-import com.nostra13.universalimageloader.core.ImageLoader;
-import com.nostra13.universalimageloader.core.assist.FailReason;
-import com.nostra13.universalimageloader.core.assist.ImageSize;
-import com.nostra13.universalimageloader.core.listener.SimpleImageLoadingListener;
-
-import io.reactivex.rxjava3.annotations.NonNull;
-import io.reactivex.rxjava3.core.SingleEmitter;
-import io.reactivex.rxjava3.core.SingleOnSubscribe;
-
-final class NotificationIcon implements SingleOnSubscribe<Bitmap> {
-
-    private final String url;
-    private final int size;
-
-    NotificationIcon(final Context context, final String url) {
-        this.url = url;
-        this.size = getIconSize(context);
-    }
-
-    @Override
-    public void subscribe(@NonNull final SingleEmitter<Bitmap> emitter) throws Throwable {
-        ImageLoader.getInstance().loadImage(
-                url,
-                new ImageSize(size, size),
-                new SimpleImageLoadingListener() {
-
-                    @Override
-                    public void onLoadingFailed(final String imageUri,
-                                                final View view,
-                                                final FailReason failReason) {
-                        emitter.onError(failReason.getCause());
-                    }
-
-                    @Override
-                    public void onLoadingComplete(final String imageUri,
-                                                  final View view,
-                                                  final Bitmap loadedImage) {
-                        emitter.onSuccess(loadedImage);
-                    }
-                }
-        );
-    }
-
-    private static int getIconSize(final Context context) {
-        final ActivityManager activityManager = (ActivityManager) context.getSystemService(
-                Context.ACTIVITY_SERVICE
-        );
-        final int size2 = activityManager != null ? activityManager.getLauncherLargeIconSize() : 0;
-        final int size1 = context.getResources()
-                .getDimensionPixelSize(android.R.dimen.app_icon_size);
-        return Math.max(size2, size1);
-    }
-}

app/src/main/java/org/schabi/newpipe/notifications/SubscriptionUpdates.kt

@@ -1,53 +0,0 @@
-package org.schabi.newpipe.notifications
-
-import android.content.Context
-import io.reactivex.FlowableEmitter
-import io.reactivex.FlowableOnSubscribe
-import org.schabi.newpipe.NewPipeDatabase
-import org.schabi.newpipe.database.stream.model.StreamEntity
-import org.schabi.newpipe.database.subscription.NotificationMode
-import org.schabi.newpipe.extractor.stream.StreamInfoItem
-import org.schabi.newpipe.local.subscription.SubscriptionManager
-import org.schabi.newpipe.util.ExtractorHelper
-
-class SubscriptionUpdates(context: Context) : FlowableOnSubscribe<ChannelUpdates?> {
-
-    private val subscriptionManager = SubscriptionManager(context)
-    private val streamTable = NewPipeDatabase.getInstance(context).streamDAO()
-
-    override fun subscribe(emitter: FlowableEmitter<ChannelUpdates?>) {
-        try {
-            val subscriptions = subscriptionManager.subscriptions().blockingFirst()
-            for (subscription in subscriptions) {
-                if (subscription.notificationMode != NotificationMode.DISABLED) {
-                    val channel = ExtractorHelper.getChannelInfo(
-                        subscription.serviceId,
-                        subscription.url, true
-                    ).blockingGet()
-                    val updates = ChannelUpdates.from(channel, filterStreams(channel.relatedItems))
-                    if (updates.isNotEmpty) {
-                        emitter.onNext(updates)
-                        // prevent duplicated notifications
-                        streamTable.upsertAll(updates.streams.map { StreamEntity(it) })
-                    }
-                }
-            }
-            emitter.onComplete()
-        } catch (e: Exception) {
-            emitter.onError(e)
-        }
-    }
-
-    private fun filterStreams(list: List<*>): List<StreamInfoItem> {
-        val streams = ArrayList<StreamInfoItem>(list.size)
-        for (o in list) {
-            if (o is StreamInfoItem) {
-                if (streamTable.exists(o.serviceId.toLong(), o.url)) {
-                    break
-                }
-                streams.add(o)
-            }
-        }
-        return streams
-    }
-}

app/src/main/java/org/schabi/newpipe/settings/NotificationsSettingsFragment.kt

@@ -14,10 +14,10 @@ import org.schabi.newpipe.database.subscription.SubscriptionEntity
 import org.schabi.newpipe.error.ErrorActivity
 import org.schabi.newpipe.error.ErrorInfo
 import org.schabi.newpipe.error.UserAction
+import org.schabi.newpipe.local.feed.notifications.NotificationHelper
+import org.schabi.newpipe.local.feed.notifications.NotificationWorker
+import org.schabi.newpipe.local.feed.notifications.ScheduleOptions
 import org.schabi.newpipe.local.subscription.SubscriptionManager
-import org.schabi.newpipe.notifications.NotificationHelper
-import org.schabi.newpipe.notifications.NotificationWorker
-import org.schabi.newpipe.notifications.ScheduleOptions
 
 class NotificationsSettingsFragment : BasePreferenceFragment(), OnSharedPreferenceChangeListener {
 
@@ -47,7 +47,7 @@ class NotificationsSettingsFragment : BasePreferenceFragment(), OnSharedPreferen
 
     override fun onResume() {
         super.onResume()
-        val enabled = NotificationHelper.isNotificationsEnabledNative(context)
+        val enabled = NotificationHelper.isNotificationsEnabledNative(requireContext())
         preferenceScreen.isEnabled = enabled
         if (!enabled) {
             if (notificationWarningSnackbar == null) {

app/src/main/res/drawable-anydpi-v24/ic_stat_newpipe.xml

@@ -1,14 +0,0 @@
-<vector xmlns:android="http://schemas.android.com/apk/res/android"
-    android:width="24dp"
-    android:height="24dp"
-    android:viewportWidth="98.91304"
-    android:viewportHeight="98.91304"
-    android:tint="#FFFFFF">
-  <group android:translateX="-6.7255435"
-      android:translateY="-0.54347825">
-    <path
-        android:pathData="m23.909,10.211v78.869c0,0 7.7,-4.556 12.4,-7.337V67.477,56.739 31.686c0,0 3.707,2.173 8.948,5.24 6.263,3.579 14.57,8.565 21.473,12.655 -9.358,5.483 -16.8,9.876 -22.496,13.234V77.053C57.974,68.927 75.176,58.762 90.762,49.581 75.551,40.634 57.144,29.768 43.467,21.715 31.963,14.94 23.909,10.211 23.909,10.211Z"
-        android:strokeWidth="1.2782383"
-        android:fillColor="#ffffff"/>
-  </group>
-</vector>

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

@@ -188,6 +188,7 @@
     <string name="disable_media_tunneling_key" translatable="false">disable_media_tunneling_key</string>
     <string name="crash_the_app_key" translatable="false">crash_the_app_key</string>
     <string name="show_image_indicators_key" translatable="false">show_image_indicators_key</string>
+    <string name="check_new_streams_key" translatable="false">check_new_streams</string>
 
     <!-- THEMES -->
     <string name="theme_key" translatable="false">theme</string>

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

@@ -481,6 +481,7 @@
     <string name="show_image_indicators_title">Show image indicators</string>
     <string name="show_image_indicators_summary">Show Picasso colored ribbons on top of images indicating their source: red for network, blue for disk and green for memory</string>
     <string name="crash_the_app">Crash the app</string>
+    <string name="check_new_streams">Run check for new streams</string>
     <!-- Subscriptions import/export -->
     <string name="import_title">Import</string>
     <string name="import_from">Import from</string>

app/src/main/res/xml/debug_settings.xml

@@ -49,6 +49,11 @@
         android:title="@string/show_image_indicators_title"
         app:iconSpaceReserved="false" />
 
+    <Preference
+        android:key="@string/check_new_streams_key"
+        android:title="@string/check_new_streams"
+        app:iconSpaceReserved="false" />
+
     <Preference
         android:key="@string/crash_the_app_key"
         android:title="@string/crash_the_app"