Merge pull request #192 from cloudstruct/feat/local-tx-monitor-protocol

feat: local-tx-monitor protocol

Author: agaffney

.gitignore

@@ -5,6 +5,7 @@
 *.so
 *.dylib
 /go-ouroboros-network
+/tx-monitor
 
 # Test binary, built with `go test -c`
 *.test

Makefile

@@ -5,7 +5,7 @@ ROOT_DIR=$(shell dirname $(realpath $(firstword $(MAKEFILE_LIST))))
 GO_FILES=$(shell find $(ROOT_DIR) -name '*.go')
 
 # Gather list of expected binaries
-BINARIES=$(shell cd $(ROOT_DIR)/cmd && ls -1)
+BINARIES=$(shell cd $(ROOT_DIR)/cmd && ls -1 | grep -v ^common)
 
 .PHONY: build mod-tidy clean test
 

cmd/common/cmdline.go

@@ -0,0 +1,65 @@
+package common
+
+import (
+	"flag"
+	"fmt"
+	"os"
+)
+
+const (
+	TESTNET_MAGIC = 1097911063
+	MAINNET_MAGIC = 764824073
+	PREPROD_MAGIC = 1
+	PREVIEW_MAGIC = 2
+)
+
+type GlobalFlags struct {
+	Flagset      *flag.FlagSet
+	Socket       string
+	Address      string
+	UseTls       bool
+	NtnProto     bool
+	NetworkMagic int
+	Testnet      bool
+	Mainnet      bool
+	Preprod      bool
+	Preview      bool
+}
+
+func NewGlobalFlags() *GlobalFlags {
+	f := &GlobalFlags{
+		Flagset: flag.NewFlagSet(os.Args[0], flag.ExitOnError),
+	}
+	f.Flagset.StringVar(&f.Socket, "socket", "", "UNIX socket path to connect to")
+	f.Flagset.StringVar(&f.Address, "address", "", "TCP address to connect to in address:port format")
+	f.Flagset.BoolVar(&f.UseTls, "tls", false, "enable TLS")
+	f.Flagset.BoolVar(&f.NtnProto, "ntn", false, "use node-to-node protocol (defaults to node-to-client)")
+	f.Flagset.IntVar(&f.NetworkMagic, "network-magic", 0, "network magic value")
+	f.Flagset.BoolVar(&f.Testnet, "testnet", false, fmt.Sprintf("alias for -network-magic=%d", TESTNET_MAGIC))
+	f.Flagset.BoolVar(&f.Mainnet, "mainnet", false, fmt.Sprintf("alias for -network-magic=%d", MAINNET_MAGIC))
+	f.Flagset.BoolVar(&f.Preprod, "preprod", false, fmt.Sprintf("alias for -network-magic=%d", PREPROD_MAGIC))
+	f.Flagset.BoolVar(&f.Preview, "preview", false, fmt.Sprintf("alias for -network-magic=%d", PREVIEW_MAGIC))
+	return f
+}
+
+func (f *GlobalFlags) Parse() {
+	if err := f.Flagset.Parse(os.Args[1:]); err != nil {
+		fmt.Printf("failed to parse command args: %s\n", err)
+		os.Exit(1)
+	}
+	if f.NetworkMagic == 0 {
+		if f.Testnet {
+			f.NetworkMagic = TESTNET_MAGIC
+		} else if f.Mainnet {
+			f.NetworkMagic = MAINNET_MAGIC
+		} else if f.Preprod {
+			f.NetworkMagic = PREPROD_MAGIC
+		} else if f.Preview {
+			f.NetworkMagic = PREVIEW_MAGIC
+		} else {
+			fmt.Printf("You must specify one of -testnet, -mainnet, -preprod, -preview, or -network-magic\n\n")
+			f.Flagset.PrintDefaults()
+			os.Exit(1)
+		}
+	}
+}

cmd/common/connect.go

@@ -0,0 +1,36 @@
+package common
+
+import (
+	"crypto/tls"
+	"fmt"
+	"net"
+	"os"
+)
+
+func CreateClientConnection(f *GlobalFlags) net.Conn {
+	var err error
+	var conn net.Conn
+	var dialProto string
+	var dialAddress string
+	if f.Socket != "" {
+		dialProto = "unix"
+		dialAddress = f.Socket
+	} else if f.Address != "" {
+		dialProto = "tcp"
+		dialAddress = f.Address
+	} else {
+		fmt.Printf("You must specify one of -socket or -address\n\n")
+		f.Flagset.PrintDefaults()
+		os.Exit(1)
+	}
+	if f.UseTls {
+		conn, err = tls.Dial(dialProto, dialAddress, nil)
+	} else {
+		conn, err = net.Dial(dialProto, dialAddress)
+	}
+	if err != nil {
+		fmt.Printf("Connection failed: %s\n", err)
+		os.Exit(1)
+	}
+	return conn
+}

cmd/tx-monitor/main.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"encoding/hex"
+	"fmt"
+	"os"
+
+	"github.com/cloudstruct/go-ouroboros-network"
+	"github.com/cloudstruct/go-ouroboros-network/cmd/common"
+
+	"github.com/fxamacker/cbor/v2"
+	"golang.org/x/crypto/blake2b"
+)
+
+func main() {
+	// Parse commandline
+	f := common.NewGlobalFlags()
+	f.Parse()
+	// Create connection
+	conn := common.CreateClientConnection(f)
+	errorChan := make(chan error)
+	go func() {
+		for {
+			err := <-errorChan
+			fmt.Printf("ERROR(async): %s\n", err)
+			os.Exit(1)
+		}
+	}()
+	o, err := ouroboros.New(
+		ouroboros.WithConnection(conn),
+		ouroboros.WithNetworkMagic(uint32(f.NetworkMagic)),
+		ouroboros.WithErrorChan(errorChan),
+		ouroboros.WithNodeToNode(f.NtnProto),
+		ouroboros.WithKeepAlive(true),
+	)
+	if err != nil {
+		fmt.Printf("ERROR: %s\n", err)
+		os.Exit(1)
+	}
+	o.LocalTxMonitor().Client.Start()
+
+	capacity, size, numberOfTxs, err := o.LocalTxMonitor().Client.GetSizes()
+	if err != nil {
+		fmt.Printf("ERROR(GetSizes): %s\n", err)
+		os.Exit(1)
+	}
+	fmt.Printf("Mempool size/capacity (bytes): %d / %d, TXs: %d\n", size, capacity, numberOfTxs)
+
+	fmt.Printf("Transactions:\n\n")
+	for {
+		tx, err := o.LocalTxMonitor().Client.NextTx()
+		if err != nil {
+			fmt.Printf("ERROR(NextTx): %s\n", err)
+			os.Exit(1)
+		}
+		if tx == nil {
+			break
+		}
+		// Unwrap raw transaction bytes into a CBOR array
+		var txUnwrap []cbor.RawMessage
+		if err := cbor.Unmarshal(tx, &txUnwrap); err != nil {
+			fmt.Printf("ERROR(unwrap): %s\n", err)
+			os.Exit(1)
+		}
+		// index 0 is the transaction body
+		// Store index 0 (transaction body) as byte array
+		txBody := txUnwrap[0]
+		// Convert the body into a blake2b256 hash string
+		txIdHash := blake2b.Sum256(txBody)
+		// Encode hash string as byte array to hex string
+		txIdHex := hex.EncodeToString(txIdHash[:])
+		fmt.Printf("%s\n", txIdHex)
+	}
+}

go.mod

@@ -5,10 +5,10 @@ go 1.17
 require (
 	github.com/cloudstruct/go-cardano-ledger v0.4.0
 	github.com/fxamacker/cbor/v2 v2.4.0
+	golang.org/x/crypto v0.0.0-20220824171710-5757bc0c5503
 )
 
 require (
 	github.com/x448/float16 v0.8.4 // indirect
-	golang.org/x/crypto v0.0.0-20220824171710-5757bc0c5503 // indirect
 	golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1 // indirect
 )

ouroboros.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cloudstruct/go-ouroboros-network/protocol/handshake"
 	"github.com/cloudstruct/go-ouroboros-network/protocol/keepalive"
 	"github.com/cloudstruct/go-ouroboros-network/protocol/localstatequery"
+	"github.com/cloudstruct/go-ouroboros-network/protocol/localtxmonitor"
 	"github.com/cloudstruct/go-ouroboros-network/protocol/localtxsubmission"
 	"github.com/cloudstruct/go-ouroboros-network/protocol/txsubmission"
 )
@@ -50,6 +51,8 @@ type Ouroboros struct {
 	blockFetchConfig        *blockfetch.Config
 	keepAlive               *keepalive.KeepAlive
 	keepAliveConfig         *keepalive.Config
+	localTxMonitor          *localtxmonitor.LocalTxMonitor
+	localTxMonitorConfig    *localtxmonitor.Config
 	localTxSubmission       *localtxsubmission.LocalTxSubmission
 	localTxSubmissionConfig *localtxsubmission.Config
 	localStateQuery         *localstatequery.LocalStateQuery
@@ -168,6 +171,11 @@ func (o *Ouroboros) KeepAlive() *keepalive.KeepAlive {
 	return o.keepAlive
 }
 
+// LocalTxMonitor returns the local-tx-monitor protocol handler
+func (o *Ouroboros) LocalTxMonitor() *localtxmonitor.LocalTxMonitor {
+	return o.localTxMonitor
+}
+
 // LocalTxSubmission returns the local-tx-submission protocol handler
 func (o *Ouroboros) LocalTxSubmission() *localtxsubmission.LocalTxSubmission {
 	return o.localTxSubmission
@@ -297,6 +305,9 @@ func (o *Ouroboros) setupConnection() error {
 		if versionNtC.EnableLocalQueryProtocol {
 			o.localStateQuery = localstatequery.New(protoOptions, o.localStateQueryConfig)
 		}
+		if versionNtC.EnableLocalTxMonitorProtocol {
+			o.localTxMonitor = localtxmonitor.New(protoOptions, o.localTxMonitorConfig)
+		}
 	}
 	// Start muxer
 	diffusionMode := muxer.DiffusionModeInitiator