forked from hyperledger/fabric
-
Notifications
You must be signed in to change notification settings - Fork 0
/
blockcutter.go
105 lines (83 loc) · 2.58 KB
/
blockcutter.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
/*
Copyright IBM Corp. All Rights Reserved.
SPDX-License-Identifier: Apache-2.0
*/
package blockcutter
import (
"sync"
"github.com/hyperledger/fabric/common/flogging"
cb "github.com/hyperledger/fabric/protos/common"
)
var logger = flogging.MustGetLogger("orderer.mocks.common.blockcutter")
// Receiver mocks the blockcutter.Receiver interface
type Receiver struct {
// IsolatedTx causes Ordered returns [][]{curBatch, []{newTx}}, false when set to true
IsolatedTx bool
// CutAncestors causes Ordered returns [][]{curBatch}, true when set to true
CutAncestors bool
// CutNext causes Ordered returns [][]{append(curBatch, newTx)}, false when set to true
CutNext bool
// SkipAppendCurBatch causes Ordered to skip appending to curBatch
SkipAppendCurBatch bool
// Lock to serialize writes access to curBatch
mutex sync.Mutex
// curBatch is the currently outstanding messages in the batch
curBatch []*cb.Envelope
// Block is a channel which is read from before returning from Ordered, it is useful for synchronization
// If you do not wish synchronization for whatever reason, simply close the channel
Block chan struct{}
}
// NewReceiver returns the mock blockcutter.Receiver implementation
func NewReceiver() *Receiver {
return &Receiver{
IsolatedTx: false,
CutAncestors: false,
CutNext: false,
Block: make(chan struct{}),
}
}
// Ordered will add or cut the batch according to the state of Receiver, it blocks reading from Block on return
func (mbc *Receiver) Ordered(env *cb.Envelope) ([][]*cb.Envelope, bool) {
defer func() {
<-mbc.Block
}()
mbc.mutex.Lock()
defer mbc.mutex.Unlock()
if mbc.IsolatedTx {
logger.Debugf("Receiver: Returning dual batch")
res := [][]*cb.Envelope{mbc.curBatch, {env}}
mbc.curBatch = nil
return res, false
}
if mbc.CutAncestors {
logger.Debugf("Receiver: Returning current batch and appending newest env")
res := [][]*cb.Envelope{mbc.curBatch}
mbc.curBatch = []*cb.Envelope{env}
return res, true
}
if !mbc.SkipAppendCurBatch {
mbc.curBatch = append(mbc.curBatch, env)
}
if mbc.CutNext {
logger.Debugf("Receiver: Returning regular batch")
res := [][]*cb.Envelope{mbc.curBatch}
mbc.curBatch = nil
return res, false
}
logger.Debugf("Appending to batch")
return nil, true
}
// Cut terminates the current batch, returning it
func (mbc *Receiver) Cut() []*cb.Envelope {
mbc.mutex.Lock()
defer mbc.mutex.Unlock()
logger.Debugf("Cutting batch")
res := mbc.curBatch
mbc.curBatch = nil
return res
}
func (mbc *Receiver) CurBatch() []*cb.Envelope {
mbc.mutex.Lock()
defer mbc.mutex.Unlock()
return mbc.curBatch
}