logs_filter.go

package rpcfilters

import (
	"context"
	"fmt"
	"math/big"
	"sync"
	"time"

	ethereum "github.com/ethereum/go-ethereum"
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/ethereum/go-ethereum/rpc"
)

type logsFilter struct {
	mu   sync.RWMutex
	logs []types.Log
	crit ethereum.FilterQuery // will be modified and different from original

	originalCrit ethereum.FilterQuery // not modified version of the criteria

	logsCache *cache

	id    rpc.ID
	timer *time.Timer

	ctx    context.Context
	cancel context.CancelFunc
	done   chan struct{}
}

func (f *logsFilter) criteria() ethereum.FilterQuery {
	f.mu.RLock()
	defer f.mu.RUnlock()
	return f.crit
}

func (f *logsFilter) add(data interface{}) error {
	logs, ok := data.([]types.Log)
	if !ok {
		return fmt.Errorf("can't cast %v to types.Log", data)
	}
	filtered := filterLogs(logs, f.crit)
	if len(filtered) > 0 {
		f.mu.Lock()
		defer f.mu.Unlock()
		added, replaced, err := f.logsCache.add(filtered)
		if err != nil {
			return err
		}
		for _, log := range replaced {
			log.Removed = true
			f.logs = append(f.logs, log)
		}
		if len(added) > 0 {
			f.logs = append(f.logs, added...)
		}
		// if there was no replaced logs - keep polling only latest logs
		if len(replaced) == 0 {
			adjustFromBlock(&f.crit)
		} else {
			// otherwise poll earliest known block in cache
			earliest := f.logsCache.earliestBlockNum()
			if earliest != 0 {
				f.crit.FromBlock = new(big.Int).SetUint64(earliest)
			}
		}
	}
	return nil
}

func (f *logsFilter) pop() interface{} {
	f.mu.Lock()
	defer f.mu.Unlock()
	rst := f.logs
	f.logs = nil
	return rst
}

func (f *logsFilter) stop() {
	select {
	case <-f.done:
		return
	default:
		close(f.done)
		if f.cancel != nil {
			f.cancel()
		}
	}
}

func (f *logsFilter) deadline() *time.Timer {
	return f.timer
}

// adjustFromBlock adjusts crit.FromBlock to latest to avoid querying same logs.
func adjustFromBlock(crit *ethereum.FilterQuery) {
	latest := big.NewInt(rpc.LatestBlockNumber.Int64())
	// don't adjust if filter is not interested in newer blocks
	if crit.ToBlock != nil && crit.ToBlock.Cmp(latest) == 1 {
		return
	}
	// don't adjust if from block is already pending
	if crit.FromBlock != nil && crit.FromBlock.Cmp(latest) == -1 {
		return
	}
	crit.FromBlock = latest
}

func includes(addresses []common.Address, a common.Address) bool {
	for _, addr := range addresses {
		if addr == a {
			return true
		}
	}
	return false
}

// filterLogs creates a slice of logs matching the given criteria.
func filterLogs(logs []types.Log, crit ethereum.FilterQuery) (
	ret []types.Log) {
	for _, log := range logs {
		if matchLog(log, crit) {
			ret = append(ret, log)
		}
	}
	return
}

func matchLog(log types.Log, crit ethereum.FilterQuery) bool {
	if crit.FromBlock != nil && crit.FromBlock.Int64() >= 0 && crit.FromBlock.Uint64() > log.BlockNumber {
		return false
	}
	if crit.ToBlock != nil && crit.ToBlock.Int64() >= 0 && crit.ToBlock.Uint64() < log.BlockNumber {
		return false
	}
	if len(crit.Addresses) > 0 && !includes(crit.Addresses, log.Address) {
		return false
	}
	if len(crit.Topics) > len(log.Topics) {
		return false
	}
	return matchTopics(log, crit.Topics)
}

func matchTopics(log types.Log, topics [][]common.Hash) bool {
	for i, sub := range topics {
		match := len(sub) == 0 // empty rule set == wildcard
		for _, topic := range sub {
			if log.Topics[i] == topic {
				match = true
				break
			}
		}
		if !match {
			return false
		}
	}
	return true
}