Feature: Add solidification stage to pipeline and only broadcast solid transactions

src/main/java/com/iota/iri/IRI.java

@@ -7,6 +7,7 @@
 import com.iota.iri.conf.ConfigFactory;
 import com.iota.iri.conf.IotaConfig;
 import com.iota.iri.network.NetworkInjectionConfiguration;
+import com.iota.iri.network.pipeline.BroadcastQueue;
 import com.iota.iri.service.API;
 import com.iota.iri.utils.IotaUtils;
 import com.iota.iri.service.restserver.resteasy.RestEasy;

src/main/java/com/iota/iri/Iota.java

@@ -6,6 +6,7 @@
 import com.iota.iri.network.NeighborRouter;
 import com.iota.iri.network.TipsRequester;
 import com.iota.iri.network.TransactionRequester;
+import com.iota.iri.network.pipeline.BroadcastQueue;
 import com.iota.iri.network.pipeline.TransactionProcessingPipeline;
 import com.iota.iri.service.ledger.LedgerService;
 import com.iota.iri.service.milestone.*;
@@ -106,6 +107,8 @@ public class Iota {
     public final TipsViewModel tipsViewModel;
     public final TipSelector tipsSelector;
 
+    private BroadcastQueue broadcastQueue;
+
     /**
      * Initializes the latest snapshot and then creates all services needed to run an IOTA node.
      *
@@ -181,7 +184,7 @@ public void init() throws Exception {
             tangle.clearMetadata(com.iota.iri.model.persistables.Transaction.class);
         }
 
-        transactionValidator.init();
+        transactionValidator.init(this.broadcastQueue);
 
         txPipeline.start();
         neighborRouter.start();

src/main/java/com/iota/iri/MainInjectionConfiguration.java

@@ -9,6 +9,7 @@
 import com.iota.iri.network.NeighborRouter;
 import com.iota.iri.network.TipsRequester;
 import com.iota.iri.network.TransactionRequester;
+import com.iota.iri.network.pipeline.BroadcastQueue;
 import com.iota.iri.network.pipeline.TransactionProcessingPipeline;
 import com.iota.iri.service.API;
 import com.iota.iri.service.ledger.LedgerService;
@@ -204,4 +205,10 @@ protected void configure() {
         bind(TipsViewModel.class).asEagerSingleton();
     }
 
+    @Singleton
+    @Provides
+    BroadcastQueue provideBroadcastQueue(){
+        return configuration.getBroadcastQueue();
+    }
+
 }

src/main/java/com/iota/iri/TransactionValidator.java

@@ -10,6 +10,7 @@
 import com.iota.iri.model.Hash;
 import com.iota.iri.model.TransactionHash;
 import com.iota.iri.network.TransactionRequester;
+import com.iota.iri.network.pipeline.BroadcastQueue;
 import com.iota.iri.service.snapshot.SnapshotProvider;
 import com.iota.iri.storage.Tangle;
 import org.slf4j.Logger;
@@ -33,7 +34,7 @@ public class TransactionValidator {
     private static final long MAX_TIMESTAMP_FUTURE = 2L * 60L * 60L;
     private static final long MAX_TIMESTAMP_FUTURE_MS = MAX_TIMESTAMP_FUTURE * 1_000L;
 
-
+    private BroadcastQueue broadcastQueue;
     /////////////////////////////////fields for solidification thread//////////////////////////////////////
 
     private Thread newSolidThread;
@@ -85,7 +86,8 @@ public class TransactionValidator {
      *
      * @see #spawnSolidTransactionsPropagation()
      */
-    public void init() {
+    public void init(BroadcastQueue broadcastQueue) {
+        this.broadcastQueue = broadcastQueue;
         newSolidThread.start();
     }
 
@@ -277,7 +279,7 @@ public boolean checkSolidity(Hash hash, int maxProcessedTransactions) throws Exc
             }
         }
         if (solid) {
-            updateSolidTransactions(tangle, snapshotProvider.getInitialSnapshot(), analyzedHashes);
+            updateSolidTransactions(tangle, snapshotProvider.getInitialSnapshot(), analyzedHashes, broadcastQueue);
         }
         analyzedHashes.clear();
         return solid;

src/main/java/com/iota/iri/conf/BaseIotaConfig.java

@@ -4,6 +4,7 @@
 import com.iota.iri.model.Hash;
 import com.iota.iri.model.HashFactory;
 import com.iota.iri.utils.IotaUtils;
+import com.iota.iri.network.pipeline.BroadcastQueue;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
@@ -56,6 +57,7 @@ public abstract class BaseIotaConfig implements IotaConfig {
     protected boolean dnsRefresherEnabled = Defaults.DNS_REFRESHER_ENABLED;
     protected boolean dnsResolutionEnabled = Defaults.DNS_RESOLUTION_ENABLED;
     protected List<String> neighbors = Collections.EMPTY_LIST;
+    protected BroadcastQueue broadcastQueue = new BroadcastQueue();
 
     //IXI
     protected String ixiDir = Defaults.IXI_DIR;
@@ -172,6 +174,11 @@ public String getApiHost() {
         return apiHost;
     }
 
+    @Override
+    public BroadcastQueue getBroadcastQueue() {
+        return broadcastQueue;
+    }
+
     @JsonProperty
     @Parameter(names = {"--api-host"}, description = APIConfig.Descriptions.API_HOST)
     protected void setApiHost(String apiHost) {

src/main/java/com/iota/iri/conf/NetworkConfig.java

@@ -1,5 +1,7 @@
 package com.iota.iri.conf;
 
+import com.iota.iri.network.pipeline.BroadcastQueue;
+
 import java.util.List;
 
 /**
@@ -89,6 +91,11 @@ public interface NetworkConfig extends Config {
      */
     int getCacheSizeBytes();
 
+    /**
+     * @return The current {@link BroadcastQueue}
+     */
+    BroadcastQueue getBroadcastQueue();
+
     interface Descriptions {
         String NEIGHBORING_SOCKET_ADDRESS = "The address to bind the TCP server socket to.";
         String NEIGHBORING_SOCKET_PORT = "The TCP Receiver Port.";

src/main/java/com/iota/iri/controllers/TransactionViewModel.java

@@ -2,6 +2,9 @@
 
 import com.iota.iri.model.*;
 import com.iota.iri.model.persistables.*;
+import com.iota.iri.network.pipeline.BroadcastQueue;
+import com.iota.iri.network.pipeline.ProcessingContext;
+import com.iota.iri.network.pipeline.BroadcastPayload;
 import com.iota.iri.service.snapshot.Snapshot;
 import com.iota.iri.storage.Indexable;
 import com.iota.iri.storage.Persistable;
@@ -712,7 +715,8 @@ public void setMetadata() {
                 : TransactionViewModel.FILLED_SLOT;
     }
 
-    public static void updateSolidTransactions(Tangle tangle, Snapshot initialSnapshot, final LinkedHashSet<Hash> analyzedHashes)
+    public static void updateSolidTransactions(Tangle tangle, Snapshot initialSnapshot,
+                                               final Set<Hash> analyzedHashes, BroadcastQueue broadcastQueue)
             throws Exception {
         Object[] hashes = analyzedHashes.toArray();
         TransactionViewModel transactionViewModel;
@@ -724,6 +728,8 @@ public static void updateSolidTransactions(Tangle tangle, Snapshot initialSnapsh
             if (!transactionViewModel.isSolid()) {
                 transactionViewModel.updateSolid(true);
                 transactionViewModel.update(tangle, initialSnapshot, "solid|height");
+                BroadcastPayload payload = new BroadcastPayload(null,transactionViewModel);
+                broadcastQueue.add(new ProcessingContext(payload));
             }
         }
     }

src/main/java/com/iota/iri/network/NetworkInjectionConfiguration.java

@@ -48,7 +48,8 @@ TransactionProcessingPipeline provideTransactionProcessingPipeline(NeighborRoute
             TipsViewModel tipsViewModel, LatestMilestoneTracker latestMilestoneTracker,
             TransactionRequester transactionRequester) {
         return new TransactionProcessingPipelineImpl(neighborRouter, configuration, txValidator, tangle,
-                snapshotProvider, tipsViewModel, latestMilestoneTracker, transactionRequester);
+                snapshotProvider, tipsViewModel, latestMilestoneTracker, transactionRequester,
+                configuration.getBroadcastQueue());
     }
 
     @Singleton

src/main/java/com/iota/iri/network/pipeline/BroadcastQueue.java

@@ -0,0 +1,46 @@
+package com.iota.iri.network.pipeline;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * A queue for transactions intended to be submitted to the {@link BroadcastStage}
+ * for processing
+ */
+public class BroadcastQueue {
+
+    /** A blocking queue to store transactions for broadcasting */
+    private BlockingQueue<ProcessingContext> broadcastStageQueue = new ArrayBlockingQueue<>(100);
+
+    /** An object to be used for synchronizing calls */
+    private final Object broadcastSync = new Object();
+
+
+    /**
+     * Add transactions to the Broadcast Queue
+     * @param context Transaction context to be passed to the {@link BroadcastStage}
+     * @return True if added properly, False if not
+     */
+    public boolean add(ProcessingContext context) {
+        synchronized (broadcastSync) {
+            try {
+                this.broadcastStageQueue.put(context);
+                return true;
+            } catch (Exception e) {
+                return false;
+            }
+        }
+
+    }
+
+    /**
+     * Getter for the current Broadcast Queue
+     * @return BlockingQueue of all transactions left to be broadcasted
+     */
+    public BlockingQueue<ProcessingContext> get(){
+        /** Call is synchronized to ensure all pending additions have completed before sending the state */
+        synchronized (broadcastSync) {
+            return this.broadcastStageQueue;
+        }
+    }
+}

src/main/java/com/iota/iri/network/pipeline/TransactionProcessingPipelineImpl.java

@@ -66,8 +66,8 @@ public class TransactionProcessingPipelineImpl implements TransactionProcessingP
     private BlockingQueue<ProcessingContext> preProcessStageQueue = new ArrayBlockingQueue<>(100);
     private BlockingQueue<ProcessingContext> validationStageQueue = new ArrayBlockingQueue<>(100);
     private BlockingQueue<ProcessingContext> receivedStageQueue = new ArrayBlockingQueue<>(100);
-    private BlockingQueue<ProcessingContext> broadcastStageQueue = new ArrayBlockingQueue<>(100);
     private BlockingQueue<ProcessingContext> replyStageQueue = new ArrayBlockingQueue<>(100);
+    private BroadcastQueue broadcastStageQueue;
 
     /**
      * Creates a {@link TransactionProcessingPipeline}.
@@ -84,7 +84,7 @@ public class TransactionProcessingPipelineImpl implements TransactionProcessingP
     public TransactionProcessingPipelineImpl(NeighborRouter neighborRouter, NodeConfig config,
             TransactionValidator txValidator, Tangle tangle, SnapshotProvider snapshotProvider,
             TipsViewModel tipsViewModel, LatestMilestoneTracker latestMilestoneTracker,
-            TransactionRequester transactionRequester) {
+            TransactionRequester transactionRequester, BroadcastQueue broadcastStageQueue) {
         FIFOCache<Long, Hash> recentlySeenBytesCache = new FIFOCache<>(config.getCacheSizeBytes());
         this.preProcessStage = new PreProcessStage(recentlySeenBytesCache);
         this.replyStage = new ReplyStage(neighborRouter, config, tangle, tipsViewModel, latestMilestoneTracker,
@@ -94,6 +94,7 @@ public TransactionProcessingPipelineImpl(NeighborRouter neighborRouter, NodeConf
         this.receivedStage = new ReceivedStage(tangle, txValidator, snapshotProvider, transactionRequester);
         this.batchedHasher = BatchedHasherFactory.create(BatchedHasherFactory.Type.BCTCURL81, 20);
         this.hashingStage = new HashingStage(batchedHasher);
+        this.broadcastStageQueue = broadcastStageQueue;
     }
 
     @Override
@@ -103,7 +104,7 @@ public void start() {
         addStage("validation", validationStageQueue, validationStage);
         addStage("reply", replyStageQueue, replyStage);
         addStage("received", receivedStageQueue, receivedStage);
-        addStage("broadcast", broadcastStageQueue, broadcastStage);
+        addStage("broadcast", broadcastStageQueue.get(), broadcastStage);
     }
 
     /**
@@ -118,7 +119,14 @@ private void addStage(String name, BlockingQueue<ProcessingContext> queue,
         stagesThreadPool.submit(new Thread(() -> {
             try {
                 while (!Thread.currentThread().isInterrupted()) {
-                    ProcessingContext ctx = stage.process(queue.take());
+                    ProcessingContext queueTake;
+                    if(name.equals("broadcast")) {
+                        queueTake = broadcastStageQueue.get().take();
+                    } else{
+                        queueTake = queue.take();
+                    }
+                    ProcessingContext ctx = stage.process(queueTake);
+
                     switch (ctx.getNextStage()) {
                         case REPLY:
                             replyStageQueue.put(ctx);
@@ -135,7 +143,7 @@ private void addStage(String name, BlockingQueue<ProcessingContext> queue,
                             receivedStageQueue.put(payload.getRight());
                             break;
                         case BROADCAST:
-                            broadcastStageQueue.put(ctx);
+                            broadcastStageQueue.add(ctx);
                             break;
                         case ABORT:
                             break;
@@ -160,7 +168,7 @@ public BlockingQueue<ProcessingContext> getReceivedStageQueue() {
 
     @Override
     public BlockingQueue<ProcessingContext> getBroadcastStageQueue() {
-        return broadcastStageQueue;
+        return broadcastStageQueue.get();
     }
 
     @Override

src/test/java/com/iota/iri/TransactionValidatorTest.java

@@ -7,6 +7,7 @@
 import com.iota.iri.crypto.SpongeFactory;
 import com.iota.iri.model.TransactionHash;
 import com.iota.iri.network.TransactionRequester;
+import com.iota.iri.network.pipeline.BroadcastQueue;
 import com.iota.iri.service.snapshot.SnapshotProvider;
 import com.iota.iri.service.snapshot.impl.SnapshotMockUtils;
 import com.iota.iri.storage.Tangle;
@@ -42,6 +43,9 @@ public class TransactionValidatorTest {
   @Mock
   private static SnapshotProvider snapshotProvider;
 
+  @Mock
+  private static BroadcastQueue broadcastQueue;
+
   @BeforeClass
   public static void setUp() throws Exception {
     dbFolder.create();
@@ -67,6 +71,7 @@ public void setUpEach() {
     TransactionRequester txRequester = new TransactionRequester(tangle, snapshotProvider);
     txValidator = new TransactionValidator(tangle, snapshotProvider, tipsViewModel, txRequester, new MainnetConfig());
     txValidator.setMwm(false, MAINNET_MWM);
+    txValidator.init(broadcastQueue);
   }
 
   @Test

src/test/java/com/iota/iri/network/pipeline/TransactionProcessingPipelineTest.java

@@ -18,6 +18,8 @@
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
 
+import java.util.concurrent.BlockingQueue;
+
 public class TransactionProcessingPipelineTest {
 
     @Rule
@@ -92,6 +94,12 @@ public class TransactionProcessingPipelineTest {
     @Mock
     private ProcessingContext abortCtx;
 
+    @Mock
+    private BroadcastQueue broadcastQueue;
+
+    @Mock
+    private BlockingQueue blockingQueue;
+
     private void mockHashingStage(TransactionProcessingPipeline pipeline) {
         Mockito.when(hashingPayload.getTxTrits()).thenReturn(null);
         Mockito.doAnswer(invocation -> {
@@ -114,7 +122,7 @@ public void processingAValidNewTransactionFlowsThroughTheEntirePipeline() throws
 
         TransactionProcessingPipeline pipeline = new TransactionProcessingPipelineImpl(neighborRouter, nodeConfig,
                 transactionValidator, tangle, snapshotProvider, tipsViewModel, latestMilestoneTracker,
-                transactionRequester);
+                transactionRequester, broadcastQueue);
 
         // inject mocks
         injectMockedStagesIntoPipeline(pipeline);
@@ -138,6 +146,10 @@ public void processingAValidNewTransactionFlowsThroughTheEntirePipeline() throws
         Mockito.when(broadcastCtx.getNextStage()).thenReturn(TransactionProcessingPipeline.Stage.BROADCAST);
         Mockito.when(receivedStage.process(receivedCtx)).thenReturn(broadcastCtx);
 
+        // mock broadcastQueue
+        Mockito.when(broadcastQueue.get()).thenReturn(blockingQueue);
+        Mockito.when(blockingQueue.take()).thenReturn(broadcastCtx);
+
         pipeline.start();
 
         // send in actual payload to kick off the 'processing'
@@ -159,7 +171,7 @@ public void processingAValidNewTransactionFlowsThroughTheEntirePipeline() throws
     public void processingAKnownTransactionOnlyFlowsToTheReplyStage() throws InterruptedException {
         TransactionProcessingPipeline pipeline = new TransactionProcessingPipelineImpl(neighborRouter, nodeConfig,
                 transactionValidator, tangle, snapshotProvider, tipsViewModel, latestMilestoneTracker,
-                transactionRequester);
+                transactionRequester, broadcastQueue);
 
         // inject mocks
         pipeline.setPreProcessStage(preProcessStage);
@@ -193,7 +205,7 @@ public void processingAValidNewTransactionNotOriginatingFromANeighborFlowsThroug
             throws InterruptedException {
         TransactionProcessingPipeline pipeline = new TransactionProcessingPipelineImpl(neighborRouter, nodeConfig,
                 transactionValidator, tangle, snapshotProvider, tipsViewModel, latestMilestoneTracker,
-                transactionRequester);
+                transactionRequester, broadcastQueue);
         // inject mocks
         injectMockedStagesIntoPipeline(pipeline);
 
@@ -214,6 +226,10 @@ public void processingAValidNewTransactionNotOriginatingFromANeighborFlowsThroug
         Mockito.when(broadcastCtx.getNextStage()).thenReturn(TransactionProcessingPipeline.Stage.BROADCAST);
         Mockito.when(receivedStage.process(receivedCtx)).thenReturn(broadcastCtx);
 
+        // mock broadcastQueue
+        Mockito.when(broadcastQueue.get()).thenReturn(blockingQueue);
+        Mockito.when(blockingQueue.take()).thenReturn(broadcastCtx);
+
         pipeline.start();
 
         // send into the pipeline but without originating from a neighbor
@@ -238,7 +254,7 @@ public void processingAValidNewTransactionNotOriginatingFromANeighborFlowsThroug
     public void anInvalidNewTransactionStopsBeingProcessedAfterTheValidationStage() throws InterruptedException {
         TransactionProcessingPipeline pipeline = new TransactionProcessingPipelineImpl(neighborRouter, nodeConfig,
                 transactionValidator, tangle, snapshotProvider, tipsViewModel, latestMilestoneTracker,
-                transactionRequester);
+                transactionRequester, broadcastQueue);
 
         // inject mocks
         injectMockedStagesIntoPipeline(pipeline);