Merge pull request #114 from cloudstruct/feature/functional-options-protocols

feat: functional options for protocol configs

Author: agaffney

cmd/go-ouroboros-network/chainsync.go

@@ -76,8 +76,8 @@ var eraIntersect = map[int]map[string][]interface{}{
 	},
 }
 
-func buildChainSyncCallbackConfig() *chainsync.ChainSyncCallbackConfig {
-	return &chainsync.ChainSyncCallbackConfig{
+func buildChainSyncConfig() chainsync.Config {
+	return chainsync.Config{
 		AwaitReplyFunc:        chainSyncAwaitReplyHandler,
 		RollBackwardFunc:      chainSyncRollBackwardHandler,
 		RollForwardFunc:       chainSyncRollForwardHandler,
@@ -87,8 +87,8 @@ func buildChainSyncCallbackConfig() *chainsync.ChainSyncCallbackConfig {
 	}
 }
 
-func buildBlockFetchCallbackConfig() *blockfetch.BlockFetchCallbackConfig {
-	return &blockfetch.BlockFetchCallbackConfig{
+func buildBlockFetchConfig() blockfetch.Config {
+	return blockfetch.Config{
 		StartBatchFunc: blockFetchStartBatchHandler,
 		NoBlocksFunc:   blockFetchNoBlocksHandler,
 		BlockFunc:      blockFetchBlockHandler,
@@ -124,13 +124,15 @@ func testChainSync(f *globalFlags) {
 		ouroboros.WithErrorChan(errorChan),
 		ouroboros.WithNodeToNode(f.ntnProto),
 		ouroboros.WithKeepAlive(true),
+		ouroboros.WithBlockFetchConfig(buildBlockFetchConfig()),
+		ouroboros.WithChainSyncConfig(buildChainSyncConfig()),
 	)
 	if err != nil {
 		fmt.Printf("ERROR: %s\n", err)
 		os.Exit(1)
 	}
-	o.ChainSync.Start(buildChainSyncCallbackConfig())
-	o.BlockFetch.Start(buildBlockFetchCallbackConfig())
+	o.ChainSync.Start()
+	o.BlockFetch.Start()
 
 	syncState.oConn = o
 	syncState.readyForNextBlockChan = make(chan bool)

cmd/go-ouroboros-network/localtxsubmission.go

@@ -31,8 +31,8 @@ func newLocalTxSubmissionFlags() *localTxSubmissionFlags {
 	return f
 }
 
-func buildLocalTxSubmissionCallbackConfig() *localtxsubmission.CallbackConfig {
-	return &localtxsubmission.CallbackConfig{
+func buildLocalTxSubmissionConfig() localtxsubmission.Config {
+	return localtxsubmission.Config{
 		AcceptTxFunc: localTxSubmissionAcceptTxHandler,
 		RejectTxFunc: localTxSubmissionRejectTxHandler,
 	}
@@ -63,12 +63,13 @@ func testLocalTxSubmission(f *globalFlags) {
 		ouroboros.WithErrorChan(errorChan),
 		ouroboros.WithNodeToNode(f.ntnProto),
 		ouroboros.WithKeepAlive(true),
+		ouroboros.WithLocalTxSubmissionConfig(buildLocalTxSubmissionConfig()),
 	)
 	if err != nil {
 		fmt.Printf("ERROR: %s\n", err)
 		os.Exit(1)
 	}
-	o.LocalTxSubmission.Start(buildLocalTxSubmissionCallbackConfig())
+	o.LocalTxSubmission.Start()
 
 	txData, err := ioutil.ReadFile(localTxSubmissionFlags.txFile)
 	if err != nil {

options.go

@@ -1,6 +1,12 @@
 package ouroboros
 
 import (
+	"github.com/cloudstruct/go-ouroboros-network/protocol/blockfetch"
+	"github.com/cloudstruct/go-ouroboros-network/protocol/chainsync"
+	"github.com/cloudstruct/go-ouroboros-network/protocol/keepalive"
+	"github.com/cloudstruct/go-ouroboros-network/protocol/localstatequery"
+	"github.com/cloudstruct/go-ouroboros-network/protocol/localtxsubmission"
+	"github.com/cloudstruct/go-ouroboros-network/protocol/txsubmission"
 	"net"
 )
 
@@ -53,3 +59,39 @@ func WithFullDuplex(fullDuplex bool) OuroborosOptionFunc {
 		o.fullDuplex = fullDuplex
 	}
 }
+
+func WithBlockFetchConfig(cfg blockfetch.Config) OuroborosOptionFunc {
+	return func(o *Ouroboros) {
+		o.blockFetchConfig = &cfg
+	}
+}
+
+func WithChainSyncConfig(cfg chainsync.Config) OuroborosOptionFunc {
+	return func(o *Ouroboros) {
+		o.chainSyncConfig = &cfg
+	}
+}
+
+func WithKeepAliveConfig(cfg keepalive.Config) OuroborosOptionFunc {
+	return func(o *Ouroboros) {
+		o.keepAliveConfig = &cfg
+	}
+}
+
+func WithLocalStateQueryConfig(cfg localstatequery.Config) OuroborosOptionFunc {
+	return func(o *Ouroboros) {
+		o.localStateQueryConfig = &cfg
+	}
+}
+
+func WithLocalTxSubmissionConfig(cfg localtxsubmission.Config) OuroborosOptionFunc {
+	return func(o *Ouroboros) {
+		o.localTxSubmissionConfig = &cfg
+	}
+}
+
+func WithTxSubmissionConfig(cfg txsubmission.Config) OuroborosOptionFunc {
+	return func(o *Ouroboros) {
+		o.txSubmissionConfig = &cfg
+	}
+}

ouroboros.go

@@ -27,13 +27,19 @@ type Ouroboros struct {
 	delayMuxerStart    bool
 	fullDuplex         bool
 	// Mini-protocols
-	Handshake         *handshake.Handshake
-	ChainSync         *chainsync.ChainSync
-	BlockFetch        *blockfetch.BlockFetch
-	KeepAlive         *keepalive.KeepAlive
-	LocalTxSubmission *localtxsubmission.LocalTxSubmission
-	LocalStateQuery   *localstatequery.LocalStateQuery
-	TxSubmission      *txsubmission.TxSubmission
+	Handshake               *handshake.Handshake
+	ChainSync               *chainsync.ChainSync
+	chainSyncConfig         *chainsync.Config
+	BlockFetch              *blockfetch.BlockFetch
+	blockFetchConfig        *blockfetch.Config
+	KeepAlive               *keepalive.KeepAlive
+	keepAliveConfig         *keepalive.Config
+	LocalTxSubmission       *localtxsubmission.LocalTxSubmission
+	localTxSubmissionConfig *localtxsubmission.Config
+	LocalStateQuery         *localstatequery.LocalStateQuery
+	localStateQueryConfig   *localstatequery.Config
+	TxSubmission            *txsubmission.TxSubmission
+	txSubmissionConfig      *txsubmission.Config
 }
 
 func New(options ...OuroborosOptionFunc) (*Ouroboros, error) {
@@ -118,11 +124,16 @@ func (o *Ouroboros) setupConnection() error {
 		protoOptions.Role = protocol.ProtocolRoleClient
 	}
 	// Perform handshake
-	o.Handshake = handshake.New(protoOptions, protoVersions)
+	handshakeConfig := &handshake.Config{
+		ProtocolVersions: protoVersions,
+		NetworkMagic:     o.networkMagic,
+		ClientFullDuplex: o.fullDuplex,
+	}
+	o.Handshake = handshake.New(protoOptions, handshakeConfig)
 	o.Handshake.Start()
 	// TODO: figure out better way to signify automatic handshaking and returning the chosen version
 	if !o.server {
-		err := o.Handshake.ProposeVersions(protoVersions, o.networkMagic, o.fullDuplex)
+		err := o.Handshake.ProposeVersions()
 		if err != nil {
 			return err
 		}
@@ -151,22 +162,22 @@ func (o *Ouroboros) setupConnection() error {
 	if o.useNodeToNodeProto {
 		versionNtN := GetProtocolVersionNtN(o.Handshake.Version)
 		protoOptions.Mode = protocol.ProtocolModeNodeToNode
-		o.ChainSync = chainsync.New(protoOptions)
-		o.BlockFetch = blockfetch.New(protoOptions)
-		o.TxSubmission = txsubmission.New(protoOptions)
+		o.ChainSync = chainsync.New(protoOptions, o.chainSyncConfig)
+		o.BlockFetch = blockfetch.New(protoOptions, o.blockFetchConfig)
+		o.TxSubmission = txsubmission.New(protoOptions, o.txSubmissionConfig)
 		if versionNtN.EnableKeepAliveProtocol {
-			o.KeepAlive = keepalive.New(protoOptions)
+			o.KeepAlive = keepalive.New(protoOptions, o.keepAliveConfig)
 			if o.sendKeepAlives {
-				o.KeepAlive.Start(nil)
+				o.KeepAlive.Start()
 			}
 		}
 	} else {
 		versionNtC := GetProtocolVersionNtC(o.Handshake.Version)
 		protoOptions.Mode = protocol.ProtocolModeNodeToClient
-		o.ChainSync = chainsync.New(protoOptions)
-		o.LocalTxSubmission = localtxsubmission.New(protoOptions)
+		o.ChainSync = chainsync.New(protoOptions, o.chainSyncConfig)
+		o.LocalTxSubmission = localtxsubmission.New(protoOptions, o.localTxSubmissionConfig)
 		if versionNtC.EnableLocalQueryProtocol {
-			o.LocalStateQuery = localstatequery.New(protoOptions)
+			o.LocalStateQuery = localstatequery.New(protoOptions, o.localStateQueryConfig)
 		}
 	}
 	// Start muxer

protocol/blockfetch/blockfetch.go

@@ -66,24 +66,26 @@ var StateMap = protocol.StateMap{
 
 type BlockFetch struct {
 	*protocol.Protocol
-	callbackConfig *BlockFetchCallbackConfig
+	config *Config
 }
 
-type BlockFetchCallbackConfig struct {
-	StartBatchFunc BlockFetchStartBatchFunc
-	NoBlocksFunc   BlockFetchNoBlocksFunc
-	BlockFunc      BlockFetchBlockFunc
-	BatchDoneFunc  BlockFetchBatchDoneFunc
+type Config struct {
+	StartBatchFunc StartBatchFunc
+	NoBlocksFunc   NoBlocksFunc
+	BlockFunc      BlockFunc
+	BatchDoneFunc  BatchDoneFunc
 }
 
 // Callback function types
-type BlockFetchStartBatchFunc func() error
-type BlockFetchNoBlocksFunc func() error
-type BlockFetchBlockFunc func(uint, interface{}) error
-type BlockFetchBatchDoneFunc func() error
-
-func New(options protocol.ProtocolOptions) *BlockFetch {
-	b := &BlockFetch{}
+type StartBatchFunc func() error
+type NoBlocksFunc func() error
+type BlockFunc func(uint, interface{}) error
+type BatchDoneFunc func() error
+
+func New(options protocol.ProtocolOptions, cfg *Config) *BlockFetch {
+	b := &BlockFetch{
+		config: cfg,
+	}
 	protoConfig := protocol.ProtocolConfig{
 		Name:                PROTOCOL_NAME,
 		ProtocolId:          PROTOCOL_ID,
@@ -100,8 +102,7 @@ func New(options protocol.ProtocolOptions) *BlockFetch {
 	return b
 }
 
-func (b *BlockFetch) Start(callbackConfig *BlockFetchCallbackConfig) {
-	b.callbackConfig = callbackConfig
+func (b *BlockFetch) Start() {
 	b.Protocol.Start()
 }
 
@@ -133,23 +134,23 @@ func (b *BlockFetch) ClientDone() error {
 }
 
 func (b *BlockFetch) handleStartBatch() error {
-	if b.callbackConfig.StartBatchFunc == nil {
+	if b.config.StartBatchFunc == nil {
 		return fmt.Errorf("received block-fetch StartBatch message but no callback function is defined")
 	}
 	// Call the user callback function
-	return b.callbackConfig.StartBatchFunc()
+	return b.config.StartBatchFunc()
 }
 
 func (b *BlockFetch) handleNoBlocks() error {
-	if b.callbackConfig.NoBlocksFunc == nil {
+	if b.config.NoBlocksFunc == nil {
 		return fmt.Errorf("received block-fetch NoBlocks message but no callback function is defined")
 	}
 	// Call the user callback function
-	return b.callbackConfig.NoBlocksFunc()
+	return b.config.NoBlocksFunc()
 }
 
 func (b *BlockFetch) handleBlock(msgGeneric protocol.Message) error {
-	if b.callbackConfig.BlockFunc == nil {
+	if b.config.BlockFunc == nil {
 		return fmt.Errorf("received block-fetch Block message but no callback function is defined")
 	}
 	msg := msgGeneric.(*MsgBlock)
@@ -163,13 +164,13 @@ func (b *BlockFetch) handleBlock(msgGeneric protocol.Message) error {
 		return err
 	}
 	// Call the user callback function
-	return b.callbackConfig.BlockFunc(wrappedBlock.Type, blk)
+	return b.config.BlockFunc(wrappedBlock.Type, blk)
 }
 
 func (b *BlockFetch) handleBatchDone() error {
-	if b.callbackConfig.BatchDoneFunc == nil {
+	if b.config.BatchDoneFunc == nil {
 		return fmt.Errorf("received block-fetch BatchDone message but no callback function is defined")
 	}
 	// Call the user callback function
-	return b.callbackConfig.BatchDoneFunc()
+	return b.config.BatchDoneFunc()
 }