Merge pull request #281 from blinklabs-io/feat/protocol-auto-start

feat: auto-start protocols with option to disable

cmd/block-fetch/main.go

@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/blinklabs-io/gouroboros"
+	ouroboros "github.com/blinklabs-io/gouroboros"
 	"github.com/blinklabs-io/gouroboros/cmd/common"
 	"github.com/blinklabs-io/gouroboros/ledger"
 	ocommon "github.com/blinklabs-io/gouroboros/protocol/common"
@@ -60,7 +60,6 @@ func main() {
 		fmt.Printf("ERROR: %s\n", err)
 		os.Exit(1)
 	}
-	o.BlockFetch().Client.Start()
 
 	blockHash, err := hex.DecodeString(f.hash)
 	if err != nil {

cmd/gouroboros/chainsync.go

@@ -151,10 +151,6 @@ func testChainSync(f *globalFlags) {
 		fmt.Printf("ERROR: %s\n", err)
 		os.Exit(1)
 	}
-	oConn.ChainSync().Client.Start()
-	if f.ntnProto {
-		oConn.BlockFetch().Client.Start()
-	}
 
 	var point common.Point
 	if chainSyncFlags.tip {

cmd/gouroboros/localtxsubmission.go

@@ -19,11 +19,12 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
+	"io/ioutil"
+	"os"
+
 	ouroboros "github.com/blinklabs-io/gouroboros"
 	"github.com/blinklabs-io/gouroboros/ledger"
 	"github.com/blinklabs-io/gouroboros/protocol/localtxsubmission"
-	"io/ioutil"
-	"os"
 )
 
 type localTxSubmissionFlags struct {
@@ -78,7 +79,6 @@ func testLocalTxSubmission(f *globalFlags) {
 		fmt.Printf("ERROR: %s\n", err)
 		os.Exit(1)
 	}
-	o.LocalTxSubmission().Client.Start()
 
 	var txBytes []byte
 	if localTxSubmissionFlags.txFile != "" {

cmd/gouroboros/mem_usage.go

@@ -85,7 +85,6 @@ func testMemUsage(f *globalFlags) {
 			fmt.Printf("ERROR: %s\n", err)
 			os.Exit(1)
 		}
-		o.ChainSync().Client.Start()
 
 		tip, err := o.ChainSync().Client.GetCurrentTip()
 		if err != nil {

cmd/gouroboros/query.go

@@ -17,9 +17,10 @@ package main
 import (
 	"flag"
 	"fmt"
+	"os"
+
 	ouroboros "github.com/blinklabs-io/gouroboros"
 	"github.com/blinklabs-io/gouroboros/protocol/localstatequery"
-	"os"
 )
 
 type queryFlags struct {
@@ -70,7 +71,6 @@ func testQuery(f *globalFlags) {
 		fmt.Printf("ERROR: %s\n", err)
 		os.Exit(1)
 	}
-	o.LocalStateQuery().Client.Start()
 
 	switch queryFlags.flagset.Args()[0] {
 	case "current-era":

cmd/peer-sharing/main.go

@@ -47,7 +47,6 @@ func main() {
 		fmt.Printf("ERROR: %s\n", err)
 		os.Exit(1)
 	}
-	o.PeerSharing().Client.Start()
 
 	peers, err := o.PeerSharing().Client.GetPeers(10)
 	if err != nil {

cmd/tx-monitor/main.go

@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/blinklabs-io/gouroboros"
+	ouroboros "github.com/blinklabs-io/gouroboros"
 	"github.com/blinklabs-io/gouroboros/cbor"
 	"github.com/blinklabs-io/gouroboros/cmd/common"
 
@@ -51,7 +51,6 @@ func main() {
 		fmt.Printf("ERROR: %s\n", err)
 		os.Exit(1)
 	}
-	o.LocalTxMonitor().Client.Start()
 
 	capacity, size, numberOfTxs, err := o.LocalTxMonitor().Client.GetSizes()
 	if err != nil {

connection.go

@@ -58,6 +58,7 @@ type Connection struct {
 	onceClose             sync.Once
 	sendKeepAlives        bool
 	delayMuxerStart       bool
+	delayProtocolStart    bool
 	fullDuplex            bool
 	// Mini-protocols
 	blockFetch              *blockfetch.BlockFetch
@@ -331,6 +332,25 @@ func (c *Connection) setupConnection() error {
 		if versionNtN.EnablePeerSharingProtocol {
 			c.peerSharing = peersharing.New(protoOptions, c.peerSharingConfig)
 		}
+		// Start protocols
+		if !c.delayProtocolStart {
+			if handshakeFullDuplex || !c.server {
+				c.blockFetch.Client.Start()
+				c.chainSync.Client.Start()
+				c.txSubmission.Client.Start()
+				if c.peerSharing != nil {
+					c.peerSharing.Client.Start()
+				}
+			}
+			if handshakeFullDuplex || c.server {
+				c.blockFetch.Server.Start()
+				c.chainSync.Server.Start()
+				c.txSubmission.Server.Start()
+				if c.peerSharing != nil {
+					c.peerSharing.Server.Start()
+				}
+			}
+		}
 	} else {
 		versionNtC := GetProtocolVersionNtC(handshakeVersion)
 		protoOptions.Mode = protocol.ProtocolModeNodeToClient
@@ -342,6 +362,29 @@ func (c *Connection) setupConnection() error {
 		if versionNtC.EnableLocalTxMonitorProtocol {
 			c.localTxMonitor = localtxmonitor.New(protoOptions, c.localTxMonitorConfig)
 		}
+		// Start protocols
+		if !c.delayProtocolStart {
+			if handshakeFullDuplex || !c.server {
+				c.chainSync.Client.Start()
+				c.localTxSubmission.Client.Start()
+				if c.localStateQuery != nil {
+					c.localStateQuery.Client.Start()
+				}
+				if c.localTxMonitor != nil {
+					c.localTxMonitor.Client.Start()
+				}
+			}
+			if handshakeFullDuplex || c.server {
+				c.chainSync.Server.Start()
+				c.localTxSubmission.Server.Start()
+				if c.localStateQuery != nil {
+					c.localStateQuery.Server.Start()
+				}
+				if c.localTxMonitor != nil {
+					c.localTxMonitor.Server.Start()
+				}
+			}
+		}
 	}
 	// Start muxer
 	diffusionMode := muxer.DiffusionModeInitiator

connection_options.go

@@ -87,6 +87,15 @@ func WithDelayMuxerStart(delayMuxerStart bool) ConnectionOptionFunc {
 	}
 }
 
+// WithDelayProtocolStart specifies whether to delay the start of the relevant mini-protocols. This is useful
+// if you are maintaining lots of connections and want to reduce resource overhead by only starting particular
+// protocols
+func WithDelayProtocolStart(delayProtocolStart bool) ConnectionOptionFunc {
+	return func(c *Connection) {
+		c.delayProtocolStart = delayProtocolStart
+	}
+}
+
 // WithFullDuplex specifies whether to enable full-duplex mode when acting as a client
 func WithFullDuplex(fullDuplex bool) ConnectionOptionFunc {
 	return func(c *Connection) {