VersionedValue: never refuse or drop an owner’s change

src/org/openlcb/implementations/BitProducerConsumer.java

@@ -64,7 +64,11 @@ public BitProducerConsumer(OlcbInterface iface, EventID eventOn, EventID eventOf
         this.eventOff = eventOff;
         this.flags = flags;
         value = new VersionedValue<>((flags & DEFAULT_TRUE) != 0);
-        valueListener = new VersionedValueListener<Boolean>(value) {
+        // Reports every change made by an owner, even one that a value from the network (which
+        // this listener sets) has overtaken before the report was made. The report is then
+        // sent after that network message, and when it comes back to us it sets the value to
+        // match the order of the messages on the bus.
+        valueListener = new VersionedValueListener<Boolean>(value, true) {
             @Override
             public void update(Boolean newValue) {
                 if ((flags & IS_PRODUCER) == 0) return;

src/org/openlcb/implementations/VersionedValue.java

@@ -14,6 +14,38 @@ public class VersionedValue<T> {
     private int defaultVersion = DEFAULT_VERSION;
     private final static Logger log = Logger.getLogger(VersionedValue.class.getCanonicalName());
 
+    /**
+     * Notification of one change of the value, sent to the property change listeners. Besides
+     * the old and new value it carries the version the change was made at, and the listener
+     * that made it.
+     */
+    public static final class Change extends java.beans.PropertyChangeEvent {
+        private final int version;
+        private final transient Object origin;
+
+        Change(VersionedValue<?> source, Object oldValue, Object newValue, int version,
+               Object origin) {
+            super(source, "updated", oldValue, newValue);
+            this.version = version;
+            this.origin = origin;
+        }
+
+        /**
+         * @return the version this change was made at.
+         */
+        public int getVersion() {
+            return version;
+        }
+
+        /**
+         * @return the VersionedValueListener that made this change, or null if it was not made
+         * through a listener.
+         */
+        public Object getOrigin() {
+            return origin;
+        }
+    }
+
     public VersionedValue(T t) {
         version = DEFAULT_VERSION;
         nextVersion = DEFAULT_VERSION + 1;
@@ -49,14 +81,18 @@ public synchronized boolean isVersionAtDefault() {
 
     /**
      * Sets the current value; possibly calling listeners. Handles versioning internally.
+     * <p>
+     * The new version is taken and the value stored as one step, so a set made at the same time
+     * on another thread cannot take a later version, be stored first, and cause this set to be
+     * refused as out of date.
      * @param t new value of data stored.
      */
     public void set(T t) {
-        int version;
-        synchronized(this) {
-            version = getNewVersion();
+        Change change;
+        synchronized (this) {
+            change = store(getNewVersion(), t, false, null);
         }
-        set(version, t);
+        fire(change);
     }
 
     /**
@@ -84,28 +120,52 @@ public boolean set(int atVersion, T t) {
     }
 
     private boolean setInternal(int atVersion, T t, boolean forceNotify) {
-        T old;
-        boolean updated = false;
+        Change change;
         synchronized (this) {
             if (atVersion <= version) return false;
-            int oldVersion = version;
-            version = atVersion;
-            if (nextVersion <= atVersion) {
-                nextVersion = atVersion + 1;
-            }
-            if (data.equals(t) && oldVersion != defaultVersion && !forceNotify) {
-                return true;
-            }
-            old = data;
-            if (oldVersion == defaultVersion || forceNotify) {
-                old = null;
-            }
-            data = t;
+            change = store(atVersion, t, forceNotify, null);
         }
-        firePropertyChange("updated", old, t);
+        fire(change);
         return true;
     }
 
+    /**
+     * Stores a value at a version newer than the current one. The caller must hold the lock of
+     * this object, and must pass the result to {@link #fire} once it has released the lock.
+     * @param atVersion    new version number, greater than the current version
+     * @param t            new value of data stored
+     * @param forceNotify  notify the listeners even if the value has not changed
+     * @param origin       listener making the change, or null
+     * @return the notification to send, or null if there is none
+     */
+    Change store(int atVersion, T t, boolean forceNotify, Object origin) {
+        int oldVersion = version;
+        version = atVersion;
+        if (nextVersion <= atVersion) {
+            nextVersion = atVersion + 1;
+        }
+        if (data.equals(t) && oldVersion != defaultVersion && !forceNotify) {
+            return null;
+        }
+        T old = data;
+        if (oldVersion == defaultVersion || forceNotify) {
+            old = null;
+        }
+        data = t;
+        return new Change(this, old, t, atVersion, origin);
+    }
+
+    /**
+     * Sends a notification returned by {@link #store} to the listeners. Must be called without
+     * holding the lock of this object.
+     * @param change notification to send, or null for none
+     */
+    void fire(Change change) {
+        if (change != null) {
+            pcs.firePropertyChange(change);
+        }
+    }
+
     /**
      * @return data stored at the current version.
      */

src/org/openlcb/implementations/VersionedValueListener.java

@@ -11,9 +11,34 @@
 public abstract class VersionedValueListener<T> implements PropertyChangeListener {
     int ownerVersion;
     protected VersionedValue<T> parent;
+    private final boolean everyChange;
 
+    /**
+     * Creates a listener that is told the latest value whenever someone else has set a value
+     * newer than any this listener has set or been told of.
+     * @param parent value to listen to
+     */
     public VersionedValueListener(VersionedValue<T> parent) {
+        this(parent, false);
+    }
+
+    /**
+     * Creates a listener.
+     * @param parent       value to listen to
+     * @param everyChange  if false, as for {@link #VersionedValueListener(VersionedValue)},
+     *                     update() is called with the latest value whenever someone else has set
+     *                     a value newer than any this listener has set or been told of, so a
+     *                     change already overtaken by a newer one is not reported. If true,
+     *                     update() is called for every change made by someone else, with the
+     *                     value of that change, even if this listener has set a newer value
+     *                     since. This suits a listener that sends the owners' changes out to the
+     *                     network and also sets the values it receives from the network: a value
+     *                     received from the network must not stop an owner's earlier change from
+     *                     being sent.
+     */
+    public VersionedValueListener(VersionedValue<T> parent, boolean everyChange) {
         this.parent = parent;
+        this.everyChange = everyChange;
         parent.addPropertyChangeListener(this);
         ownerVersion = parent.getVersion();
     }
@@ -24,18 +49,26 @@ public void release() {
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     public void propertyChange(PropertyChangeEvent propertyChangeEvent) {
-        int pVersion = parent.getVersion();
-        boolean isUpdated = false;
-        synchronized (this) {
-            if (pVersion > ownerVersion) {
-                ownerVersion = pVersion;
-                isUpdated = true;
+        if (everyChange && propertyChangeEvent instanceof VersionedValue.Change) {
+            VersionedValue.Change change = (VersionedValue.Change) propertyChangeEvent;
+            if (change.getOrigin() != this) {
+                update((T) change.getNewValue());
             }
+            return;
         }
-        if (isUpdated) {
-            update(parent.getLatestData());
+        T data;
+        synchronized (this) {
+            synchronized (parent) {
+                if (parent.version <= ownerVersion) {
+                    return;
+                }
+                ownerVersion = parent.version;
+                data = parent.data;
+            }
         }
+        update(data);
     }
 
     /**
@@ -44,12 +77,7 @@ public void propertyChange(PropertyChangeEvent propertyChangeEvent) {
      * @param t new value
      */
     public void setFromOwner(T t) {
-        int newVersion;
-        synchronized (this) {
-            newVersion = parent.getNewVersion();
-            ownerVersion = newVersion;
-        }
-        parent.set(newVersion, t);
+        setFromOwner(t, false);
     }
 
     /**
@@ -58,12 +86,25 @@ public void setFromOwner(T t) {
      * @param t new value
      */
     public void setFromOwnerWithForceNotify(T t) {
-        int newVersion;
+        setFromOwner(t, true);
+    }
+
+    /**
+     * Takes a new version and stores the value at it as one step, so that a set made at the same
+     * time on another thread cannot take a later version, be stored first, and cause this set
+     * to be refused as out of date.
+     * @param t            new value
+     * @param forceNotify  notify the other listeners even if the value has not changed
+     */
+    private void setFromOwner(T t, boolean forceNotify) {
+        VersionedValue.Change change;
         synchronized (this) {
-            newVersion = parent.getNewVersion();
-            ownerVersion = newVersion;
+            synchronized (parent) {
+                ownerVersion = parent.getNewVersion();
+                change = parent.store(ownerVersion, t, forceNotify, this);
+            }
         }
-        parent.setWithForceNotify(newVersion, t);
+        parent.fire(change);
     }
 
     /** Calls the updater with the latest data. */

test/org/openlcb/implementations/BitProducerConsumerTest.java

@@ -8,7 +8,12 @@
 
 import org.openlcb.EventID;
 import org.openlcb.NodeID;
+import org.openlcb.OlcbInterface;
+import org.mockito.InOrder;
 
+import java.util.concurrent.CountDownLatch;
+
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -620,6 +625,100 @@ public void testResetToDefault() {
                 ":X194C5333N0504030201000708;");
     }
 
+    /**
+     * Holds back the delivery of outgoing messages, both to the bus and to the local
+     * listeners, until release() is called. This lets a test act while messages are still in
+     * the send queue.
+     */
+    private static class HeldLoopback implements OlcbInterface.SyncExecutor {
+        private final CountDownLatch released = new CountDownLatch(1);
+
+        @Override
+        public void schedule(Runnable r) throws InterruptedException {
+            released.await();
+            r.run();
+        }
+
+        void release() {
+            released.countDown();
+        }
+    }
+
+    /**
+     * The owner turns the bit on, and while that report is still in the send queue a report
+     * from another node turning it off is heard. Our report goes on the bus after the other
+     * node's, so every node watching the bus ends with the bit on, and so must we.
+     */
+    @Test
+    public void testFollowsBusOrderWhenRemoteReportHeardBeforeOursIsSent() throws Exception {
+        createWithDefaults();
+        MockVersionedValueListener<Boolean> owner = new MockVersionedValueListener<>(pc
+                .getValue());
+        HeldLoopback loopback = new HeldLoopback();
+        iface.setLoopbackThread(loopback);
+
+        owner.setFromOwner(true);
+        sendFrame(":X195B4444N0504030201000709;");
+        Assert.assertFalse(pc.getValue().getLatestData());
+
+        loopback.release();
+        expectFrame(":X195B4333N0504030201000708;");
+        expectNoFrames();
+
+        Assert.assertTrue(pc.getValue().getLatestData());
+        InOrder order = inOrder(owner.stub);
+        order.verify(owner.stub).update(false);
+        order.verify(owner.stub).update(true);
+        verifyNoMoreInteractions(owner.stub);
+    }
+
+    /**
+     * The owner turns the bit on and straight back off while the first report is still in the
+     * send queue. Both reports are sent, and as each comes back it sets the value it carried,
+     * so the bit follows the order of the messages on the bus and ends off.
+     */
+    @Test
+    public void testOwnReportsSetValueInBusOrder() throws Exception {
+        createWithDefaults();
+        MockVersionedValueListener<Boolean> owner = new MockVersionedValueListener<>(pc
+                .getValue());
+        HeldLoopback loopback = new HeldLoopback();
+        iface.setLoopbackThread(loopback);
+
+        owner.setFromOwner(true);
+        owner.setFromOwner(false);
+        loopback.release();
+        expectFrame(":X195B4333N0504030201000708;");
+        expectFrame(":X195B4333N0504030201000709;");
+        expectNoFrames();
+
+        Assert.assertFalse(pc.getValue().getLatestData());
+        InOrder order = inOrder(owner.stub);
+        order.verify(owner.stub).update(true);
+        order.verify(owner.stub).update(false);
+        verifyNoMoreInteractions(owner.stub);
+    }
+
+    @Test
+    public void testOtherLocalObjectSeesEventReports() throws Exception {
+        createWithDefaults();
+        BitProducerConsumer other = new BitProducerConsumer(iface, onEvent, offEvent,
+                BitProducerConsumer.IS_CONSUMER);
+        expectFrame(":X194C7333N0504030201000708;");
+        expectFrame(":X194C7333N0504030201000709;");
+        expectNoFrames();
+
+        // Messages from one object still reach the other objects on the same interface.
+        pc.getValue().set(true);
+        expectFrame(":X195B4333N0504030201000708;");
+        Assert.assertTrue(other.getValue().getLatestData());
+
+        pc.getValue().set(false);
+        expectFrame(":X195B4333N0504030201000709;");
+        Assert.assertFalse(other.getValue().getLatestData());
+        other.release();
+    }
+
     private void createWithDefaults() {
         pc = new BitProducerConsumer(iface, onEvent, offEvent, false);
         expectFrame(":X19547333N0504030201000708;", times(1));

test/org/openlcb/implementations/MockVersionedValueListener.java

@@ -10,6 +10,10 @@ public MockVersionedValueListener(VersionedValue<T> parent) {
         super(parent);
     }
 
+    public MockVersionedValueListener(VersionedValue<T> parent, boolean everyChange) {
+        super(parent, everyChange);
+    }
+
     public class UpdaterStub {
         public void update(T t) {}
     }