/*
 * Copyright (C) 2017 The Android Open Source Project
 *
 * 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 androidx.room;

import androidx.annotation.Nullable;
import androidx.annotation.RestrictTo;
import androidx.arch.core.executor.ArchTaskExecutor;

import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.disposables.Disposables;
import io.reactivex.functions.Action;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;

/**
 * Helper class to add RxJava2 support to Room.
 */
@SuppressWarnings("WeakerAccess")
public class RxRoom {
    /**
     * Data dispatched by the publisher created by {@link #createFlowable(RoomDatabase, String...)}.
     */
    public static final Object NOTHING = new Object();

    /**
     * Creates a {@link Flowable} that emits at least once and also re-emits whenever one of the
     * observed tables is updated.
     * <p>
     * You can easily chain a database operation to downstream of this {@link Flowable} to ensure
     * that it re-runs when database is modified.
     * <p>
     * Since database invalidation is batched, multiple changes in the database may results in just
     * 1 emission.
     *
     * @param database   The database instance
     * @param tableNames The list of table names that should be observed
     * @return A {@link Flowable} which emits {@link #NOTHING} when one of the observed tables
     * is modified (also once when the invalidation tracker connection is established).
     */
    public static Flowable<Object> createFlowable(final RoomDatabase database,
            final String... tableNames) {
        return Flowable.create(new FlowableOnSubscribe<Object>() {
            @Override
            public void subscribe(final FlowableEmitter<Object> emitter) throws Exception {
                final InvalidationTracker.Observer observer = new InvalidationTracker.Observer(
                        tableNames) {
                    @Override
                    public void onInvalidated(@androidx.annotation.NonNull Set<String> tables) {
                        if (!emitter.isCancelled()) {
                            emitter.onNext(NOTHING);
                        }
                    }
                };
                if (!emitter.isCancelled()) {
                    database.getInvalidationTracker().addObserver(observer);
                    emitter.setDisposable(Disposables.fromAction(new Action() {
                        @Override
                        public void run() throws Exception {
                            database.getInvalidationTracker().removeObserver(observer);
                        }
                    }));
                }

                // emit once to avoid missing any data and also easy chaining
                if (!emitter.isCancelled()) {
                    emitter.onNext(NOTHING);
                }
            }
        }, BackpressureStrategy.LATEST);
    }

    /**
     * Helper method used by generated code to bind a Callable such that it will be run in
     * our disk io thread and will automatically block null values since RxJava2 does not like null.
     *
     * @hide
     */
    @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP)
    public static <T> Flowable<T> createFlowable(final RoomDatabase database,
            final String[] tableNames, final Callable<T> callable) {
        return createFlowable(database, tableNames).observeOn(sAppToolkitIOScheduler)
                .map(new Function<Object, Optional<T>>() {
                    @Override
                    public Optional<T> apply(@NonNull Object o) throws Exception {
                        T data = callable.call();
                        return new Optional<>(data);
                    }
                }).filter(new Predicate<Optional<T>>() {
                    @Override
                    public boolean test(@NonNull Optional<T> optional) throws Exception {
                        return optional.mValue != null;
                    }
                }).map(new Function<Optional<T>, T>() {
                    @Override
                    public T apply(@NonNull Optional<T> optional) throws Exception {
                        return optional.mValue;
                    }
                });
    }

    private static Scheduler sAppToolkitIOScheduler = new Scheduler() {
        @Override
        public Worker createWorker() {
            final AtomicBoolean mDisposed = new AtomicBoolean(false);
            return new Worker() {
                @Override
                public Disposable schedule(@NonNull Runnable run, long delay,
                        @NonNull TimeUnit unit) {
                    DisposableRunnable disposable = new DisposableRunnable(run, mDisposed);
                    ArchTaskExecutor.getInstance().executeOnDiskIO(run);
                    return disposable;
                }

                @Override
                public void dispose() {
                    mDisposed.set(true);
                }

                @Override
                public boolean isDisposed() {
                    return mDisposed.get();
                }
            };
        }
    };

    private static class DisposableRunnable implements Disposable, Runnable {
        private final Runnable mActual;
        private volatile boolean mDisposed = false;
        private final AtomicBoolean mGlobalDisposed;

        DisposableRunnable(Runnable actual, AtomicBoolean globalDisposed) {
            mActual = actual;
            mGlobalDisposed = globalDisposed;
        }

        @Override
        public void dispose() {
            mDisposed = true;
        }

        @Override
        public boolean isDisposed() {
            return mDisposed || mGlobalDisposed.get();
        }

        @Override
        public void run() {
            if (!isDisposed()) {
                mActual.run();
            }
        }
    }

    static class Optional<T> {
        @Nullable
        final T mValue;

        Optional(@Nullable T value) {
            this.mValue = value;
        }
    }
}

@@ -642,45 +642,8 @@
 m;%0A%0A
-import androidx.annotation.Nullable;%0A
 impo
@@ -1030,16 +1030,76 @@
 scribe;%0A
+import io.reactivex.Maybe;%0Aimport io.reactivex.MaybeSource;%0A
 import i
@@ -1332,49 +1332,8 @@
 ion;
-%0Aimport io.reactivex.functions.Predicate;
 %0A%0A/*
@@ -4223,24 +4223,85 @@
 callable) %7B%0A
+        final Maybe%3CT%3E maybe = Maybe.fromCallable(callable);%0A
         retu
@@ -4391,19 +4391,28 @@
        .
-map
+flatMapMaybe
 (new Fun
@@ -4429,668 +4429,114 @@
 ct, 
-Optional%3CT%3E%3E() %7B%0A                    @Override%0A                    public Optional%3CT%3E apply(@NonNull Object o) throws Exception %7B%0A                        T data = callable.call();%0A                        return new Optional%3C%3E(data);%0A                    %7D%0A                %7D).filter(new Predicate%3COptional%3CT%3E%3E() %7B%0A                    @Override%0A                    public boolean test(@NonNull Optional%3CT%3E optional) throws Exception %7B%0A                        return optional.mValue != null;%0A                    %7D%0A                %7D).map(new Function%3COptional%3CT%3E, T%3E() %7B%0A                    @Override%0A                    public T apply(@NonNull Optional%3CT%3E optional
+MaybeSource%3CT%3E%3E() %7B%0A                    @Override%0A                    public MaybeSource%3CT%3E apply(Object o
 ) th
@@ -4587,22 +4587,12 @@
 urn 
-optional.mValu
+mayb
 e;%0A 
@@ -6285,169 +6285,7 @@
    %7D
-%0A%0A    static class Optional%3CT%3E %7B%0A        @Nullable%0A        final T mValue;%0A%0A        Optional(@Nullable T value) %7B%0A            this.mValue = value;%0A        %7D%0A    %7D
 %0A%7D%0A

Use flatMapMaybe to reduce intermediate operator count.