/*
 * Copyright 2016 requery.io
 *
 * 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 io.requery.rx;

import io.requery.TransactionIsolation;
import io.requery.TransactionListener;
import io.requery.meta.Type;
import io.requery.proxy.EntityProxy;
import io.requery.util.function.Supplier;
import rx.subjects.BehaviorSubject;

import java.util.LinkedHashSet;
import java.util.Set;

/**
 * {@link TransactionListener} for listening to entity commits and emitting the types that have
 * changed through an Rx {@link rx.subjects.Subject}.
 *
 * @author Nikhil Purushe
 */
class TypeChangeListener implements Supplier<TransactionListener> {

    private final BehaviorSubject<Type<?>> commitSubject;
    private final BehaviorSubject<Type<?>> rollbackSubject;

    TypeChangeListener() {
        commitSubject = BehaviorSubject.create();
        rollbackSubject = BehaviorSubject.create();
    }

    @Override
    public TransactionListener get() {
        return new TransactionListener() {
            @Override
            public void beforeBegin(TransactionIsolation isolation) {
            }

            @Override
            public void afterBegin(TransactionIsolation isolation) {
            }

            @Override
            public void beforeCommit(Set<EntityProxy<?>> entities) {

            }

            @Override
            public void afterCommit(Set<EntityProxy<?>> entities) {
                emitTypes(commitSubject, entities);
            }

            @Override
            public void beforeRollback(Set<EntityProxy<?>> entities) {

            }

            @Override
            public void afterRollback(Set<EntityProxy<?>> entities) {
                emitTypes(rollbackSubject, entities);
            }
        };
    }

    BehaviorSubject<Type<?>> commitSubject() {
        return commitSubject;
    }

    private void emitTypes(BehaviorSubject<Type<?>> subject, Set<EntityProxy<?>> entities) {
        Set<Type<?>> types = new LinkedHashSet<>();
        for (EntityProxy proxy : entities) {
            types.add(proxy.type());
        }
        for (Type type : types) {
            subject.onNext(type);
        }
    }
}

@@ -818,24 +818,23 @@
 ubjects.
-Behavior
+Publish
 Subject;
@@ -1160,32 +1160,31 @@
 ivate final 
-Behavior
+Publish
 Subject%3CType
@@ -1221,24 +1221,23 @@
 e final 
-Behavior
+Publish
 Subject%3C
@@ -1310,32 +1310,31 @@
 itSubject = 
-Behavior
+Publish
 Subject.crea
@@ -1365,24 +1365,23 @@
 bject = 
-Behavior
+Publish
 Subject.
@@ -2261,24 +2261,23 @@
  %7D%0A%0A    
-Behavior
+Publish
 Subject%3C
@@ -2371,16 +2371,15 @@
 pes(
-Behavior
+Publish
 Subj

Change TypeChangeListener from BehaviorSubject to PublishSubject