/*
 * Copyright (C) 2018 Yaroslav Mytkalyk
 *
 * 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 com.doctoror.rxcursorloader;

import android.content.ContentResolver;
import android.database.ContentObserver;
import android.database.Cursor;
import android.os.Handler;
import android.os.Looper;
import android.support.annotation.NonNull;
import android.util.Log;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.functions.Action;

import static com.doctoror.rxcursorloader.RxCursorLoader.TAG;
import static com.doctoror.rxcursorloader.RxCursorLoader.isDebugLoggingEnabled;

final class RxCursorLoaderFlowableFactory {

    @NonNull
    static Flowable<Cursor> create(
            @NonNull final ContentResolver resolver,
            @NonNull final RxCursorLoader.Query query,
            @NonNull final Scheduler scheduler,
            @NonNull final BackpressureStrategy backpressureStrategy) {
        //noinspection ConstantConditions
        if (resolver == null) {
            throw new NullPointerException("ContentResolver must not be null");
        }
        //noinspection ConstantConditions
        if (query == null) {
            throw new NullPointerException("Query must not be null");
        }

        final CursorLoaderOnSubscribe onSubscribe = new CursorLoaderOnSubscribe(
                resolver, query, scheduler);

        return Flowable
                .create(onSubscribe, backpressureStrategy)
                .subscribeOn(scheduler)
                .doFinally(new Action() {
                    @Override
                    public void run() {
                        onSubscribe.release();
                    }
                });
    }

    private static final class CursorLoaderOnSubscribe
            implements FlowableOnSubscribe<Cursor> {

        private final Object mLock = new Object();

        @NonNull
        private final ContentResolver mContentResolver;

        @NonNull
        private final RxCursorLoader.Query mQuery;

        @NonNull
        private final Scheduler mScheduler;

        @NonNull
        private final Handler mHandler = new Handler(Looper.getMainLooper());

        private FlowableEmitter<Cursor> mEmitter;

        CursorLoaderOnSubscribe(
                @NonNull final ContentResolver resolver,
                @NonNull final RxCursorLoader.Query query,
                @NonNull final Scheduler scheduler) {
            mContentResolver = resolver;
            mQuery = query;
            mScheduler = scheduler;
        }

        @Override
        public void subscribe(final FlowableEmitter<Cursor> emitter) {
            synchronized (mLock) {
                mEmitter = emitter;
                mContentResolver.registerContentObserver(mQuery.contentUri, true,
                        mContentObserver);
            }
            reload();
        }

        private void release() {
            synchronized (mLock) {
                mContentResolver.unregisterContentObserver(mContentObserver);
                mEmitter = null;
                mHandler.removeCallbacksAndMessages(null);
            }
        }

        /**
         * Loads new {@link Cursor}.
         * <p>
         * This must be called from {@link #subscribe(FlowableEmitter)} thread
         */
        private synchronized void reload() {
            synchronized (mLock) {
                if (isDebugLoggingEnabled()) {
                    Log.d(TAG, mQuery.toString());
                }

                final Cursor c = mContentResolver.query(
                        mQuery.contentUri,
                        mQuery.projection,
                        mQuery.selection,
                        mQuery.selectionArgs,
                        mQuery.sortOrder);

                if (mEmitter != null && !mEmitter.isCancelled()) {
                    if (c != null) {
                        mEmitter.onNext(c);
                    } else {
                        mEmitter.onError(new QueryReturnedNullException());
                    }
                }
            }
        }

        private final ContentObserver mContentObserver = new ContentObserver(mHandler) {

            @Override
            public void onChange(final boolean selfChange) {
                mScheduler.scheduleDirect(mReloadRunnable);
            }
        };

        private final Runnable mReloadRunnable = new Runnable() {
            @Override
            public void run() {
                reload();
            }
        };
    }
}

@@ -2470,16 +2470,23 @@
 Object m
+Emitter
 Lock = n
@@ -3278,32 +3278,39 @@
  synchronized (m
+Emitter
 Lock) %7B%0A        
@@ -3333,32 +3333,42 @@
  = emitter;%0A    
+        %7D%0A
             mCon
@@ -3404,16 +3404,37 @@
 bserver(
+%0A                    
 mQuery.c
@@ -3453,32 +3453,8 @@
 rue,
-%0A                       
  mCo
@@ -3469,30 +3469,16 @@
 erver);%0A
-            %7D%0A
         
@@ -3539,47 +3539,8 @@
 ) %7B%0A
-            synchronized (mLock) %7B%0A    
     
@@ -3625,86 +3625,69 @@
     
-    mEmitter = null;%0A                mHandler.removeCallbacksAndMessages(
+synchronized (mEmitterLock) %7B%0A                mEmitter = 
 null
-)
 ;%0A  
@@ -3913,47 +3913,8 @@
 ) %7B%0A
-            synchronized (mLock) %7B%0A    
     
@@ -3968,20 +3968,16 @@
         
-    
 Log.d(TA
@@ -4007,28 +4007,24 @@
             
-    
 %7D%0A%0A         
@@ -4022,28 +4022,24 @@
             
-  
-  
 final Cursor
@@ -4079,36 +4079,32 @@
                 
-    
 mQuery.contentUr
@@ -4122,28 +4122,24 @@
             
-   
- 
 mQuery.proje
@@ -4157,36 +4157,32 @@
                 
-    
 mQuery.selection
@@ -4175,36 +4175,32 @@
 uery.selection,%0A
-    
                 
@@ -4237,36 +4237,32 @@
                 
-    
 mQuery.sortOrder
@@ -4261,24 +4261,66 @@
 ortOrder);%0A%0A
+            synchronized (mEmitterLock) %7B%0A
             

Synchronize only on Emitter