/
queue.go
120 lines (102 loc) · 3.02 KB
/
queue.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
/*
* Copyright (C) 2021 The "MysteriumNetwork/payments" Authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package transfer
import (
"errors"
"sync"
"github.com/ethereum/go-ethereum/core/types"
)
// Queue is channel queue which is able to
// queue transaction execution funcs.
type Queue struct {
queue chan queueEntry
stop chan struct{}
once sync.Once
}
// ErrQueueClosed is returned when queue is closed and transaction was not processed.
var ErrQueueClosed = errors.New("queue was closed")
// ErrQueueMissingResult is returned if neither an error or a transaction exists in the queue response.
var ErrQueueMissingResult = errors.New("transaction missing with no previous error, state unknown")
type queueEntry struct {
exec TransactionSendFn
resp chan<- QueueResponse
}
// NewQueue returns a new queue. Size for the queue can be given
// so that more than 1 transaction could be queued at a time.
func NewQueue(size uint) *Queue {
return &Queue{
queue: make(chan queueEntry, size),
stop: make(chan struct{}, 0),
}
}
// Run will start to read the queue.
func (q *Queue) Run() {
for {
select {
case <-q.stop:
close(q.queue)
for entry := range q.queue {
q.resp(nil, ErrQueueClosed, entry.resp)
}
return
case entry := <-q.queue:
tx, err := entry.exec()
q.resp(tx, err, entry.resp)
}
}
}
func (q *Queue) resp(tx *types.Transaction, err error, ch chan<- QueueResponse) {
ch <- QueueResponse{
tx: tx,
err: err,
}
close(ch)
}
// Stop will stop the thread. No new transactions can be enqueued after.
func (q *Queue) Stop() {
q.once.Do(func() {
close(q.stop)
})
}
// TransactionEnqueue will enqueue a given transaction and respond on the given resp channel.
//
// Enqueue will fail and instantly return an error if the queue is closed.
// The given `resp` channel should be single use only. After receiving a message that channel will be closed.
func (q *Queue) TransactionEnqueue(exec TransactionSendFn, resp chan<- QueueResponse) error {
select {
case <-q.stop:
// If stop is closed, dont submit new entries
return ErrQueueClosed
default:
q.queue <- queueEntry{
exec: exec,
resp: resp,
}
}
return nil
}
type QueueResponse struct {
tx *types.Transaction
err error
}
// Result extracts the innards of the queue response.
func (qr *QueueResponse) Result() (*types.Transaction, error) {
if qr.tx == nil && qr.err == nil {
return nil, ErrQueueMissingResult
}
return qr.tx, qr.err
}