package reactor.pipe.concurrent;

import reactor.fn.tuple.Tuple2;

import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
import reactor.fn.Function;
import reactor.fn.Predicate;
import reactor.fn.UnaryOperator;

/**
 * Generic Atom
 */
public class Atom<T> {

  private AtomicReference<T> ref;

  public Atom(T ref) {
    this.ref = new AtomicReference<T>(ref);
  }

  public T deref() {
    return ref.get();
  }

  public T update(UnaryOperator<T> swapOp) {
    for (; ; ) {
      T old = ref.get();
      T newv = swapOp.apply(old);
      if (ref.compareAndSet(old, newv)) {
        return newv;
      }
    }
  }

  public T updateAndReturnOld(UnaryOperator<T> swapOp) {
    for (; ; ) {
      T old = ref.get();
      T newv = swapOp.apply(old);
      if (ref.compareAndSet(old, newv)) {
        return old;
      }
    }
  }

  public <O> O updateAndReturnOther(Function<T, Tuple2<T, O>> swapOp) {
    for (; ; ) {
      T old = ref.get();
      Tuple2<T, O> newvtuple = swapOp.apply(old);
      if (ref.compareAndSet(old, newvtuple.getT1())) {
        return newvtuple.getT2();
      }
      LockSupport.parkNanos(1L);
    }
  }

  public <O> O updateAndReturnOther(Predicate<T> pred,
                                    Function<T, Tuple2<T, O>> swapOp) {
    for (; ; ) {
      T old = ref.get();
      if (pred.test(old)) {
        Tuple2<T, O> newvtuple = swapOp.apply(old);
        if (ref.compareAndSet(old, newvtuple.getT1())) {
          return newvtuple.getT2();
        }
        LockSupport.parkNanos(1L); //TODO: Maybe park everywhere?
      }
    }
  }

  public T reset(T newv) {
    for (; ; ) {
      T old = ref.get();
      if (ref.compareAndSet(old, newv)) {
        return newv;
      }
    }
  }


}

@@ -27,16 +27,134 @@
 rrent;%0A%0A
+import reactor.fn.Consumer;%0Aimport reactor.fn.Function;%0Aimport reactor.fn.Predicate;%0Aimport reactor.fn.UnaryOperator;%0A
 import r
@@ -280,98 +280,8 @@
 ort;
-%0Aimport reactor.fn.Function;%0Aimport reactor.fn.Predicate;%0Aimport reactor.fn.UnaryOperator;
 %0A%0A/*
@@ -336,16 +336,22 @@
 private 
+final 
 AtomicRe
@@ -370,74 +370,332 @@
 ef;%0A
-%0A  public Atom(T ref) %7B%0A    this.ref = new AtomicReference%3CT%3E(ref)
+  private final Consumer%3CT%3E        newValueConsumer;%0A%0A  public Atom(T ref) %7B%0A    this(ref, null);%0A  %7D%0A  // TODO: make atom an intereace, extract consumer impl elsewhere%0A  public Atom(T ref,%0A              Consumer%3CT%3E newValueConsumer) %7B%0A    this.ref = new AtomicReference%3CT%3E(ref);%0A    this.newValueConsumer = newValueConsumer
 ;%0A  
@@ -900,32 +900,144 @@
 t(old, newv)) %7B%0A
+        if (newValueConsumer != null && !newv.equals(old)) %7B%0A          newValueConsumer.accept(newv);%0A        %7D%0A
         return n

Add callback to atom