-
Notifications
You must be signed in to change notification settings - Fork 649
/
storage.go
164 lines (133 loc) · 4.13 KB
/
storage.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
package jsonrpc
import (
"errors"
"fmt"
"time"
"github.com/0xPolygonHermez/zkevm-node/hex"
"github.com/google/uuid"
"github.com/gorilla/websocket"
)
// ErrNotFound represent a not found error.
var ErrNotFound = errors.New("object not found")
// ErrFilterInvalidPayload indicates there is an invalid payload when creating a filter
var ErrFilterInvalidPayload = errors.New("invalid argument 0: cannot specify both BlockHash and FromBlock/ToBlock, choose one or the other")
// Storage uses memory to store the data
// related to the json rpc server
type Storage struct {
filters map[string]*Filter
}
// NewStorage creates and initializes an instance of Storage
func NewStorage() *Storage {
return &Storage{
filters: make(map[string]*Filter),
}
}
// NewLogFilter persists a new log filter
func (s *Storage) NewLogFilter(wsConn *websocket.Conn, filter LogFilter) (string, error) {
if filter.BlockHash != nil && (filter.FromBlock != nil || filter.ToBlock != nil) {
return "", ErrFilterInvalidPayload
}
return s.createFilter(FilterTypeLog, filter, wsConn)
}
// NewBlockFilter persists a new block log filter
func (s *Storage) NewBlockFilter(wsConn *websocket.Conn) (string, error) {
return s.createFilter(FilterTypeBlock, nil, wsConn)
}
// NewPendingTransactionFilter persists a new pending transaction filter
func (s *Storage) NewPendingTransactionFilter(wsConn *websocket.Conn) (string, error) {
return s.createFilter(FilterTypePendingTx, nil, wsConn)
}
// create persists the filter to the memory and provides the filter id
func (s *Storage) createFilter(t FilterType, parameters interface{}, wsConn *websocket.Conn) (string, error) {
lastPoll := time.Now().UTC()
id, err := s.generateFilterID()
if err != nil {
return "", fmt.Errorf("failed to generate filter ID: %w", err)
}
s.filters[id] = &Filter{
ID: id,
Type: t,
Parameters: parameters,
LastPoll: lastPoll,
WsConn: wsConn,
}
return id, nil
}
func (s *Storage) generateFilterID() (string, error) {
r, err := uuid.NewRandom()
if err != nil {
return "", err
}
b, err := r.MarshalBinary()
if err != nil {
return "", err
}
id := hex.EncodeToHex(b)
return id, nil
}
// GetAllBlockFiltersWithWSConn returns an array with all filter that have
// a web socket connection and are filtering by new blocks
func (s *Storage) GetAllBlockFiltersWithWSConn() ([]*Filter, error) {
filtersWithWSConn := []*Filter{}
for _, filter := range s.filters {
if filter.WsConn == nil || filter.Type != FilterTypeBlock {
continue
}
f := filter
filtersWithWSConn = append(filtersWithWSConn, f)
}
return filtersWithWSConn, nil
}
// GetAllLogFiltersWithWSConn returns an array with all filter that have
// a web socket connection and are filtering by new logs
func (s *Storage) GetAllLogFiltersWithWSConn() ([]*Filter, error) {
filtersWithWSConn := []*Filter{}
for _, filter := range s.filters {
if filter.WsConn == nil || filter.Type != FilterTypeLog {
continue
}
f := filter
filtersWithWSConn = append(filtersWithWSConn, f)
}
return filtersWithWSConn, nil
}
// GetFilter gets a filter by its id
func (s *Storage) GetFilter(filterID string) (*Filter, error) {
filter, found := s.filters[filterID]
if !found {
return nil, ErrNotFound
}
return filter, nil
}
// UpdateFilterLastPoll updates the last poll to now
func (s *Storage) UpdateFilterLastPoll(filterID string) error {
filter, found := s.filters[filterID]
if !found {
return ErrNotFound
}
filter.LastPoll = time.Now().UTC()
s.filters[filterID] = filter
return nil
}
// UninstallFilter deletes a filter by its id
func (s *Storage) UninstallFilter(filterID string) error {
_, found := s.filters[filterID]
if !found {
return ErrNotFound
}
delete(s.filters, filterID)
return nil
}
// UninstallFilterByWSConn deletes all filters connected to the provided web socket connection
func (s *Storage) UninstallFilterByWSConn(wsConn *websocket.Conn) error {
filterIDsToDelete := []string{}
for id, filter := range s.filters {
if filter.WsConn == wsConn {
filterIDsToDelete = append(filterIDsToDelete, id)
}
}
for _, filterID := range filterIDsToDelete {
delete(s.filters, filterID)
}
return nil
}