/*
 * 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.PublishSubject;

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 PublishSubject<Type<?>> commitSubject;
    private final PublishSubject<Type<?>> rollbackSubject;

    TypeChangeListener() {
        commitSubject = PublishSubject.create();
        rollbackSubject = PublishSubject.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);
            }
        };
    }

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

    private void emitTypes(PublishSubject<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);
        }
    }
}

@@ -833,16 +833,82 @@
 Subject;
+%0Aimport rx.subjects.SerializedSubject;%0Aimport rx.subjects.Subject;
 %0A%0Aimport
@@ -1226,39 +1226,51 @@
 ivate final 
-PublishSubject%3C
+SerializedSubject%3CType%3C?%3E, 
 Type%3C?%3E%3E com
@@ -1299,31 +1299,43 @@
 e final 
-PublishSubject%3C
+SerializedSubject%3CType%3C?%3E, 
 Type%3C?%3E%3E
@@ -1396,32 +1396,56 @@
 commitSubject = 
+new SerializedSubject%3C%3E(
 PublishSubject.c
@@ -1439,32 +1439,41 @@
 lishSubject.
+%3CType%3C?%3E%3E
 create()
 ;%0A        ro
@@ -1456,24 +1456,25 @@
 %3C?%3E%3Ecreate()
+)
 ;%0A        ro
@@ -1489,16 +1489,40 @@
 bject = 
+new SerializedSubject%3C%3E(
 PublishS
@@ -1528,24 +1528,33 @@
 Subject.
+%3CType%3C?%3E%3E
 create()
 ;%0A    %7D%0A
@@ -1545,16 +1545,17 @@
 create()
+)
 ;%0A    %7D%0A
@@ -2419,31 +2419,24 @@
  %7D%0A%0A    
-Publish
 Subject%3C
 Type%3C?%3E%3E
@@ -2419,32 +2419,41 @@
  %7D%0A%0A    Subject%3C
+Type%3C?%3E, 
 Type%3C?%3E%3E commitS
@@ -2531,23 +2531,16 @@
 pes(
-Publish
 Subject%3C
 Type
@@ -2531,24 +2531,33 @@
 pes(Subject%3C
+Type%3C?%3E, 
 Type%3C?%3E%3E sub

Use SerializedSubject for type changes