Move broadcaster from wallet to base and to mempool (#113)

* Move broadcaster from wallet to base and to mempool

* Move broadcaster to base project and use signal events instead of an EventArgs, alsointroduce BroadcastCheck interface instead of overriding the broadcast manager

Author: dangershony

src/Blockcore/Base/BaseFeature.cs

@@ -11,6 +11,7 @@
 using Blockcore.Configuration;
 using Blockcore.Configuration.Settings;
 using Blockcore.Connection;
+using Blockcore.Connection.Broadcasting;
 using Blockcore.Consensus;
 using Blockcore.Consensus.Rules;
 using Blockcore.Consensus.Validators;
@@ -109,6 +110,7 @@ public sealed class BaseFeature : FullNodeFeature
         private readonly Network network;
 
         private readonly INodeStats nodeStats;
+        private readonly IBroadcasterManager broadcasterManager;
         private readonly IProvenBlockHeaderStore provenBlockHeaderStore;
 
         private readonly IConsensusManager consensusManager;
@@ -148,6 +150,7 @@ public BaseFeature(NodeSettings nodeSettings,
             ITipsManager tipsManager,
             IKeyValueRepository keyValueRepo,
             INodeStats nodeStats,
+            IBroadcasterManager broadcasterManager,
             IProvenBlockHeaderStore provenBlockHeaderStore = null)
         {
             this.chainState = Guard.NotNull(chainState, nameof(chainState));
@@ -164,6 +167,7 @@ public BaseFeature(NodeSettings nodeSettings,
             this.blockStore = blockStore;
             this.network = network;
             this.nodeStats = nodeStats;
+            this.broadcasterManager = broadcasterManager;
             this.provenBlockHeaderStore = provenBlockHeaderStore;
             this.partialValidator = partialValidator;
             this.peerBanning = Guard.NotNull(peerBanning, nameof(peerBanning));
@@ -211,6 +215,7 @@ public override async Task InitializeAsync()
             connectionParameters.TemplateBehaviors.Add(new PeerBanningBehavior(this.loggerFactory, this.peerBanning, this.nodeSettings));
             connectionParameters.TemplateBehaviors.Add(new BlockPullerBehavior(this.blockPuller, this.initialBlockDownloadState, this.dateTimeProvider, this.loggerFactory));
             connectionParameters.TemplateBehaviors.Add(new ConnectionManagerBehavior(this.connectionManager, this.loggerFactory));
+            connectionParameters.TemplateBehaviors.Add(new BroadcasterBehavior(this.broadcasterManager, this.loggerFactory));
 
             this.StartAddressManager(connectionParameters);
 
@@ -397,6 +402,8 @@ public static IFullNodeBuilder UseBaseFeature(this IFullNodeBuilder fullNodeBuil
                     services.AddSingleton<IKeyValueRepository, KeyValueRepository>();
                     services.AddSingleton<ITipsManager, TipsManager>();
                     services.AddSingleton<IAsyncProvider, AsyncProvider>();
+                    services.AddSingleton<IBroadcasterManager, BroadcasterManager>();
+                    services.AddSingleton<IBroadcastCheck, NoCheckBroadcastCheck>();
 
                     // Consensus
                     services.AddSingleton<ConsensusSettings>();

…roadcasting/TransactionBroadcastEntry.cs …/BroadcastTransactionStateChanedEntry.cssrc/Features/Blockcore.Features.Wallet/Broadcasting/TransactionBroadcastEntry.cs renamed to src/Blockcore/Connection/Broadcasting/BroadcastTransactionStateChanedEntry.cs src/Features/Blockcore.Features.Wallet/Broadcasting/TransactionBroadcastEntry.cs renamed to src/Blockcore/Connection/Broadcasting/BroadcastTransactionStateChanedEntry.cs

@@ -1,23 +1,22 @@
 using System;
-using Blockcore.Features.MemoryPool;
 
-namespace Blockcore.Features.Wallet.Broadcasting
+namespace Blockcore.Connection.Broadcasting
 {
-    public class TransactionBroadcastEntry
+    public class BroadcastTransactionStateChanedEntry
     {
         public NBitcoin.Transaction Transaction { get; }
 
         public TransactionBroadcastState TransactionBroadcastState { get; set; }
 
-        public string ErrorMessage => (this.MempoolError == null) ? string.Empty : (this.MempoolError.ConsensusError?.Message ?? this.MempoolError.Code ?? "Failed");
+        public string ErrorMessage => (this.MempoolError == null) ? string.Empty : $"Failed: {this.ErrorMessage}";
 
-        public MempoolError MempoolError { get; set; }
+        public string MempoolError { get; set; }
 
-        public TransactionBroadcastEntry(NBitcoin.Transaction transaction, TransactionBroadcastState transactionBroadcastState, MempoolError mempoolError)
+        public BroadcastTransactionStateChanedEntry(NBitcoin.Transaction transaction, TransactionBroadcastState transactionBroadcastState, string mempoolError)
         {
             this.Transaction = transaction ?? throw new ArgumentNullException(nameof(transaction));
             this.TransactionBroadcastState = transactionBroadcastState;
             this.MempoolError = mempoolError;
         }
     }
-}
+}

…llet/Broadcasting/BroadcasterBehavior.cs …tion/Broadcasting/BroadcasterBehavior.cssrc/Features/Blockcore.Features.Wallet/Broadcasting/BroadcasterBehavior.cs renamed to src/Blockcore/Connection/Broadcasting/BroadcasterBehavior.cs src/Features/Blockcore.Features.Wallet/Broadcasting/BroadcasterBehavior.cs renamed to src/Blockcore/Connection/Broadcasting/BroadcasterBehavior.cs

@@ -2,14 +2,14 @@
 using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
-using Blockcore.Features.Wallet.Interfaces;
+using Blockcore.Interfaces;
 using Blockcore.P2P.Peer;
 using Blockcore.P2P.Protocol;
 using Blockcore.P2P.Protocol.Behaviors;
 using Blockcore.P2P.Protocol.Payloads;
 using Microsoft.Extensions.Logging;
 
-namespace Blockcore.Features.Wallet.Broadcasting
+namespace Blockcore.Connection.Broadcasting
 {
     public class BroadcasterBehavior : NetworkPeerBehavior
     {
@@ -91,7 +91,7 @@ private void ProcessInvPayload(InvPayload invPayload)
             // if node has transaction we broadcast
             foreach (InventoryVector inv in invPayload.Inventory.Where(x => x.Type == InventoryType.MSG_TX))
             {
-                TransactionBroadcastEntry txEntry = this.broadcasterManager.GetTransaction(inv.Hash);
+                BroadcastTransactionStateChanedEntry txEntry = this.broadcasterManager.GetTransaction(inv.Hash);
                 if (txEntry != null)
                 {
                     this.broadcasterManager.AddOrUpdate(txEntry.Transaction, TransactionBroadcastState.Propagated);
@@ -104,7 +104,7 @@ protected async Task ProcessGetDataPayloadAsync(INetworkPeer peer, GetDataPayloa
             // If node asks for transaction we want to broadcast.
             foreach (InventoryVector inv in getDataPayload.Inventory.Where(x => x.Type == InventoryType.MSG_TX))
             {
-                TransactionBroadcastEntry txEntry = this.broadcasterManager.GetTransaction(inv.Hash);
+                BroadcastTransactionStateChanedEntry txEntry = this.broadcasterManager.GetTransaction(inv.Hash);
                 if ((txEntry != null) && (txEntry.TransactionBroadcastState != TransactionBroadcastState.CantBroadcast))
                 {
                     await peer.SendMessageAsync(new TxPayload(txEntry.Transaction)).ConfigureAwait(false);

…t/Broadcasting/BroadcasterManagerBase.cs …ction/Broadcasting/BroadcasterManager.cssrc/Features/Blockcore.Features.Wallet/Broadcasting/BroadcasterManagerBase.cs renamed to src/Blockcore/Connection/Broadcasting/BroadcasterManager.cs src/Features/Blockcore.Features.Wallet/Broadcasting/BroadcasterManagerBase.cs renamed to src/Blockcore/Connection/Broadcasting/BroadcasterManager.cs

@@ -2,56 +2,56 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
-using Blockcore.Connection;
-using Blockcore.Features.MemoryPool;
-using Blockcore.Features.Wallet.Interfaces;
+using Blockcore.Interfaces;
 using Blockcore.P2P.Peer;
 using Blockcore.P2P.Protocol.Payloads;
+using Blockcore.Signals;
 using Blockcore.Utilities;
 using ConcurrentCollections;
 using NBitcoin;
 
-namespace Blockcore.Features.Wallet.Broadcasting
+namespace Blockcore.Connection.Broadcasting
 {
-    public abstract class BroadcasterManagerBase : IBroadcasterManager
+    public class BroadcasterManager : IBroadcasterManager
     {
-        public event EventHandler<TransactionBroadcastEntry> TransactionStateChanged;
-
-        /// <summary> Connection manager for managing node connections.</summary>
         protected readonly IConnectionManager connectionManager;
+        private readonly ISignals signals;
+        private readonly IEnumerable<IBroadcastCheck> broadcastChecks;
 
-        public BroadcasterManagerBase(IConnectionManager connectionManager)
+        public BroadcasterManager(IConnectionManager connectionManager, ISignals signals, IEnumerable<IBroadcastCheck> broadcastChecks)
         {
             Guard.NotNull(connectionManager, nameof(connectionManager));
 
             this.connectionManager = connectionManager;
-            this.Broadcasts = new ConcurrentHashSet<TransactionBroadcastEntry>();
+            this.signals = signals;
+            this.broadcastChecks = broadcastChecks;
+            this.Broadcasts = new ConcurrentHashSet<BroadcastTransactionStateChanedEntry>();
         }
 
-        public void OnTransactionStateChanged(TransactionBroadcastEntry entry)
+        public void OnTransactionStateChanged(BroadcastTransactionStateChanedEntry entry)
         {
-            this.TransactionStateChanged?.Invoke(this, entry);
+            this.signals.Publish(new TransactionBroadcastEvent(this, entry));
         }
 
         /// <summary>Transactions to broadcast.</summary>
-        private ConcurrentHashSet<TransactionBroadcastEntry> Broadcasts { get; }
+        private ConcurrentHashSet<BroadcastTransactionStateChanedEntry> Broadcasts { get; }
 
         /// <summary>Retrieves a transaction with provided hash from the collection of transactions to broadcast.</summary>
         /// <param name="transactionHash">Hash of the transaction to retrieve.</param>
-        public TransactionBroadcastEntry GetTransaction(uint256 transactionHash)
+        public BroadcastTransactionStateChanedEntry GetTransaction(uint256 transactionHash)
         {
-            TransactionBroadcastEntry txEntry = this.Broadcasts.FirstOrDefault(x => x.Transaction.GetHash() == transactionHash);
+            BroadcastTransactionStateChanedEntry txEntry = this.Broadcasts.FirstOrDefault(x => x.Transaction.GetHash() == transactionHash);
             return txEntry ?? null;
         }
 
         /// <summary>Adds or updates a transaction from the collection of transactions to broadcast.</summary>
-        public void AddOrUpdate(Transaction transaction, TransactionBroadcastState transactionBroadcastState, MempoolError mempoolError = null)
+        public void AddOrUpdate(Transaction transaction, TransactionBroadcastState transactionBroadcastState, string errorMessage = null)
         {
-            TransactionBroadcastEntry broadcastEntry = this.Broadcasts.FirstOrDefault(x => x.Transaction.GetHash() == transaction.GetHash());
+            BroadcastTransactionStateChanedEntry broadcastEntry = this.Broadcasts.FirstOrDefault(x => x.Transaction.GetHash() == transaction.GetHash());
 
             if (broadcastEntry == null)
             {
-                broadcastEntry = new TransactionBroadcastEntry(transaction, transactionBroadcastState, mempoolError);
+                broadcastEntry = new BroadcastTransactionStateChanedEntry(transaction, transactionBroadcastState, errorMessage);
                 this.Broadcasts.Add(broadcastEntry);
                 this.OnTransactionStateChanged(broadcastEntry);
             }
@@ -62,7 +62,28 @@ public void AddOrUpdate(Transaction transaction, TransactionBroadcastState trans
             }
         }
 
-        public abstract Task BroadcastTransactionAsync(Transaction transaction);
+        public async Task BroadcastTransactionAsync(Transaction transaction)
+        {
+            Guard.NotNull(transaction, nameof(transaction));
+
+            if (this.IsPropagated(transaction))
+            {
+                return;
+            }
+
+            foreach (IBroadcastCheck broadcastCheck in this.broadcastChecks)
+            {
+                string error = await broadcastCheck.CheckTransaction(transaction).ConfigureAwait(false);
+
+                if (!string.IsNullOrEmpty(error))
+                {
+                    this.AddOrUpdate(transaction, TransactionBroadcastState.CantBroadcast, error);
+                    return;
+                }
+            }
+
+            await this.PropagateTransactionToPeersAsync(transaction, this.connectionManager.ConnectedPeers.ToList()).ConfigureAwait(false);
+        }
 
         /// <summary>
         /// Sends transaction to peers.
@@ -90,8 +111,8 @@ protected async Task PropagateTransactionToPeersAsync(Transaction transaction, L
         /// <summary>Checks if transaction was propagated to any peers on the network.</summary>
         protected bool IsPropagated(Transaction transaction)
         {
-            TransactionBroadcastEntry broadcastEntry = this.GetTransaction(transaction.GetHash());
+            BroadcastTransactionStateChanedEntry broadcastEntry = this.GetTransaction(transaction.GetHash());
             return (broadcastEntry != null) && (broadcastEntry.TransactionBroadcastState == TransactionBroadcastState.Propagated);
         }
     }
-}
+}

src/Blockcore/Connection/Broadcasting/NoCheckBroadcastCheck.cs

@@ -0,0 +1,23 @@
+using System.Linq;
+using System.Threading.Tasks;
+using Blockcore.Interfaces;
+using Blockcore.Utilities;
+using NBitcoin;
+
+namespace Blockcore.Connection.Broadcasting
+{
+    /// <summary>
+    /// Broadcast that makes not checks.
+    /// </summary>
+    public class NoCheckBroadcastCheck : IBroadcastCheck
+    {
+        public NoCheckBroadcastCheck()
+        {
+        }
+
+        public Task<string> CheckTransaction(Transaction transaction)
+        {
+            return Task.FromResult(string.Empty);
+        }
+    }
+}

src/Blockcore/Connection/Broadcasting/TransactionBroadcastEvent.cs

@@ -0,0 +1,21 @@
+using Blockcore.EventBus;
+using Blockcore.Interfaces;
+
+namespace Blockcore.Connection.Broadcasting
+{
+    /// <summary>
+    /// Event that is executed when a transaction is broadcasted.
+    /// </summary>
+    public class TransactionBroadcastEvent : EventBase
+    {
+        public BroadcastTransactionStateChanedEntry BroadcastEntry { get; }
+
+        public IBroadcasterManager BroadcasterManager { get; }
+
+        public TransactionBroadcastEvent(IBroadcasterManager broadcasterManager, BroadcastTransactionStateChanedEntry broadcastEntry)
+        {
+            this.BroadcasterManager = broadcasterManager;
+            this.BroadcastEntry = broadcastEntry;
+        }
+    }
+}

…roadcasting/TransactionBroadcastState.cs …roadcasting/TransactionBroadcastState.cssrc/Features/Blockcore.Features.Wallet/Broadcasting/TransactionBroadcastState.cs renamed to src/Blockcore/Connection/Broadcasting/TransactionBroadcastState.cs src/Features/Blockcore.Features.Wallet/Broadcasting/TransactionBroadcastState.cs renamed to src/Blockcore/Connection/Broadcasting/TransactionBroadcastState.cs

@@ -1,4 +1,4 @@
-namespace Blockcore.Features.Wallet.Broadcasting
+namespace Blockcore.Connection.Broadcasting
 {
     public enum TransactionBroadcastState
     {

src/Blockcore/EventBus/EventBase.cs

@@ -8,7 +8,6 @@ namespace Blockcore.EventBus
     /// <seealso cref="Blockcore.EventBus.IEvent" />
     public abstract class EventBase
     {
-        /// <inheritdoc />
         public Guid CorrelationId { get; }
 
         public EventBase()
@@ -22,4 +21,4 @@ public override string ToString()
             return $"{this.CorrelationId.ToString()} - {this.GetType().Name}";
         }
     }
-}
+}

src/Blockcore/Interfaces/IBroadcasterManager.cs

@@ -0,0 +1,24 @@
+using System;
+using System.Threading.Tasks;
+using Blockcore.Connection.Broadcasting;
+using NBitcoin;
+
+namespace Blockcore.Interfaces
+{
+    /// <summary>
+    /// Allow to check a transaction is valid before broadcasting it.
+    /// </summary>
+    public interface IBroadcastCheck
+    {
+        Task<string> CheckTransaction(Transaction transaction);
+    }
+
+    public interface IBroadcasterManager
+    {
+        Task BroadcastTransactionAsync(Transaction transaction);
+
+        BroadcastTransactionStateChanedEntry GetTransaction(uint256 transactionHash);
+
+        void AddOrUpdate(Transaction transaction, TransactionBroadcastState transactionBroadcastState, string errorMessage = null);
+    }
+}

src/Blockcore/Utilities/Extensions/ServiceCollectionExtensions.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Linq;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Blockcore.Utilities.Extensions
+{
+    public static class ServiceCollectionExtensions
+    {
+        /// <summary>
+        /// Converts a long that represents a number of bytes to be represented in MB.
+        /// </summary>
+        public static bool RemoveSingleton<T>(this IServiceCollection services)
+        {
+            // Remove the service if it exists.
+            return services.Remove(services.Where(sd => sd.ServiceType == typeof(T)).FirstOrDefault());
+        }
+    }
+}