Merge pull request #590 from libotony/subscribe-txpool

add pending tx subscription API

api/api.go

@@ -79,7 +79,7 @@ func New(
  Mount(router, "/debug")
  node.New(nw).
  Mount(router, "/node")
- subs := subscriptions.New(repo, origins, backtraceLimit)
+ subs := subscriptions.New(repo, origins, backtraceLimit, txPool)
  subs.Mount(router, "/subscriptions")
 
  if pprofOn {

api/subscriptions/pending_tx.go

@@ -0,0 +1,70 @@
+// Copyright (c) 2023 The VeChainThor developers
+
+// Distributed under the GNU Lesser General Public License v3.0 software license, see the accompanying
+// file LICENSE or <https://www.gnu.org/licenses/lgpl-3.0.html>
+package subscriptions
+
+import (
+ "sync"
+
+ "github.com/vechain/thor/tx"
+ "github.com/vechain/thor/txpool"
+)
+
+type pendingTx struct {
+ txPool *txpool.TxPool
+ listeners map[chan *tx.Transaction]struct{}
+ mu sync.RWMutex
+}
+
+func newPendingTx(txPool *txpool.TxPool) *pendingTx {
+ p := &pendingTx{
+ txPool: txPool,
+ listeners: make(map[chan *tx.Transaction]struct{}),
+ }
+
+ return p
+}
+
+func (p *pendingTx) Subscribe(ch chan *tx.Transaction) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ p.listeners[ch] = struct{}{}
+}
+
+func (p *pendingTx) Unsubscribe(ch chan *tx.Transaction) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ delete(p.listeners, ch)
+}
+
+func (p *pendingTx) DispatchLoop(done <-chan struct{}) {
+ txCh := make(chan *txpool.TxEvent)
+ sub := p.txPool.SubscribeTxEvent(txCh)
+ defer sub.Unsubscribe()
+
+ for {
+ select {
+ case txEv := <-txCh:
+ if txEv.Executable == nil || !*txEv.Executable {
+ continue
+ }
+ p.mu.RLock()
+ func() {
+ for lsn := range p.listeners {
+ select {
+ case lsn <- txEv.Tx:
+ case <-done:
+ return
+ default: // broadcast in a non-blocking manner, so there's no guarantee that all subscriber receives it
+ }
+ }
+ }()
+ p.mu.RUnlock()
+ case <-done:
+ return
+ }
+ }
+}

api/subscriptions/subscriptions.go

@@ -18,12 +18,17 @@ import (
  "github.com/vechain/thor/block"
  "github.com/vechain/thor/chain"
  "github.com/vechain/thor/thor"
+ "github.com/vechain/thor/tx"
+ "github.com/vechain/thor/txpool"
 )
 
+const txQueueSize = 20
+
 type Subscriptions struct {
  backtraceLimit uint32
  repo *chain.Repository
  upgrader *websocket.Upgrader
+ pendingTx *pendingTx
  done chan struct{}
  wg sync.WaitGroup
 }
@@ -43,8 +48,8 @@ const (
  pingPeriod = (pongWait * 7) / 10
 )
 
-func New(repo *chain.Repository, allowedOrigins []string, backtraceLimit uint32) *Subscriptions {
- return &Subscriptions{
+func New(repo *chain.Repository, allowedOrigins []string, backtraceLimit uint32, txpool *txpool.TxPool) *Subscriptions {
+ sub := &Subscriptions{
  backtraceLimit: backtraceLimit,
  repo: repo,
  upgrader: &websocket.Upgrader{
@@ -62,8 +67,17 @@ func New(repo *chain.Repository, allowedOrigins []string, backtraceLimit uint32)
  return false
  },
  },
- done: make(chan struct{}),
+ pendingTx: newPendingTx(txpool),
+ done: make(chan struct{}),
  }
+
+ sub.wg.Add(1)
+ go func() {
+ defer sub.wg.Done()
+
+ sub.pendingTx.DispatchLoop(sub.done)
+ }()
+ return sub
 }
 
 func (s *Subscriptions) handleBlockReader(w http.ResponseWriter, req *http.Request) (*blockReader, error) {
@@ -188,33 +202,63 @@ func (s *Subscriptions) handleSubject(w http.ResponseWriter, req *http.Request)
  return utils.HTTPError(errors.New("not found"), http.StatusNotFound)
  }
 
- conn, err := s.upgrader.Upgrade(w, req, nil)
+ conn, closed, err := s.setupConn(w, req)
+ // since the conn is hijacked here, no error should be returned in lines below
+ if err != nil {
+ log.Debug("upgrade to websocket", "err", err)
+ return nil
+ }
+
+ err = s.pipe(conn, reader, closed)
+ s.closeConn(conn, err)
+ return nil
+}
+
+func (s *Subscriptions) handlePendingTransactions(w http.ResponseWriter, req *http.Request) error {
+ s.wg.Add(1)
+ defer s.wg.Done()
+
+ conn, closed, err := s.setupConn(w, req)
  // since the conn is hijacked here, no error should be returned in lines below
  if err != nil {
  log.Debug("upgrade to websocket", "err", err)
  return nil
  }
+ defer s.closeConn(conn, err)
+
+ pingTicker := time.NewTicker(pingPeriod)
+ defer pingTicker.Stop()
 
+ txCh := make(chan *tx.Transaction, txQueueSize)
+ s.pendingTx.Subscribe(txCh)
  defer func() {
- if err := conn.Close(); err != nil {
- log.Debug("close websocket", "err", err)
- }
+ s.pendingTx.Unsubscribe(txCh)
+ close(txCh)
  }()
 
- var closeMsg []byte
- if err := s.pipe(conn, reader); err != nil {
- closeMsg = websocket.FormatCloseMessage(websocket.CloseInternalServerErr, err.Error())
- } else {
- closeMsg = websocket.FormatCloseMessage(websocket.CloseGoingAway, "")
+ for {
+ select {
+ case tx := <-txCh:
+ err = conn.WriteJSON(&PendingTxIDMessage{ID: tx.ID()})
+ if err != nil {
+ return nil
+ }
+ case <-s.done:
+ return nil
+ case <-closed:
+ return nil
+ case <-pingTicker.C:
+ conn.WriteMessage(websocket.PingMessage, nil)
+ }
  }
+}
 
- if err := conn.WriteMessage(websocket.CloseMessage, closeMsg); err != nil {
- log.Debug("write close message", "err", err)
+func (s *Subscriptions) setupConn(w http.ResponseWriter, req *http.Request) (*websocket.Conn, chan struct{}, error) {
+ conn, err := s.upgrader.Upgrade(w, req, nil)
+ if err != nil {
+ return nil, nil, err
  }
- return nil
-}
 
-func (s *Subscriptions) pipe(conn *websocket.Conn, reader msgReader) error {
  closed := make(chan struct{})
  // start read loop to handle close event
  s.wg.Add(1)
@@ -233,6 +277,28 @@ func (s *Subscriptions) pipe(conn *websocket.Conn, reader msgReader) error {
  }
  }
  }()
+
+ return conn, closed, nil
+}
+
+func (s *Subscriptions) closeConn(conn *websocket.Conn, err error) {
+ var closeMsg []byte
+ if err != nil {
+ closeMsg = websocket.FormatCloseMessage(websocket.CloseInternalServerErr, err.Error())
+ } else {
+ closeMsg = websocket.FormatCloseMessage(websocket.CloseGoingAway, "")
+ }
+
+ if err := conn.WriteMessage(websocket.CloseMessage, closeMsg); err != nil {
+ log.Debug("write close message", "err", err)
+ }
+
+ if err := conn.Close(); err != nil {
+ log.Debug("close websocket", "err", err)
+ }
+}
+
+func (s *Subscriptions) pipe(conn *websocket.Conn, reader msgReader, closed chan struct{}) error {
  ticker := s.repo.NewTicker()
  pingTicker := time.NewTicker(pingPeriod)
  defer pingTicker.Stop()
@@ -316,4 +382,5 @@ func (s *Subscriptions) Mount(root *mux.Router, pathPrefix string) {
  sub := root.PathPrefix(pathPrefix).Subrouter()
 
  sub.Path("/{subject}").Methods("Get").HandlerFunc(utils.WrapHandlerFunc(s.handleSubject))
+ sub.Path("/txpool/pending").Methods("Get").HandlerFunc(utils.WrapHandlerFunc(s.handlePendingTransactions))
 }

api/subscriptions/types.go

@@ -219,3 +219,7 @@ type Beat2Message struct {
  K uint8 `json:"k"`
  Obsolete bool `json:"obsolete"`
 }
+
+type PendingTxIDMessage struct {
+ ID thor.Bytes32 `json:"id"`
+}

api/utils/http.go

@@ -69,8 +69,7 @@ func WrapHandlerFunc(f HandlerFunc) http.HandlerFunc {
 
 // content types
 const (
- JSONContentType = "application/json; charset=utf-8"
- OctetStreamContentType = "application/octet-stream"
+ JSONContentType = "application/json; charset=utf-8"
 )
 
 // ParseJSON parse a JSON object using strict mode.