/*
 * Copyright 2015 Maxim Tuev.
 *
 * 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.agna.ferro.rx;


import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func2;
import rx.observers.SerializedSubscriber;

/**
 * This operator freezes all rx events (onNext, onError, onComplete) when freeze selector emits true,
 * and unfreeze it after freeze selector emits false.
 * If freeze selector does not emit any elements, all events would be frozen
 * If you want reduce num of elements in freeze buffer, you can define replaceFrozenEventPredicate.
 * When Observable frozen and source observable emits normal (onNext) event, before it is added to
 * the end of buffer, it compare with all already buffered events using replaceFrozenEventPredicate,
 * and if replaceFrozenEventPredicate return true, buffered element would be removed.
 */

public class OperatorFreeze<T> implements Observable.Operator<T, T> {

    private final Observable<Boolean> freezeSelector;
    private final Func2<T, T, Boolean> replaceFrozenEventPredicate;

    public OperatorFreeze(Observable<Boolean> freezeSelector,
                          Func2<T, T, Boolean> replaceFrozenEventPredicate) {
        this.freezeSelector = freezeSelector;
        this.replaceFrozenEventPredicate = replaceFrozenEventPredicate;
    }

    public OperatorFreeze(Observable<Boolean> freezeSelector) {
        this(freezeSelector, new Func2<T, T, Boolean>() {
            @Override
            public Boolean call(T frozenEvent, T newEvent) {
                return false;
            }
        });

    }


    @Override
    public Subscriber<? super T> call(Subscriber<? super T> child) {

        final FreezeSubscriber<T> freezeSubscriber = new FreezeSubscriber<>(
                new SerializedSubscriber<>(child),
                replaceFrozenEventPredicate);

        final Subscriber<Boolean> freezeSelectorSubscriber = new Subscriber<Boolean>() {
            @Override
            public void onCompleted() {
                freezeSubscriber.forceOnComplete();
            }

            @Override
            public void onError(Throwable e) {
                freezeSubscriber.forceOnError(e);
            }

            @Override
            public void onNext(Boolean freeze) {
                freezeSubscriber.setFrozen(freeze);
            }
        };
        child.add(freezeSubscriber);
        child.add(freezeSelectorSubscriber);
        freezeSelector.unsafeSubscribe(freezeSelectorSubscriber);

        return freezeSubscriber;

    }


    private static final class FreezeSubscriber<T> extends Subscriber<T> {

        private final Subscriber<T> child;
        private final Func2<T, T, Boolean> replaceFrozenEventPredicate;
        private final List<T> frozenEventsBuffer = new LinkedList<>();

        private boolean frozen = true;
        private boolean done = false;
        private Throwable error = null;

        public FreezeSubscriber(Subscriber<T> child, Func2<T, T, Boolean> replaceFrozenEventPredicate) {
            this.child = child;
            this.replaceFrozenEventPredicate = replaceFrozenEventPredicate;
        }

        @Override
        public void onCompleted() {
            if (error != null) {
                return;
            }
            synchronized (this) {
                if (frozen) {
                    done = true;
                } else {
                    child.onCompleted();
                    unsubscribe();
                }
            }
        }

        @Override
        public void onError(Throwable e) {
            if (done) {
                return;
            }
            synchronized (this) {
                if (frozen) {
                    error = e;
                } else {
                    child.onError(e);
                    unsubscribe();
                }
            }
        }

        @Override
        public void onNext(T event) {
            if (done || error != null) {
                return;
            }
            synchronized (this) {
                if (frozen) {
                    bufferEvent(event);
                } else {
                    child.onNext(event);
                }
            }
        }

        private void bufferEvent(T event) {
            for(ListIterator<T> it = frozenEventsBuffer.listIterator(); it.hasNext(); ){
                T frozenEvent = it.next();
                if(replaceFrozenEventPredicate.call(frozenEvent, event)){
                    it.remove();
                }
            }
            frozenEventsBuffer.add(event);
        }

        public void forceOnComplete() {
            child.onCompleted();
            unsubscribe();
        }

        public void forceOnError(Throwable e) {
            child.onError(e);
            unsubscribe();
        }

        public synchronized void setFrozen(boolean frozen) {
            this.frozen = frozen;
            if (!frozen) {
                emitFrozenEvents();
                if (done) {
                    forceOnComplete();
                }
                if (error != null) {
                    forceOnError(error);
                }
            }
        }

        private void emitFrozenEvents() {
            for (T event : frozenEventsBuffer) {
                child.onNext(event);
            }
            frozenEventsBuffer.clear();
        }
    }
}

@@ -744,16 +744,49 @@
 criber;%0A
+import rx.exceptions.Exceptions;%0A
 import r
@@ -1470,16 +1470,89 @@
 emoved.%0A
+ *%0A * Observable after this operator can emit event in different threads%0A
  */%0A%0Apub
@@ -3894,32 +3894,40 @@
             if (
+done %7C%7C 
 error != null) %7B
@@ -4279,32 +4279,49 @@
         if (done
+ %7C%7C error != null
 ) %7B%0A            
@@ -4989,16 +4989,17 @@
      for
+ 
 (ListIte
@@ -5062,16 +5062,17 @@
 ext(); )
+ 
 %7B%0A      
@@ -5104,24 +5104,50 @@
  it.next();%0A
+                try %7B%0A    
             
@@ -5152,16 +5152,17 @@
       if
+ 
 (replace
@@ -5207,16 +5207,17 @@
  event))
+ 
 %7B%0A      
@@ -5230,16 +5230,20 @@
         
+ 
+   
 it.remov
@@ -5243,24 +5243,232 @@
 t.remove();%0A
+                    %7D%0A                %7D catch (Throwable ex) %7B%0A                    Exceptions.throwIfFatal(ex);%0A                    unsubscribe();%0A                    onError(ex);%0A                    return;%0A
             
@@ -5946,20 +5946,29 @@
     if (
-done
+error != null
 ) %7B%0A    
@@ -5986,33 +5986,35 @@
      forceOn
-Complete(
+Error(error
 );%0A         
@@ -6038,37 +6038,28 @@
         if (
-error != null
+done
 ) %7B%0A        
@@ -6073,35 +6073,33 @@
      forceOn
-Error(error
+Complete(
 );%0A         

fix OperatorFreeze