integration/rpctest: improve test harness

btcd.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/btcsuite/btcd/blockchain/indexers"
 	"github.com/btcsuite/btcd/database"
+	"github.com/btcsuite/btcd/debugstream"
 	"github.com/btcsuite/btcd/limits"
 	"github.com/btcsuite/btcd/ossec"
 )
@@ -56,6 +57,24 @@ func btcdMain(serverChan chan<- *server) error {
 		}
 	}()
 
+	// DebugStream is enabled only if btcd is compiled with the debug tag.
+	// Otherwise a nop implementation is used.
+	debugstream.S = debugstream.New()
+	if dsListen := cfg.DebugStream; dsListen != "" {
+		err := debugstream.S.Listen(dsListen)
+		if err != nil {
+			return fmt.Errorf("error starting debug stream: %v",
+				err)
+		}
+		debugstream.S.Broadcast(debugstream.Event{
+			Code: debugstream.DEStart,
+		})
+		defer debugstream.S.Shutdown()
+		defer debugstream.S.Broadcast(debugstream.Event{
+			Code: debugstream.DEShutdown,
+		})
+	}
+
 	// Get a channel that will be closed when a shutdown signal has been
 	// triggered either from an OS signal such as SIGINT (Ctrl+C) or from
 	// another subsystem such as the RPC server.

config.go

@@ -120,6 +120,7 @@ type config struct {
 	DataDir              string        `short:"b" long:"datadir" description:"Directory to store data"`
 	DbType               string        `long:"dbtype" description:"Database backend to use for the Block Chain"`
 	DebugLevel           string        `short:"d" long:"debuglevel" description:"Logging level for all subsystems {trace, debug, info, warn, error, critical} -- You may also specify <subsystem>=<level>,<subsystem2>=<level>,... to set the log level for individual subsystems -- Use show to list available subsystems"`
+	DebugStream          string        `long:"debugstream" hidden:"true" description:"TCP listen address of the debug stream. To use this feature btcd must also be compiled with debug tag."`
 	DropAddrIndex        bool          `long:"dropaddrindex" description:"Deletes the address-based transaction index from the database on start up and then exits."`
 	DropCfIndex          bool          `long:"dropcfindex" description:"Deletes the index used for committed filtering (CF) support from the database on start up and then exits."`
 	DropTxIndex          bool          `long:"droptxindex" description:"Deletes the hash-based transaction index from the database on start up and then exits."`

debugstream/client.go

@@ -0,0 +1,113 @@
+package debugstream
+
+import (
+	"fmt"
+	"net"
+	"time"
+)
+
+// Client calls handler with events coming from the [StreamServer].
+type Client struct {
+	addr string
+
+	conn    *net.TCPConn
+	handler func(e Event)
+
+	stop chan struct{}
+	done chan struct{}
+}
+
+// NewClient initializes a [Client] instance.
+func NewClient(streamAddr string, handler func(ev Event)) *Client {
+	done := make(chan struct{})
+	close(done)
+	stop := make(chan struct{})
+	close(stop)
+	return &Client{
+		addr:    streamAddr,
+		handler: handler,
+		done:    done,
+		stop:    stop,
+	}
+}
+
+func (c *Client) connect() (*net.TCPConn, error) {
+	const maxConnAttempts = 7
+	var (
+		conn net.Conn
+		err  error
+	)
+
+	for i := range maxConnAttempts {
+		if i > 0 {
+			time.Sleep((time.Millisecond * 100) << (i - 1))
+		}
+		conn, err = net.Dial("tcp", c.addr)
+		if err != nil {
+			continue
+		}
+		break
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return conn.(*net.TCPConn), nil
+}
+
+func (c *Client) loop() {
+	for {
+		var ev Event
+		err := ev.read(c.conn)
+		if err == nil {
+			c.handler(ev)
+			continue
+		}
+		select {
+		case <-c.stop:
+			return
+		default:
+			cliLog.Errorf("client loop %s", err)
+			return
+		}
+	}
+}
+
+func (c *Client) Start() error {
+	select {
+	case <-c.done:
+		c.done = make(chan struct{})
+	default:
+		return fmt.Errorf("client running")
+	}
+
+	conn, err := c.connect()
+	if err != nil {
+		close(c.done)
+		return err
+	}
+
+	c.conn = conn
+	c.stop = make(chan struct{})
+
+	go func() {
+		c.loop()
+		close(c.done)
+	}()
+
+	return nil
+}
+
+func (c *Client) Stop() {
+	select {
+	case <-c.stop:
+		return
+	default:
+	}
+
+	close(c.stop)
+	c.conn.Close()
+	<-c.done
+
+	c.conn = nil
+}

debugstream/debug.go

@@ -0,0 +1,11 @@
+//go:build debug
+
+package debugstream
+
+// Stream is the real implementation of the debug stream, [StreamServer] when
+// compiling with debug tag.
+type Stream = StreamServer
+
+func New() *Stream {
+	return NewStreamServer()
+}

debugstream/event.go

@@ -0,0 +1,61 @@
+package debugstream
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+)
+
+// The following are guard event codes, that can be used to assert debug events
+// in tests.
+const (
+	DEStart = iota + 1
+	DEShutdown
+)
+
+// Event is the message that is sent from the Stream to the Client.
+type Event struct {
+	Code uint64
+	Data []byte
+}
+
+func (e *Event) write(w io.Writer) error {
+	err := binary.Write(w, binary.BigEndian, e.Code)
+	if err != nil {
+		return err
+	}
+	dataLen := uint64(len(e.Data))
+	err = binary.Write(w, binary.BigEndian, dataLen)
+	if err != nil {
+		return err
+	}
+	n, err := w.Write(e.Data)
+	if err != nil {
+		return err
+	}
+	if n != int(dataLen) {
+		return fmt.Errorf("nWrite != dataLen")
+	}
+	return nil
+}
+
+func (e *Event) read(r io.Reader) error {
+	err := binary.Read(r, binary.BigEndian, &e.Code)
+	if err != nil {
+		return err
+	}
+	var dataLen uint64
+	err = binary.Read(r, binary.BigEndian, &dataLen)
+	if err != nil {
+		return err
+	}
+	e.Data = make([]byte, dataLen)
+	n, err := io.ReadFull(r, e.Data)
+	if err != nil {
+		return err
+	}
+	if n != int(dataLen) {
+		return fmt.Errorf("nRead != dataLen")
+	}
+	return nil
+}

debugstream/log.go

@@ -0,0 +1,23 @@
+package debugstream
+
+import "github.com/btcsuite/btclog"
+
+var strLog btclog.Logger
+var cliLog btclog.Logger
+
+func init() {
+	DisableLog()
+}
+
+// DisableLog disables all library log output.  Logging output is disabled
+// by default until either UseLogger or SetLogWriter are called.
+func DisableLog() {
+	strLog, cliLog = btclog.Disabled, btclog.Disabled
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLoggers(strLogger, cliLogger btclog.Logger) {
+	strLog, cliLog = strLogger, cliLogger
+}

debugstream/nop.go

@@ -0,0 +1,11 @@
+//go:build !debug
+
+package debugstream
+
+// Stream is a [StreamNOP] when debug flag is not set, effectively doing nothing
+// because all its procedures are also NOP.
+type Stream = StreamNOP
+
+func New() *Stream {
+	return NewStreamNOP()
+}