forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-23 13:16:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

send logs of each block together

Browse source 
Signed-off-by: jsvisa <delweng@gmail.com>
This commit is contained in:
Sina Mahmoodi 2023-08-11 21:31:06 +02:00 committed by jsvisa
parent 3083f77c87
commit b196709758
2 changed files with 15 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
eth/filters/api.go
View file

@ -335,7 +335,7 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
// 2. if a reorg occurs before the currently delivered block, then we need to stop the historical delivery, and send all replaced logs instead // 2. if a reorg occurs before the currently delivered block, then we need to stop the historical delivery, and send all replaced logs instead
var ( var (
liveLogs = make(chan []*types.Log) liveLogs = make(chan []*types.Log)
histLogs = make(chan *types.Log) histLogs = make(chan []*types.Log)
histDone = make(chan error) histDone = make(chan error)
closeC = make(chan struct{}) closeC = make(chan struct{})
) )
@ -394,12 +394,18 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
// Send them all. // Send them all.
notifyLogsIf(notifier, rpcSub.ID, logs, nil) notifyLogsIf(notifier, rpcSub.ID, logs, nil)
} }
case log := <-histLogs: case logs := <-histLogs:
// If ChainReorg is detected, we need to deliver the last block's full logs if len(logs) == 0 {
if !reorged || delivered == log.BlockNumber { continue
delivered = log.BlockNumber }
if reorged {
continue
}
for _, log := range logs {
notifier.Notify(rpcSub.ID, &log) notifier.Notify(rpcSub.ID, &log)
} }
// Assuming batch = all logs of a single block
delivered = logs[0].BlockNumber
case <-rpcSub.Err(): // client send an unsubscribe request case <-rpcSub.Err(): // client send an unsubscribe request
liveLogsSub.Unsubscribe() liveLogsSub.Unsubscribe()
closeC <- struct{}{} closeC <- struct{}{}
@ -416,7 +422,7 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
} }
// doHistLogs retrieves the logs older than current header, and forward them to the histLogs channel. // doHistLogs retrieves the logs older than current header, and forward them to the histLogs channel.
func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, histLogs chan<- *types.Log, closeC chan struct{}) error { func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, histLogs chan<- []*types.Log, closeC chan struct{}) error {
// The original request ctx will be canceled as soon as the parent goroutine // The original request ctx will be canceled as soon as the parent goroutine
// returns a subscription. Use a new context instead. // returns a subscription. Use a new context instead.
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
@ -430,8 +436,8 @@ func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, histLogs chan<
case <-closeC: case <-closeC:
cancel() cancel()
return nil return nil
case log := <-logsCh: case logs := <-logsCh:
histLogs <- log histLogs <- logs
case err := <-errChan: case err := <-errChan:
if err != nil { if err != nil {
logger.Error("Error while fetching historical logs", "err", err) logger.Error("Error while fetching historical logs", "err", err)

2
eth/filters/filter.go
View file

@ -32,7 +32,7 @@ import (
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
) )
// Filter can be used to retrieve and filter logs. // Filter can be used to retrieve and filter historical logs.
type Filter struct { type Filter struct {
sys *FilterSystem sys *FilterSystem