feat!: provide context object for callback functions

cmd/gouroboros/chainsync.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -256,12 +256,13 @@ func testChainSync(f *globalFlags) {
 	select {}
 }
 
-func chainSyncRollBackwardHandler(point common.Point, tip chainsync.Tip) error {
+func chainSyncRollBackwardHandler(ctx chainsync.CallbackContext, point common.Point, tip chainsync.Tip) error {
 	fmt.Printf("roll backward: point = %#v, tip = %#v\n", point, tip)
 	return nil
 }
 
 func chainSyncRollForwardHandler(
+	ctx chainsync.CallbackContext,
 	blockType uint,
 	blockData interface{},
 	tip chainsync.Tip,
@@ -312,7 +313,7 @@ func chainSyncRollForwardHandler(
 	return nil
 }
 
-func blockFetchBlockHandler(blockData ledger.Block) error {
+func blockFetchBlockHandler(ctx blockfetch.CallbackContext, blockData ledger.Block) error {
 	switch block := blockData.(type) {
 	case *ledger.ByronEpochBoundaryBlock:
 		fmt.Printf("era = Byron (EBB), epoch = %d, slot = %d, id = %s\n", block.Header.ConsensusData.Epoch, block.SlotNumber(), block.Hash())

cmd/tx-submission/main.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -145,6 +145,7 @@ func main() {
 }
 
 func handleRequestTxIds(
+	ctx txsubmission.CallbackContext,
 	blocking bool,
 	ack uint16,
 	req uint16,
@@ -167,6 +168,7 @@ func handleRequestTxIds(
 }
 
 func handleRequestTxs(
+	ctx txsubmission.CallbackContext,
 	txIds []txsubmission.TxId,
 ) ([]txsubmission.TxBody, error) {
 	ret := []txsubmission.TxBody{

connection.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/blinklabs-io/gouroboros/connection"
 	"github.com/blinklabs-io/gouroboros/muxer"
 	"github.com/blinklabs-io/gouroboros/protocol"
 	"github.com/blinklabs-io/gouroboros/protocol/blockfetch"
@@ -49,6 +50,8 @@ const (
 	DefaultConnectTimeout = 30 * time.Second
 )
 
+type ConnectionId = connection.ConnectionId
+
 // The Connection type is a wrapper around a net.Conn object that handles communication using the Ouroboros network protocol over that connection
 type Connection struct {
 	id                    ConnectionId
@@ -90,19 +93,6 @@ type Connection struct {
 	txSubmissionConfig      *txsubmission.Config
 }
 
-type ConnectionId struct {
-	LocalAddr  net.Addr
-	RemoteAddr net.Addr
-}
-
-func (c ConnectionId) String() string {
-	return fmt.Sprintf(
-		"%s<->%s",
-		c.LocalAddr.String(),
-		c.RemoteAddr.String(),
-	)
-}
-
 // NewConnection returns a new Connection object with the specified options. If a connection is provided, the
 // handshake will be started. An error will be returned if the handshake fails
 func NewConnection(options ...ConnectionOptionFunc) (*Connection, error) {
@@ -289,8 +279,9 @@ func (c *Connection) setupConnection() error {
 		}
 	}()
 	protoOptions := protocol.ProtocolOptions{
-		Muxer:     c.muxer,
-		ErrorChan: c.protoErrorChan,
+		ConnectionId: c.id,
+		Muxer:        c.muxer,
+		ErrorChan:    c.protoErrorChan,
 	}
 	if c.useNodeToNodeProto {
 		protoOptions.Mode = protocol.ProtocolModeNodeToNode
@@ -319,7 +310,7 @@ func (c *Connection) setupConnection() error {
 	var handshakeFullDuplex bool
 	handshakeConfig := handshake.NewConfig(
 		handshake.WithProtocolVersionMap(protoVersions),
-		handshake.WithFinishedFunc(func(version uint16, versionData protocol.VersionData) error {
+		handshake.WithFinishedFunc(func(ctx handshake.CallbackContext, version uint16, versionData protocol.VersionData) error {
 			c.handshakeVersion = version
 			c.handshakeVersionData = versionData
 			if c.useNodeToNodeProto {

connection/id.go

@@ -0,0 +1,33 @@
+// Copyright 2024 Blink Labs Software
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connection
+
+import (
+	"fmt"
+	"net"
+)
+
+type ConnectionId struct {
+	LocalAddr  net.Addr
+	RemoteAddr net.Addr
+}
+
+func (c ConnectionId) String() string {
+	return fmt.Sprintf(
+		"%s<->%s",
+		c.LocalAddr.String(),
+		c.RemoteAddr.String(),
+	)
+}

protocol/blockfetch/blockfetch.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@ package blockfetch
 import (
 	"time"
 
+	"github.com/blinklabs-io/gouroboros/connection"
 	"github.com/blinklabs-io/gouroboros/protocol"
 	"github.com/blinklabs-io/gouroboros/protocol/common"
 
@@ -92,9 +93,16 @@ type Config struct {
 	BlockTimeout      time.Duration
 }
 
+// Callback context
+type CallbackContext struct {
+	ConnectionId connection.ConnectionId
+	Client       *Client
+	Server       *Server
+}
+
 // Callback function types
-type BlockFunc func(ledger.Block) error
-type RequestRangeFunc func(common.Point, common.Point) error
+type BlockFunc func(CallbackContext, ledger.Block) error
+type RequestRangeFunc func(CallbackContext, common.Point, common.Point) error
 
 func New(protoOptions protocol.ProtocolOptions, cfg *Config) *BlockFetch {
 	b := &BlockFetch{

protocol/blockfetch/client.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@ import (
 type Client struct {
 	*protocol.Protocol
 	config               *Config
+	callbackContext      CallbackContext
 	blockChan            chan ledger.Block
 	startBatchResultChan chan error
 	busyMutex            sync.Mutex
@@ -46,6 +47,10 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		blockChan:            make(chan ledger.Block),
 		startBatchResultChan: make(chan error),
 	}
+	c.callbackContext = CallbackContext{
+		Client:       c,
+		ConnectionId: protoOptions.ConnectionId,
+	}
 	// Update state map with timeouts
 	stateMap := StateMap.Copy()
 	if entry, ok := stateMap[StateBusy]; ok {
@@ -186,7 +191,7 @@ func (c *Client) handleBlock(msgGeneric protocol.Message) error {
 	}
 	// We use the callback when requesting ranges and the internal channel for a single block
 	if c.blockUseCallback {
-		if err := c.config.BlockFunc(blk); err != nil {
+		if err := c.config.BlockFunc(c.callbackContext, blk); err != nil {
 			return err
 		}
 	} else {

protocol/blockfetch/server.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -23,13 +23,18 @@ import (
 
 type Server struct {
 	*protocol.Protocol
-	config *Config
+	config          *Config
+	callbackContext CallbackContext
 }
 
 func NewServer(protoOptions protocol.ProtocolOptions, cfg *Config) *Server {
 	s := &Server{
 		config: cfg,
 	}
+	s.callbackContext = CallbackContext{
+		Server:       s,
+		ConnectionId: protoOptions.ConnectionId,
+	}
 	protoConfig := protocol.ProtocolConfig{
 		Name:                ProtocolName,
 		ProtocolId:          ProtocolId,
@@ -98,7 +103,7 @@ func (s *Server) handleRequestRange(msg protocol.Message) error {
 		)
 	}
 	msgRequestRange := msg.(*MsgRequestRange)
-	return s.config.RequestRangeFunc(msgRequestRange.Start, msgRequestRange.End)
+	return s.config.RequestRangeFunc(s.callbackContext, msgRequestRange.Start, msgRequestRange.End)
 }
 
 func (s *Server) handleClientDone() error {

protocol/chainsync/chainsync.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package chainsync
 import (
 	"time"
 
+	"github.com/blinklabs-io/gouroboros/connection"
 	"github.com/blinklabs-io/gouroboros/protocol"
 	"github.com/blinklabs-io/gouroboros/protocol/common"
 )
@@ -121,11 +122,18 @@ type Config struct {
 	PipelineLimit     int
 }
 
+// Callback context
+type CallbackContext struct {
+	ConnectionId connection.ConnectionId
+	Client       *Client
+	Server       *Server
+}
+
 // Callback function types
-type RollBackwardFunc func(common.Point, Tip) error
-type RollForwardFunc func(uint, interface{}, Tip) error
-type FindIntersectFunc func([]common.Point) (common.Point, Tip, error)
-type RequestNextFunc func() error
+type RollBackwardFunc func(CallbackContext, common.Point, Tip) error
+type RollForwardFunc func(CallbackContext, uint, interface{}, Tip) error
+type FindIntersectFunc func(CallbackContext, []common.Point) (common.Point, Tip, error)
+type RequestNextFunc func(CallbackContext) error
 
 // New returns a new ChainSync object
 func New(protoOptions protocol.ProtocolOptions, cfg *Config) *ChainSync {

protocol/chainsync/client.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@ import (
 type Client struct {
 	*protocol.Protocol
 	config                *Config
+	callbackContext       CallbackContext
 	busyMutex             sync.Mutex
 	intersectResultChan   chan error
 	readyForNextBlockChan chan bool
@@ -63,6 +64,10 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		firstBlockChan:        make(chan common.Point),
 		intersectPointChan:    make(chan common.Point),
 	}
+	c.callbackContext = CallbackContext{
+		Client:       c,
+		ConnectionId: protoOptions.ConnectionId,
+	}
 	// Update state map with timeouts
 	stateMap := StateMap.Copy()
 	if entry, ok := stateMap[stateIntersect]; ok {
@@ -343,7 +348,7 @@ func (c *Client) handleRollForward(msgGeneric protocol.Message) error {
 			return nil
 		}
 		// Call the user callback function
-		callbackErr = c.config.RollForwardFunc(blockType, blockHeader, msg.Tip)
+		callbackErr = c.config.RollForwardFunc(c.callbackContext, blockType, blockHeader, msg.Tip)
 	} else {
 		msg := msgGeneric.(*MsgRollForwardNtC)
 		blk, err := ledger.NewBlockFromCbor(msg.BlockType(), msg.BlockCbor())
@@ -360,7 +365,7 @@ func (c *Client) handleRollForward(msgGeneric protocol.Message) error {
 			return nil
 		}
 		// Call the user callback function
-		callbackErr = c.config.RollForwardFunc(msg.BlockType(), blk, msg.Tip)
+		callbackErr = c.config.RollForwardFunc(c.callbackContext, msg.BlockType(), blk, msg.Tip)
 	}
 	if callbackErr != nil {
 		if callbackErr == StopSyncProcessError {
@@ -388,7 +393,7 @@ func (c *Client) handleRollBackward(msg protocol.Message) error {
 			)
 		}
 		// Call the user callback function
-		if callbackErr := c.config.RollBackwardFunc(msgRollBackward.Point, msgRollBackward.Tip); callbackErr != nil {
+		if callbackErr := c.config.RollBackwardFunc(c.callbackContext, msgRollBackward.Point, msgRollBackward.Tip); callbackErr != nil {
 			if callbackErr == StopSyncProcessError {
 				// Signal that we're cancelling the sync
 				c.readyForNextBlockChan <- false

protocol/chainsync/server.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2024 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -25,7 +25,8 @@ import (
 // Server implements the ChainSync server
 type Server struct {
 	*protocol.Protocol
-	config *Config
+	config          *Config
+	callbackContext CallbackContext
 }
 
 // NewServer returns a new ChainSync server object
@@ -41,6 +42,10 @@ func NewServer(protoOptions protocol.ProtocolOptions, cfg *Config) *Server {
 	s := &Server{
 		config: cfg,
 	}
+	s.callbackContext = CallbackContext{
+		Server:       s,
+		ConnectionId: protoOptions.ConnectionId,
+	}
 	protoConfig := protocol.ProtocolConfig{
 		Name:                ProtocolName,
 		ProtocolId:          ProtocolId,
@@ -112,7 +117,7 @@ func (s *Server) handleRequestNext(msg protocol.Message) error {
 			"received chain-sync RequestNext message but no callback function is defined",
 		)
 	}
-	return s.config.RequestNextFunc()
+	return s.config.RequestNextFunc(s.callbackContext)
 }
 
 func (s *Server) handleFindIntersect(msg protocol.Message) error {
@@ -122,7 +127,7 @@ func (s *Server) handleFindIntersect(msg protocol.Message) error {
 		)
 	}
 	msgFindIntersect := msg.(*MsgFindIntersect)
-	point, tip, err := s.config.FindIntersectFunc(msgFindIntersect.Points)
+	point, tip, err := s.config.FindIntersectFunc(s.callbackContext, msgFindIntersect.Points)
 	if err != nil {
 		if err == IntersectNotFoundError {
 			msgResp := NewMsgIntersectNotFound(tip)