TransactionBroadcast: invoke the progress listener if the broadcast already started, to avoid people accidentally writing races when using the PeerGroup convenience APIs.

core/src/main/java/org/bitcoinj/core/TransactionBroadcast.java

@@ -165,6 +165,9 @@ public class TransactionBroadcast {
         }
     }
 
+    private int numSeemPeers;
+    private boolean mined;
+
     private class ConfidenceChange implements TransactionConfidence.Listener {
         @Override
         public void onConfidenceChanged(TransactionConfidence conf, ChangeReason reason) {
@@ -175,7 +178,7 @@ public class TransactionBroadcast {
                     numSeenPeers, mined ? " and mined" : "");
 
             // Progress callback on the requested thread.
-            invokeProgressCallback(numSeenPeers, mined);
+            invokeAndRecord(numSeenPeers, mined);
 
             if (numSeenPeers >= numWaitingFor || mined) {
                 // We've seen the min required number of peers announce the transaction, or it was included
@@ -199,6 +202,14 @@ public class TransactionBroadcast {
         }
     }
 
+    private void invokeAndRecord(int numSeenPeers, boolean mined) {
+        synchronized (this) {
+            this.numSeemPeers = numSeenPeers;
+            this.mined = mined;
+        }
+        invokeProgressCallback(numSeenPeers, mined);
+    }
+
     private void invokeProgressCallback(int numSeenPeers, boolean mined) {
         final ProgressCallback callback;
         Executor executor;
@@ -243,7 +254,8 @@ public class TransactionBroadcast {
 
     /**
      * Sets the given callback for receiving progress values, which will run on the user thread. See
-     * {@link org.bitcoinj.utils.Threading} for details.
+     * {@link org.bitcoinj.utils.Threading} for details.  If the broadcast has already started then the callback will
+     * be invoked immediately with the current progress.
      */
     public void setProgressCallback(ProgressCallback callback) {
         setProgressCallback(callback, Threading.USER_THREAD);
@@ -252,10 +264,21 @@ public class TransactionBroadcast {
     /**
      * Sets the given callback for receiving progress values, which will run on the given executor. If the executor
      * is null then the callback will run on a network thread and may be invoked multiple times in parallel. You
-     * probably want to provide your UI thread or Threading.USER_THREAD for the second parameter.
+     * probably want to provide your UI thread or Threading.USER_THREAD for the second parameter. If the broadcast
+     * has already started then the callback will be invoked immediately with the current progress.
      */
-    public synchronized void setProgressCallback(ProgressCallback callback, @Nullable Executor executor) {
+    public void setProgressCallback(ProgressCallback callback, @Nullable Executor executor) {
+        boolean shouldInvoke;
+        int num;
+        boolean mined;
+        synchronized (this) {
             this.callback = callback;
             this.progressCallbackExecutor = executor;
+            num = this.numSeemPeers;
+            mined = this.mined;
+            shouldInvoke = numWaitingFor > 0;
+        }
+        if (shouldInvoke)
+            invokeProgressCallback(num, mined);
     }
 }

core/src/main/java/org/bitcoinj/params/UnitTestParams.java

@@ -16,8 +16,7 @@
 
 package org.bitcoinj.params;
 
-import org.bitcoinj.core.Block;
-import org.bitcoinj.core.NetworkParameters;
+import org.bitcoinj.core.*;
 
 import java.math.BigInteger;
 
@@ -26,6 +25,10 @@ import java.math.BigInteger;
  * {@link org.bitcoinj.core.Block#solve()} by setting difficulty to the easiest possible.
  */
 public class UnitTestParams extends NetworkParameters {
+    // A simple static key/address for re-use in unit tests, to speed things up.
+    public static ECKey TEST_KEY = new ECKey();
+    public static Address TEST_ADDRESS;
+
     public UnitTestParams() {
         super();
         id = ID_UNITTESTNET;
@@ -52,6 +55,7 @@ public class UnitTestParams extends NetworkParameters {
     public static synchronized UnitTestParams get() {
         if (instance == null) {
             instance = new UnitTestParams();
+            TEST_ADDRESS = TEST_KEY.toAddress(instance);
         }
         return instance;
     }

core/src/test/java/org/bitcoinj/core/TransactionBroadcastTest.java

@@ -18,6 +18,7 @@
 package org.bitcoinj.core;
 
 import com.google.common.util.concurrent.*;
+import org.bitcoinj.params.*;
 import org.bitcoinj.testing.*;
 import org.bitcoinj.utils.*;
 import org.junit.*;
@@ -100,6 +101,27 @@ public class TransactionBroadcastTest extends TestWithPeerGroup {
         assertNull(outbound(channels[1]));
     }
 
+    @Test
+    public void lateProgressCallback() throws Exception {
+        // Check that if we register a progress callback on a broadcast after the broadcast has started, it's invoked
+        // immediately with the latest state. This avoids API users writing accidentally racy code when they use
+        // a convenience method like peerGroup.broadcastTransaction.
+        InboundMessageQueuer[] channels = { connectPeer(1), connectPeer(2), connectPeer(3), connectPeer(4) };
+        Transaction tx = FakeTxBuilder.createFakeTx(params, CENT, UnitTestParams.TEST_ADDRESS);
+        tx.getConfidence().setSource(TransactionConfidence.Source.SELF);
+        TransactionBroadcast broadcast = peerGroup.broadcastTransaction(tx);
+        inbound(channels[1], InventoryMessage.with(tx));
+        pingAndWait(channels[1]);
+        final AtomicDouble p = new AtomicDouble();
+        broadcast.setProgressCallback(new TransactionBroadcast.ProgressCallback() {
+            @Override
+            public void onBroadcastProgress(double progress) {
+                p.set(progress);
+            }
+        }, Threading.SAME_THREAD);
+        assertEquals(1.0, p.get(), 0.01);
+    }
+
     @Test
     public void rejectHandling() throws Exception {
         InboundMessageQueuer[] channels = { connectPeer(0), connectPeer(1), connectPeer(2), connectPeer(3), connectPeer(4) };