feat: server-side support for local-tx-monitor (#299)

Fixes #40

localtxmonitor.go

@@ -0,0 +1,44 @@
+// 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 dingo
+
+import (
+	olocaltxmonitor "github.com/blinklabs-io/gouroboros/protocol/localtxmonitor"
+)
+
+const (
+	localtxmonitorMempoolCapacity = 10 * 1024 * 1024 // TODO: replace with configurable value
+)
+
+func (n *Node) localtxmonitorServerConnOpts() []olocaltxmonitor.LocalTxMonitorOptionFunc {
+	return []olocaltxmonitor.LocalTxMonitorOptionFunc{
+		olocaltxmonitor.WithGetMempoolFunc(n.localtxmonitorServerGetMempool),
+	}
+}
+
+func (n *Node) localtxmonitorServerGetMempool(
+	ctx olocaltxmonitor.CallbackContext,
+) (uint64, uint32, []olocaltxmonitor.TxAndEraId, error) {
+	tip := n.ledgerState.Tip()
+	mempoolTxs := n.mempool.Transactions()
+	retTxs := make([]olocaltxmonitor.TxAndEraId, len(mempoolTxs))
+	for i := 0; i < len(mempoolTxs); i++ {
+		retTxs[i] = olocaltxmonitor.TxAndEraId{
+			EraId: mempoolTxs[i].Type,
+			Tx:    mempoolTxs[i].Cbor,
+		}
+	}
+	return tip.Point.Slot, localtxmonitorMempoolCapacity, retTxs, nil
+}

mempool/mempool.go

@@ -255,6 +255,16 @@ func (m *Mempool) GetTransaction(txHash string) (MempoolTransaction, bool) {
 	return *ret, true
 }
 
+func (m *Mempool) Transactions() []MempoolTransaction {
+	m.Lock()
+	defer m.Unlock()
+	ret := make([]MempoolTransaction, len(m.transactions))
+	for i := 0; i < len(m.transactions); i++ {
+		ret[i] = *m.transactions[i]
+	}
+	return ret
+}
+
 func (m *Mempool) getTransaction(txHash string) *MempoolTransaction {
 	for _, tx := range m.transactions {
 		if tx.Hash == txHash {

node.go

@@ -30,6 +30,7 @@ import (
 	ouroboros "github.com/blinklabs-io/gouroboros"
 	oblockfetch "github.com/blinklabs-io/gouroboros/protocol/blockfetch"
 	ochainsync "github.com/blinklabs-io/gouroboros/protocol/chainsync"
+	olocaltxmonitor "github.com/blinklabs-io/gouroboros/protocol/localtxmonitor"
 	olocaltxsubmission "github.com/blinklabs-io/gouroboros/protocol/localtxsubmission"
 	opeersharing "github.com/blinklabs-io/gouroboros/protocol/peersharing"
 	otxsubmission "github.com/blinklabs-io/gouroboros/protocol/txsubmission"
@@ -152,13 +153,17 @@ func (n *Node) configureConnManager() error {
 						n.chainsyncServerConnOpts()...,
 					),
 				),
+				ouroboros.WithLocalTxMonitorConfig(
+					olocaltxmonitor.NewConfig(
+						n.localtxmonitorServerConnOpts()...,
+					),
+				),
 				ouroboros.WithLocalTxSubmissionConfig(
 					olocaltxsubmission.NewConfig(
 						n.localtxsubmissionServerConnOpts()...,
 					),
 				),
 				// TODO: add localstatequery
-				// TODO: add localtxmonitor
 			)
 		} else {
 			// Node-to-node config