/*
 * Copyright (c) 2015 FUJI Goro (gfx).
 *
 * 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.github.gfx.android.orma.event;

import com.github.gfx.android.orma.Schema;
import com.github.gfx.android.orma.Selector;
import com.github.gfx.android.orma.annotation.Experimental;

import com.github.gfx.android.orma.core.Database;
import android.support.annotation.RestrictTo;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.WeakHashMap;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.subjects.PublishSubject;

/**
 * Helper class for query observables. This class is NOT thread-safe.
 */
@RestrictTo(RestrictTo.Scope.LIBRARY)
@Experimental
public class DataSetChangedTrigger {

    final WeakHashMap<Observer<DataSetChangedEvent<?>>, Selector<?, ?>> observerMap = new WeakHashMap<>();

    Set<Schema<?>> changedDataSetInTransaction = null;

    public <S extends Selector<?, ?>> Observable<DataSetChangedEvent<S>> create(S selector) {
        PublishSubject<DataSetChangedEvent<S>> subject = PublishSubject.create();
        register(subject, selector);
        return subject;
    }

    @SuppressWarnings("unchecked")
    public <S extends Selector<?, ?>> void register(Observer<DataSetChangedEvent<S>> observer, Selector<?, ?> selector) {
        observerMap.put((Observer<DataSetChangedEvent<?>>)(Object)observer, selector);
    }

    public <Model> void fire(Database db, DataSetChangedEvent.Type type, Schema<Model> schema) {
        if (observerMap.isEmpty()) {
            return;
        }
        if (db.inTransaction()) {
            addChangedDataSetInTransaction(schema);
            return;
        }

        for (Map.Entry<Observer<DataSetChangedEvent<?>>, Selector<?, ?>> entry : observerMap.entrySet()) {
            Selector<?, ?> selector = entry.getValue();
            if (schema == selector.getSchema()) {
                Observer<DataSetChangedEvent<?>> observer = entry.getKey();
                observer.onNext(new DataSetChangedEvent<>(type, selector));
            }
        }
    }

    private void addChangedDataSetInTransaction(Schema<?> schema) {
        if (changedDataSetInTransaction == null) {
            changedDataSetInTransaction = new HashSet<>();
        }

        changedDataSetInTransaction.add(schema);
    }

    public void fireForTransaction() {
        Set<Schema<?>> schemaSet = changedDataSetInTransaction;
        changedDataSetInTransaction = null;

        if (schemaSet == null) {
            return;
        }
        for (Map.Entry<Observer<DataSetChangedEvent<?>>, Selector<?, ?>> entry : observerMap.entrySet()) {
            Selector<?, ?> selector = entry.getValue();
            if (schemaSet.contains(selector.getSchema())) {
                Observer<DataSetChangedEvent<?>> observer = entry.getKey();
                observer.onNext(new DataSetChangedEvent<>(DataSetChangedEvent.Type.TRANSACTION, selector));
            }
        }
    }
}

@@ -786,17 +786,16 @@
 mental;%0A
-%0A
 import c
@@ -836,16 +836,17 @@
 tabase;%0A
+%0A
 import a
@@ -884,16 +884,52 @@
 ictTo;%0A%0A
+import java.lang.ref.WeakReference;%0A
 import j
@@ -1007,20 +1007,37 @@
 va.util.
-Weak
+concurrent.Concurrent
 HashMap;
@@ -1317,30 +1317,85 @@
 r %7B%0A
-%0A    final WeakHashMap
+    // TODO: use well-tested concurrent weak maps%0A    final Map%3CWeakReference
 %3CObs
@@ -1415,32 +1415,33 @@
 ChangedEvent%3C?%3E%3E
+%3E
 , Selector%3C?, ?%3E
@@ -1460,20 +1460,26 @@
 p = new 
-Weak
+Concurrent
 HashMap%3C
@@ -1965,16 +1965,36 @@
 Map.put(
+new WeakReference%3C%3E(
 (Observe
@@ -2035,16 +2035,17 @@
 observer
+)
 , select
@@ -2352,32 +2352,46 @@
  for (Map.Entry%3C
+WeakReference%3C
 Observer%3CDataSet
@@ -2398,32 +2398,33 @@
 ChangedEvent%3C?%3E%3E
+%3E
 , Selector%3C?, ?%3E
@@ -2573,32 +2573,46 @@
                 
+WeakReference%3C
 Observer%3CDataSet
@@ -2619,32 +2619,33 @@
 ChangedEvent%3C?%3E%3E
+%3E
  observer = entr
@@ -2629,32 +2629,35 @@
 nt%3C?%3E%3E%3E observer
+Ref
  = entry.getKey(
@@ -2671,32 +2671,36 @@
             
+if (
 observer
 .onNext(new 
@@ -2691,59 +2691,213 @@
 rver
-.onNext(new DataSetChangedEvent%3C%3E(type, selector));
+Ref.get() != null) %7B%0A                    observerRef.get().onNext(new DataSetChangedEvent%3C%3E(type, selector));%0A                %7D else %7B%0A                    observerMap.remove(observerRef);%0A                %7D
 %0A   
@@ -3403,16 +3403,30 @@
 p.Entry%3C
+WeakReference%3C
 Observer
@@ -3449,16 +3449,17 @@
 vent%3C?%3E%3E
+%3E
 , Select
@@ -3634,16 +3634,30 @@
         
+WeakReference%3C
 Observer
@@ -3680,16 +3680,17 @@
 vent%3C?%3E%3E
+%3E
  observe
@@ -3690,16 +3690,19 @@
 observer
+Ref
  = entry
@@ -3728,24 +3728,28 @@
         
+if (
 observer
 .onNext(
@@ -3740,16 +3740,74 @@
 observer
+Ref.get() != null) %7B%0A                    observerRef.get()
 .onNext(
@@ -3874,32 +3874,128 @@
 ON, selector));%0A
+                %7D else %7B%0A                    observerMap.remove(observerRef);%0A                %7D%0A
             %7D%0A  

use ConcurrentHashMap+WeakReference instead of WeakHashMap