Skip to content

Commit bafb37f

Browse files
yacovmyacovm
committed
Gossip messageStore- move to separate package
This is part of FAB-872 Gossip MultiChannel support I want the channel handling to be in a different package than gossip, so I need this one to be also outside of gossip otherwise I'll have a dependency cycle. Change-Id: Ibed52855ab5709c9ef2e5fb87dbfd402e366c08a Signed-off-by: Yacov Manevich <yacovm@il.ibm.com>
1 parent 43f4e57 commit bafb37f

File tree

3 files changed

+39
-38
lines changed

3 files changed

+39
-38
lines changed

‎gossip/gossip/gossip_impl.go

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
"github.com/hyperledger/fabric/gossip/comm"
2929
"github.com/hyperledger/fabric/gossip/common"
3030
"github.com/hyperledger/fabric/gossip/discovery"
31+
"github.com/hyperledger/fabric/gossip/gossip/msgstore"
3132
"github.com/hyperledger/fabric/gossip/gossip/pull"
3233
"github.com/hyperledger/fabric/gossip/identity"
3334
"github.com/hyperledger/fabric/gossip/proto"
@@ -55,7 +56,7 @@ type gossipServiceImpl struct {
5556
conf *Config
5657
toDieChan chan struct{}
5758
stopFlag int32
58-
msgStore messageStore
59+
msgStore msgstore.MessageStore
5960
emitter batchingEmitter
6061
goRoutines []uint64
6162
discAdapter *discoveryAdapter
@@ -108,7 +109,7 @@ func NewGossipService(conf *Config, s *grpc.Server, mcs api.MessageCryptoService
108109
Endpoint: conf.SelfEndpoint, PKIid: g.comm.GetPKIid(), Metadata: []byte{},
109110
}, g.discAdapter, g.disSecAdap)
110111

111-
g.msgStore = newMessageStore(proto.NewGossipMessageComparator(g.conf.MaxMessageCountToStore), func(m interface{}) {
112+
g.msgStore = msgstore.NewMessageStore(proto.NewGossipMessageComparator(g.conf.MaxMessageCountToStore), func(m interface{}) {
112113
g.blocksPuller.Remove(m.(*proto.GossipMessage))
113114
})
114115

@@ -145,7 +146,7 @@ func (g *gossipServiceImpl) createBlockPuller() pull.Mediator {
145146
g.logger.Warning("Invalid DataMessage:", dataMsg)
146147
return
147148
}
148-
added := g.msgStore.add(msg)
149+
added := g.msgStore.Add(msg)
149150
// if we can't add the message to the msgStore,
150151
// no point in disseminating it to others...
151152
if !added {
@@ -283,7 +284,7 @@ func (g *gossipServiceImpl) handleMessage(m comm.ReceivedMessage) {
283284
}
284285
}
285286

286-
added := g.msgStore.add(msg)
287+
added := g.msgStore.Add(msg)
287288
if added {
288289
g.emitter.Add(msg)
289290
if dataMsg := m.GetGossipMessage().GetDataMsg(); dataMsg != nil {
@@ -340,7 +341,7 @@ func (g *gossipServiceImpl) gossipBatch(msgs []*proto.GossipMessage) {
340341
func (g *gossipServiceImpl) Gossip(msg *proto.GossipMessage) {
341342
g.logger.Info(msg)
342343
if dataMsg := msg.GetDataMsg(); dataMsg != nil {
343-
g.msgStore.add(msg)
344+
g.msgStore.Add(msg)
344345
g.blocksPuller.Add(msg)
345346
}
346347
g.emitter.Add(msg)

‎gossip/gossip/msgs.go renamed to ‎gossip/gossip/msgstore/msgs.go

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
1414
limitations under the License.
1515
*/
1616

17-
package gossip
17+
package msgstore
1818

1919
import (
2020
"sync"
@@ -27,26 +27,26 @@ import (
2727
// then the invalidation trigger on 0 was called when 1 was added.
2828
type invalidationTrigger func(message interface{})
2929

30-
func newMessageStore(pol common.MessageReplacingPolicy, trigger invalidationTrigger) messageStore {
30+
func NewMessageStore(pol common.MessageReplacingPolicy, trigger invalidationTrigger) MessageStore {
3131
return &messageStoreImpl{pol: pol, lock: &sync.RWMutex{}, messages: make([]*msg, 0), invTrigger: trigger}
3232
}
3333

34-
// messageStore adds messages to an internal buffer.
34+
// MessageStore adds messages to an internal buffer.
3535
// When a message is received, it might:
3636
// - Be added to the buffer
3737
// - Discarded because of some message already in the buffer (invalidated)
3838
// - Make a message already in the buffer to be discarded (invalidates)
3939
// When a message is invalidated, the invalidationTrigger is invoked on that message.
40-
type messageStore interface {
40+
type MessageStore interface {
4141
// add adds a message to the store
4242
// returns true or false whether the message was added to the store
43-
add(msg interface{}) bool
43+
Add(msg interface{}) bool
4444

4545
// size returns the amount of messages in the store
46-
size() int
46+
Size() int
4747

4848
// get returns all messages in the store
49-
get() []interface{}
49+
Get() []interface{}
5050
}
5151

5252
type messageStoreImpl struct {
@@ -61,7 +61,7 @@ type msg struct {
6161
}
6262

6363
// add adds a message to the store
64-
func (s *messageStoreImpl) add(message interface{}) bool {
64+
func (s *messageStoreImpl) Add(message interface{}) bool {
6565
s.lock.Lock()
6666
defer s.lock.Unlock()
6767

@@ -87,14 +87,14 @@ func (s *messageStoreImpl) add(message interface{}) bool {
8787
}
8888

8989
// size returns the amount of messages in the store
90-
func (s *messageStoreImpl) size() int {
90+
func (s *messageStoreImpl) Size() int {
9191
s.lock.RLock()
9292
defer s.lock.RUnlock()
9393
return len(s.messages)
9494
}
9595

9696
// get returns all messages in the store
97-
func (s *messageStoreImpl) get() []interface{} {
97+
func (s *messageStoreImpl) Get() []interface{} {
9898
s.lock.RLock()
9999
defer s.lock.RUnlock()
100100

‎gossip/gossip/msgs_test.go renamed to ‎gossip/gossip/msgstore/msgs_test.go

Lines changed: 23 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
1414
limitations under the License.
1515
*/
1616

17-
package gossip
17+
package msgstore
1818

1919
import (
2020
"math/rand"
@@ -52,24 +52,24 @@ func compareInts(this interface{}, that interface{}) common.InvalidationResult {
5252
}
5353

5454
func TestSize(t *testing.T) {
55-
msgStore := newMessageStore(alwaysNoAction, noopTrigger)
56-
msgStore.add(0)
57-
msgStore.add(1)
58-
msgStore.add(2)
59-
assert.Equal(t, 3, msgStore.size())
55+
msgStore := NewMessageStore(alwaysNoAction, noopTrigger)
56+
msgStore.Add(0)
57+
msgStore.Add(1)
58+
msgStore.Add(2)
59+
assert.Equal(t, 3, msgStore.Size())
6060
}
6161

6262
func TestNewMessagesInvalidates(t *testing.T) {
6363
invalidated := make([]int, 9)
64-
msgStore := newMessageStore(compareInts, func(m interface{}) {
64+
msgStore := NewMessageStore(compareInts, func(m interface{}) {
6565
invalidated = append(invalidated, m.(int))
6666
})
67-
assert.True(t, msgStore.add(0))
67+
assert.True(t, msgStore.Add(0))
6868
for i := 1; i < 10; i++ {
69-
assert.True(t, msgStore.add(i))
69+
assert.True(t, msgStore.Add(i))
7070
assert.Equal(t, i-1, invalidated[len(invalidated)-1])
71-
assert.Equal(t, 1, msgStore.size())
72-
assert.Equal(t, i, msgStore.get()[0].(int))
71+
assert.Equal(t, 1, msgStore.Size())
72+
assert.Equal(t, i, msgStore.Get()[0].(int))
7373
}
7474
}
7575

@@ -83,33 +83,33 @@ func TestMessagesGet(t *testing.T) {
8383
return false
8484
}
8585

86-
msgStore := newMessageStore(alwaysNoAction, noopTrigger)
86+
msgStore := NewMessageStore(alwaysNoAction, noopTrigger)
8787
expected := []int{}
8888
for i := 0; i < 2; i++ {
8989
n := rand.Int()
9090
expected = append(expected, n)
91-
msgStore.add(n)
91+
msgStore.Add(n)
9292
}
9393

9494
for _, num2Search := range expected {
95-
assert.True(t, contains(msgStore.get(), num2Search), "Value %v not found in array", num2Search)
95+
assert.True(t, contains(msgStore.Get(), num2Search), "Value %v not found in array", num2Search)
9696
}
9797

9898
}
9999

100100
func TestNewMessagesInvalidated(t *testing.T) {
101-
msgStore := newMessageStore(compareInts, noopTrigger)
102-
assert.True(t, msgStore.add(10))
101+
msgStore := NewMessageStore(compareInts, noopTrigger)
102+
assert.True(t, msgStore.Add(10))
103103
for i := 9; i >= 0; i-- {
104-
assert.False(t, msgStore.add(i))
105-
assert.Equal(t, 1, msgStore.size())
106-
assert.Equal(t, 10, msgStore.get()[0].(int))
104+
assert.False(t, msgStore.Add(i))
105+
assert.Equal(t, 1, msgStore.Size())
106+
assert.Equal(t, 10, msgStore.Get()[0].(int))
107107
}
108108
}
109109

110110
func TestConcurrency(t *testing.T) {
111111
stopFlag := int32(0)
112-
msgStore := newMessageStore(compareInts, noopTrigger)
112+
msgStore := NewMessageStore(compareInts, noopTrigger)
113113
looper := func(f func()) func() {
114114
return func() {
115115
for {
@@ -122,15 +122,15 @@ func TestConcurrency(t *testing.T) {
122122
}
123123

124124
addProcess := looper(func() {
125-
msgStore.add(rand.Int())
125+
msgStore.Add(rand.Int())
126126
})
127127

128128
getProcess := looper(func() {
129-
msgStore.get()
129+
msgStore.Get()
130130
})
131131

132132
sizeProcess := looper(func() {
133-
msgStore.size()
133+
msgStore.Size()
134134
})
135135

136136
go addProcess()

0 commit comments

Comments
 (0)