Skip to content

Commit 7703c81

Browse files
C0rWinC0rWin
committed
Extract common gossip data type
Right now PKIidType data type separetely placed in two different packages, while it serves exactly same purpose and for one who would like to use both dicsovery and communication module from the gossip it will require to make needless convertations. This commits converges and extracts that data type into the common place so it could be used by both packages. Change-Id: Iedfa689e745a472cdb39c70b1ff99cc7880ff91d Signed-off-by: Artem Barger <[email protected]>
1 parent 148322b commit 7703c81

12 files changed

+105
-82
lines changed

‎gossip/comm/comm.go

+7-11
Original file line numberDiff line numberDiff line change
@@ -19,29 +19,29 @@ package comm
1919
import (
2020
"fmt"
2121

22+
"github.com/hyperledger/fabric/gossip/common"
2223
"github.com/hyperledger/fabric/gossip/proto"
23-
"github.com/hyperledger/fabric/gossip/util"
2424
)
2525

2626
// Comm is an object that enables to communicate with other peers
2727
// that also embed a CommModule.
2828
type Comm interface {
2929

3030
// GetPKIid returns this instance's PKI id
31-
GetPKIid() PKIidType
31+
GetPKIid() common.PKIidType
3232

3333
// Send sends a message to remote peers
3434
Send(msg *proto.GossipMessage, peers ...*RemotePeer)
3535

3636
// Probe probes a remote node and returns nil if its responsive
37-
Probe(endpoint string, pkiID PKIidType) error
37+
Probe(endpoint string, pkiID common.PKIidType) error
3838

3939
// Accept returns a dedicated read-only channel for messages sent by other nodes that match a certain predicate.
4040
// Each message from the channel can be used to send a reply back to the sender
41-
Accept(util.MessageAcceptor) <-chan ReceivedMessage
41+
Accept(common.MessageAcceptor) <-chan ReceivedMessage
4242

4343
// PresumedDead returns a read-only channel for node endpoints that are suspected to be offline
44-
PresumedDead() <-chan PKIidType
44+
PresumedDead() <-chan common.PKIidType
4545

4646
// CloseConn closes a connection to a certain endpoint
4747
CloseConn(peer *RemotePeer)
@@ -50,17 +50,13 @@ type Comm interface {
5050
Stop()
5151

5252
// BlackListPKIid prohibits the module communicating with the given PKIid
53-
BlackListPKIid(PKIid PKIidType)
53+
BlackListPKIid(PKIid common.PKIidType)
5454
}
5555

56-
// PKIidType defines the type that holds the PKI-id
57-
// which is the security identifier of a peer
58-
type PKIidType []byte
59-
6056
// RemotePeer defines a peer's endpoint and its PKIid
6157
type RemotePeer struct {
6258
Endpoint string
63-
PKIID PKIidType
59+
PKIID common.PKIidType
6460
}
6561

6662
// String converts a RemotePeer to a string

‎gossip/comm/comm_impl.go

+17-16
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
"crypto/tls"
2929
"os"
3030

31+
"github.com/hyperledger/fabric/gossip/common"
3132
"github.com/hyperledger/fabric/gossip/proto"
3233
"github.com/hyperledger/fabric/gossip/util"
3334
"github.com/op/go-logging"
@@ -62,7 +63,7 @@ func (c *commImpl) SetDialOpts(opts ...grpc.DialOption) {
6263
}
6364

6465
// NewCommInstanceWithServer creates a comm instance that creates an underlying gRPC server
65-
func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID PKIidType, dialOpts ...grpc.DialOption) (Comm, error) {
66+
func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID common.PKIidType, dialOpts ...grpc.DialOption) (Comm, error) {
6667
var ll net.Listener
6768
var s *grpc.Server
6869
var secOpt grpc.DialOption
@@ -86,11 +87,11 @@ func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID PKIidType, d
8687
gSrv: s,
8788
msgPublisher: NewChannelDemultiplexer(),
8889
lock: &sync.RWMutex{},
89-
deadEndpoints: make(chan PKIidType, 100),
90+
deadEndpoints: make(chan common.PKIidType, 100),
9091
stopping: int32(0),
9192
exitChan: make(chan struct{}, 1),
9293
subscriptions: make([]chan ReceivedMessage, 0),
93-
blackListedPKIIDs: make([]PKIidType, 0),
94+
blackListedPKIIDs: make([]common.PKIidType, 0),
9495
}
9596
commInst.connStore = newConnStore(commInst, pkID, commInst.logger)
9697

@@ -112,7 +113,7 @@ func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID PKIidType, d
112113
}
113114

114115
// NewCommInstance creates a new comm instance that binds itself to the given gRPC server
115-
func NewCommInstance(s *grpc.Server, sec SecurityProvider, PKIID PKIidType, dialOpts ...grpc.DialOption) (Comm, error) {
116+
func NewCommInstance(s *grpc.Server, sec SecurityProvider, PKIID common.PKIidType, dialOpts ...grpc.DialOption) (Comm, error) {
116117
commInst, err := NewCommInstanceWithServer(-1, sec, PKIID)
117118
if err != nil {
118119
return nil, err
@@ -128,7 +129,7 @@ type commImpl struct {
128129
connStore *connectionStore
129130
PKIID []byte
130131
port int
131-
deadEndpoints chan PKIidType
132+
deadEndpoints chan common.PKIidType
132133
msgPublisher *ChannelDeMultiplexer
133134
lock *sync.RWMutex
134135
lsnr net.Listener
@@ -137,10 +138,10 @@ type commImpl struct {
137138
stopping int32
138139
stopWG sync.WaitGroup
139140
subscriptions []chan ReceivedMessage
140-
blackListedPKIIDs []PKIidType
141+
blackListedPKIIDs []common.PKIidType
141142
}
142143

143-
func (c *commImpl) createConnection(endpoint string, expectedPKIID PKIidType) (*connection, error) {
144+
func (c *commImpl) createConnection(endpoint string, expectedPKIID common.PKIidType) (*connection, error) {
144145
c.logger.Debug("Entering", endpoint, expectedPKIID)
145146
defer c.logger.Debug("Exiting")
146147
if c.isStopping() {
@@ -204,7 +205,7 @@ func (c *commImpl) Send(msg *proto.GossipMessage, peers ...*RemotePeer) {
204205
}
205206
}
206207

207-
func (c *commImpl) BlackListPKIid(PKIID PKIidType) {
208+
func (c *commImpl) BlackListPKIid(PKIID common.PKIidType) {
208209
c.logger.Info("Entering", PKIID)
209210
defer c.logger.Info("Exiting")
210211
c.lock.Lock()
@@ -213,7 +214,7 @@ func (c *commImpl) BlackListPKIid(PKIID PKIidType) {
213214
c.blackListedPKIIDs = append(c.blackListedPKIIDs, PKIID)
214215
}
215216

216-
func (c *commImpl) isPKIblackListed(p PKIidType) bool {
217+
func (c *commImpl) isPKIblackListed(p common.PKIidType) bool {
217218
c.lock.RLock()
218219
defer c.lock.RUnlock()
219220
for _, pki := range c.blackListedPKIIDs {
@@ -251,7 +252,7 @@ func (c *commImpl) isStopping() bool {
251252
return atomic.LoadInt32(&c.stopping) == int32(1)
252253
}
253254

254-
func (c *commImpl) Probe(endpoint string, pkiID PKIidType) error {
255+
func (c *commImpl) Probe(endpoint string, pkiID common.PKIidType) error {
255256
if c.isStopping() {
256257
return fmt.Errorf("Stopping!")
257258
}
@@ -274,7 +275,7 @@ func (c *commImpl) Probe(endpoint string, pkiID PKIidType) error {
274275
return err
275276
}
276277

277-
func (c *commImpl) Accept(acceptor util.MessageAcceptor) <-chan ReceivedMessage {
278+
func (c *commImpl) Accept(acceptor common.MessageAcceptor) <-chan ReceivedMessage {
278279
genericChan := c.msgPublisher.AddChannel(acceptor)
279280
specificChan := make(chan ReceivedMessage, 10)
280281

@@ -311,7 +312,7 @@ func (c *commImpl) Accept(acceptor util.MessageAcceptor) <-chan ReceivedMessage
311312
return specificChan
312313
}
313314

314-
func (c *commImpl) PresumedDead() <-chan PKIidType {
315+
func (c *commImpl) PresumedDead() <-chan common.PKIidType {
315316
return c.deadEndpoints
316317
}
317318

@@ -351,7 +352,7 @@ func (c *commImpl) Stop() {
351352
c.stopWG.Wait()
352353
}
353354

354-
func (c *commImpl) GetPKIid() PKIidType {
355+
func (c *commImpl) GetPKIid() common.PKIidType {
355356
return c.PKIID
356357
}
357358

@@ -366,7 +367,7 @@ func extractRemoteAddress(stream stream) string {
366367
return remoteAddress
367368
}
368369

369-
func (c *commImpl) authenticateRemotePeer(stream stream) (PKIidType, error) {
370+
func (c *commImpl) authenticateRemotePeer(stream stream) (common.PKIidType, error) {
370371
ctx := stream.Context()
371372
remoteAddress := extractRemoteAddress(stream)
372373
tlsUnique := ExtractTLSUnique(ctx)
@@ -455,7 +456,7 @@ func (c *commImpl) Ping(context.Context, *proto.Empty) (*proto.Empty, error) {
455456
return &proto.Empty{}, nil
456457
}
457458

458-
func (c *commImpl) disconnect(pkiID PKIidType) {
459+
func (c *commImpl) disconnect(pkiID common.PKIidType) {
459460
if c.isStopping() {
460461
return
461462
}
@@ -485,7 +486,7 @@ func readWithTimeout(stream interface{}, timeout time.Duration) *proto.GossipMes
485486
}
486487
}
487488

488-
func createConnectionMsg(pkiID PKIidType, sig []byte) *proto.GossipMessage {
489+
func createConnectionMsg(pkiID common.PKIidType, sig []byte) *proto.GossipMessage {
489490
return &proto.GossipMessage{
490491
Nonce: 0,
491492
Content: &proto.GossipMessage_Conn{

‎gossip/comm/comm_test.go

+3-2
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import (
3131
"golang.org/x/net/context"
3232
"google.golang.org/grpc"
3333
"google.golang.org/grpc/credentials"
34+
"github.com/hyperledger/fabric/gossip/common"
3435
)
3536

3637
func init() {
@@ -91,7 +92,7 @@ func TestHandshake(t *testing.T) {
9192
clientTLSUnique := ExtractTLSUnique(stream.Context())
9293
sig, err := naiveSec.Sign(clientTLSUnique)
9394
assert.NoError(t, err, "%v", err)
94-
msg := createConnectionMsg(PKIidType("localhost:9610"), sig)
95+
msg := createConnectionMsg(common.PKIidType("localhost:9610"), sig)
9596
stream.Send(msg)
9697
msg, err = stream.Recv()
9798
assert.NoError(t, err, "%v", err)
@@ -150,7 +151,7 @@ func TestHandshake(t *testing.T) {
150151
} else {
151152
sig[0] = 0
152153
}
153-
msg = createConnectionMsg(PKIidType("localhost:9612"), sig)
154+
msg = createConnectionMsg(common.PKIidType("localhost:9612"), sig)
154155
stream.Send(msg)
155156
msg, err = stream.Recv()
156157
assert.Equal(t, []byte("localhost:9611"), msg.GetConn().PkiID)

‎gossip/comm/conn.go

+8-7
Original file line numberDiff line numberDiff line change
@@ -24,17 +24,18 @@ import (
2424
"github.com/hyperledger/fabric/gossip/proto"
2525
"github.com/hyperledger/fabric/gossip/util"
2626
"google.golang.org/grpc"
27+
"github.com/hyperledger/fabric/gossip/common"
2728
)
2829

2930
type handler func(*proto.GossipMessage)
3031

3132
type connFactory interface {
32-
createConnection(endpoint string, pkiID PKIidType) (*connection, error)
33+
createConnection(endpoint string, pkiID common.PKIidType) (*connection, error)
3334
}
3435

3536
type connectionStore struct {
3637
logger *util.Logger // logger
37-
selfPKIid PKIidType // pkiID of this peer
38+
selfPKIid common.PKIidType // pkiID of this peer
3839
isClosing bool // whether this connection store is shutting down
3940
connFactory connFactory // creates a connection to remote peer
4041
sync.RWMutex // synchronize access to shared variables
@@ -43,7 +44,7 @@ type connectionStore struct {
4344
// used to prevent concurrent connection establishment to the same remote endpoint
4445
}
4546

46-
func newConnStore(connFactory connFactory, pkiID PKIidType, logger *util.Logger) *connectionStore {
47+
func newConnStore(connFactory connFactory, pkiID common.PKIidType, logger *util.Logger) *connectionStore {
4748
return &connectionStore{
4849
connFactory: connFactory,
4950
isClosing: false,
@@ -154,7 +155,7 @@ func (cs *connectionStore) shutdown() {
154155
wg.Wait()
155156
}
156157

157-
func (cs *connectionStore) onConnected(serverStream proto.Gossip_GossipStreamServer, pkiID PKIidType) *connection {
158+
func (cs *connectionStore) onConnected(serverStream proto.Gossip_GossipStreamServer, pkiID common.PKIidType) *connection {
158159
cs.Lock()
159160
defer cs.Unlock()
160161

@@ -165,15 +166,15 @@ func (cs *connectionStore) onConnected(serverStream proto.Gossip_GossipStreamSer
165166
return cs.registerConn(pkiID, serverStream)
166167
}
167168

168-
func (cs *connectionStore) registerConn(pkiID PKIidType, serverStream proto.Gossip_GossipStreamServer) *connection {
169+
func (cs *connectionStore) registerConn(pkiID common.PKIidType, serverStream proto.Gossip_GossipStreamServer) *connection {
169170
conn := newConnection(nil, nil, nil, serverStream)
170171
conn.pkiID = pkiID
171172
conn.logger = cs.logger
172173
cs.pki2Conn[string(pkiID)] = conn
173174
return conn
174175
}
175176

176-
func (cs *connectionStore) closeByPKIid(pkiID PKIidType) {
177+
func (cs *connectionStore) closeByPKIid(pkiID common.PKIidType) {
177178
cs.Lock()
178179
defer cs.Unlock()
179180
if conn, exists := cs.pki2Conn[string(pkiID)]; exists {
@@ -199,7 +200,7 @@ func newConnection(cl proto.GossipClient, c *grpc.ClientConn, cs proto.Gossip_Go
199200
type connection struct {
200201
outBuff chan *msgSending
201202
logger *util.Logger // logger
202-
pkiID PKIidType // pkiID of the remote endpoint
203+
pkiID common.PKIidType // pkiID of the remote endpoint
203204
handler handler // function to invoke upon a message reception
204205
conn *grpc.ClientConn // gRPC connection to remote endpoint
205206
cl proto.GossipClient // gRPC stub of remote endpoint

‎gossip/comm/demux.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ import (
2020
"sync"
2121
"sync/atomic"
2222

23-
"github.com/hyperledger/fabric/gossip/util"
23+
"github.com/hyperledger/fabric/gossip/common"
2424
)
2525

2626
// ChannelDeMultiplexer is a struct that can receive channel registrations (AddChannel)
@@ -42,7 +42,7 @@ func NewChannelDemultiplexer() *ChannelDeMultiplexer {
4242
}
4343

4444
type channel struct {
45-
pred util.MessageAcceptor
45+
pred common.MessageAcceptor
4646
ch chan interface{}
4747
}
4848

@@ -66,7 +66,7 @@ func (m *ChannelDeMultiplexer) Close() {
6666
}
6767

6868
// AddChannel registers a channel with a certain predicate
69-
func (m *ChannelDeMultiplexer) AddChannel(predicate util.MessageAcceptor) chan interface{} {
69+
func (m *ChannelDeMultiplexer) AddChannel(predicate common.MessageAcceptor) chan interface{} {
7070
m.lock.Lock()
7171
defer m.lock.Unlock()
7272
ch := &channel{ch: make(chan interface{}, 10), pred: predicate}

‎gossip/common/common.go

+23
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
/*
2+
Copyright IBM Corp. 2016 All Rights Reserved.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package common
18+
19+
// PKIidType defines the type that holds the PKI-id
20+
// which is the security identifier of a peer
21+
type PKIidType []byte
22+
23+
type MessageAcceptor func(interface{}) bool

‎gossip/discovery/discovery.go

+6-6
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,10 @@ limitations under the License.
1616

1717
package discovery
1818

19-
import "github.com/hyperledger/fabric/gossip/proto"
19+
import (
20+
"github.com/hyperledger/fabric/gossip/proto"
21+
"github.com/hyperledger/fabric/gossip/common"
22+
)
2023

2124
// CryptoService is an interface that the discovery expects to be implemented and passed on creation
2225
type CryptoService interface {
@@ -43,20 +46,17 @@ type CommService interface {
4346
Accept() <-chan *proto.GossipMessage
4447

4548
// PresumedDead returns a read-only channel for peers that are presumed to be dead
46-
PresumedDead() <-chan PKIidType
49+
PresumedDead() <-chan common.PKIidType
4750

4851
// CloseConn orders to close the connection with a certain peer
4952
CloseConn(peer *NetworkMember)
5053
}
5154

52-
// PKIidType represents a peer's security identity
53-
type PKIidType []byte
54-
5555
// NetworkMember is a peer's representation
5656
type NetworkMember struct {
5757
Endpoint string
5858
Metadata []byte
59-
PKIid PKIidType
59+
PKIid common.PKIidType
6060
}
6161

6262
// Discovery is the interface that represents a discovery module

0 commit comments

Comments
 (0)