Created ReactiveAtomView. Small tweaks to ReactiveAtom and CallbackSet.

src/clojure/freactive/core.clj

@@ -1,7 +1,7 @@
 (ns freactive.core
   (:refer-clojure
    :exclude [atom agent ref swap! reset! compare-and-set!])
-  (:import [clojure.lang ReactiveAtom Reactive StatefulReactive]))
+  (:import [clojure.lang ReactiveAtom Reactive StatefulReactive ReactiveAtomView]))
 
 ;; Copying clojure.core atom stuff here so that we can use my ReactiveAtom class.
 
@@ -69,6 +69,12 @@ current value. Returns newval."
 (defn reactive-state [init-state f & options]
   (#'clojure.core/setup-reference (StatefulReactive. init-state f) options))
 
+(defn reactive-atom-view
+  ([ratom view-transform]
+   (reactive-atom-view view-transform (fn [x _] x)))
+  ([ratom view-transform update-transform]
+   (ReactiveAtomView. ratom view-transform update-transform)))
+
 ;; (import '(java.util TimerTask Timer))
 
 ;; (defn test1 []

src/java/clojure/lang/CallbackSet.java

@@ -120,4 +120,8 @@ public class CallbackSet {
     public void invokeAndRemoveAll(Object arg1, Object arg2) {
         invokeAll(takeAll(), arg1, arg2);
     }
+
+    public IPersistentMap getCallbacks() {
+        return (IPersistentMap)callbacks.deref();
+    }
 }

src/java/clojure/lang/ReactiveAtom.java

@@ -110,12 +110,12 @@ public Object reset(Object newval){
 }
     
 public void notifyWatches(Object oldVal, Object newVal){
-        if(oldVal != newVal)
+        if(!oldVal.equals(newVal))
                 super.notifyWatches(oldVal, newVal);
 }
 
 @Override
-public IInvalidates addInvalidationWatch(Object key, IFn callback) {
+public synchronized IInvalidates addInvalidationWatch(Object key, IFn callback) {
     addWatch(key, new AFn() {
         @Override
         public Object invoke(Object key, Object ref, Object oldV, Object newV) {
@@ -126,7 +126,7 @@ public IInvalidates addInvalidationWatch(Object key, IFn callback) {
 }
 
 @Override
-public IInvalidates removeInvalidationWatch(Object key) {
+public synchronized IInvalidates removeInvalidationWatch(Object key) {
     removeWatch(key);
     return this;
 }

src/java/clojure/lang/ReactiveAtomView.java

@@ -0,0 +1,154 @@
+package clojure.lang;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class ReactiveAtomView implements IReactiveAtom {
+    private final IReactiveAtom source;
+    private volatile Object curView;
+    private volatile Object cur;
+    private final CallbackSet invalidationWatches = new CallbackSet(this);
+    private final CallbackSet watches = new CallbackSet(this);
+    private final boolean lazy = false;
+    private final AtomicBoolean dirty = new AtomicBoolean(true);
+    private final IFn viewTransform;
+    private final IFn updateTransform;
+
+    public ReactiveAtomView(IReactiveAtom source, IFn viewTransform,
+                            IFn updateTransform) {
+        this.source = source;
+        this.viewTransform = viewTransform;
+        this.updateTransform = updateTransform;
+        source.addInvalidationWatch(this, new AFn() {
+            @Override
+            public Object invoke(Object arg1, Object arg2) {
+                if(lazy) {
+                    if(dirty.compareAndSet(false, true)) {
+                        invalidationWatches.invokeAll();
+                    }
+                } else {
+                    cur = source.deref();
+                    Object newView = viewTransform.invoke(cur);
+                    if(!newView.equals(curView)) {
+                        curView = newView;
+                        invalidationWatches.invokeAll();
+                    }
+                }
+                return null;
+            }
+        });
+        source.addWatch(this, new AFn() {
+            @Override
+            public Object invoke(Object k, Object r, Object o, Object n) {
+                cur = n;
+                Object newView = viewTransform.invoke(cur);
+                if(!newView.equals(curView)) {
+                    Object oldView = curView;
+                    curView = newView;
+                    watches.invokeAll(oldView, newView);
+                }
+                return null;
+            }
+        });
+    }
+
+    @Override
+    public Object swap(IFn f) {
+        return source.swap(new AFn() {
+            @Override
+            public Object invoke(Object cur) {
+                return updateTransform.invoke(cur,
+                        f.invoke(viewTransform.invoke(cur)));
+            }
+        });
+    }
+
+    @Override
+    public Object swap(IFn f, Object arg) {
+        return source.swap(new AFn() {
+            @Override
+            public Object invoke(Object cur) {
+                return updateTransform.invoke(cur,
+                        f.invoke(viewTransform.invoke(cur), arg));
+            }
+        });
+    }
+
+    @Override
+    public Object swap(IFn f, Object arg1, Object arg2) {
+        return source.swap(new AFn() {
+            @Override
+            public Object invoke(Object cur) {
+                return updateTransform.invoke(cur,
+                        f.invoke(viewTransform.invoke (cur), arg1, arg2));
+            }
+        });
+    }
+
+    @Override
+    public Object swap(IFn f, Object x, Object y, ISeq args) {
+        return source.swap(new AFn() {
+            @Override
+            public Object invoke(Object cur) {
+                return updateTransform.invoke(
+                        f.applyTo(
+                                RT.listStar(viewTransform.invoke(cur), x, y, args)));
+            }
+        });
+    }
+
+    @Override
+    public boolean compareAndSet(Object oldv, Object newv) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Object reset(Object newval) {
+        return source.reset(updateTransform.invoke(newval));
+    }
+
+    @Override
+    public IInvalidates addInvalidationWatch(Object key, IFn callback) {
+        source.addInvalidationWatch(key, callback);
+        return this;
+    }
+
+    @Override
+    public IInvalidates removeInvalidationWatch(Object key) {
+        source.removeInvalidationWatch(key);
+        return this;
+    }
+
+    @Override
+    public void setValidator(IFn iFn) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public IFn getValidator() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public IPersistentMap getWatches() {
+        return watches.getCallbacks();
+    }
+
+    @Override
+    public IRef addWatch(Object key, IFn iFn) {
+        watches.add(key, iFn);
+        return this;
+    }
+
+    @Override
+    public IRef removeWatch(Object key) {
+        watches.remove(key);
+        return this;
+    }
+
+    @Override
+    public Object deref() {
+        if(dirty.get())
+            curView = viewTransform.invoke(source.deref());
+        return curView;
+    }
+}