mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2025-01-11 05:20:05 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

add pendingTransactionWithBody subscription method (#5675)

Browse Source 
Co-authored-by: dc <dctrlbox@gmail.com>
This commit is contained in:
net.wyman 2022-10-11 23:52:07 +08:00 committed by GitHub
parent ab0ac1bfd4
commit 00276ce7ab
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 44 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
cmd/rpcdaemon/commands/eth_filters.go
View File

@ -159,7 +159,7 @@ func (api *APIImpl) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
return rpcSub, nil
}
// NewPendingTransactions send a notification each time a new (header) block is appended to the chain.
// NewPendingTransactions send a notification each time when a transaction had added into mempool.
func (api *APIImpl) NewPendingTransactions(ctx context.Context) (*rpc.Subscription, error) {
if api.filters == nil {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
@ -202,6 +202,49 @@ func (api *APIImpl) NewPendingTransactions(ctx context.Context) (*rpc.Subscripti
return rpcSub, nil
}
// NewPendingTransactionsWithBody send a notification each time when a transaction had added into mempool.
func (api *APIImpl) NewPendingTransactionsWithBody(ctx context.Context) (*rpc.Subscription, error) {
if api.filters == nil {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
}
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
}
rpcSub := notifier.CreateSubscription()
go func() {
defer debug.LogPanic()
txsCh := make(chan []types.Transaction, 1)
id := api.filters.SubscribePendingTxs(txsCh)
defer api.filters.UnsubscribePendingTxs(id)
for {
select {
case txs, ok := <-txsCh:
for _, t := range txs {
if t != nil {
err := notifier.Notify(rpcSub.ID, t)
if err != nil {
log.Warn("error while notifying subscription", "err", err)
return
}
}
}
if !ok {
log.Warn("new pending transactions channel was closed")
return
}
case <-rpcSub.Err():
return
}
}
}()
return rpcSub, nil
}
// Logs send a notification each time a new log appears.
func (api *APIImpl) Logs(ctx context.Context, crit filters.FilterCriteria) (*rpc.Subscription, error) {
if api.filters == nil {