Skip to content
This repository was archived by the owner on Apr 25, 2025. It is now read-only.

Commit 9f82f8e

Browse files
bstasyszynbstasyszyn
committed
[FAB-9412] Remove stream from connection
The comm/connection should be split into two: connection and streamconnection since not all connections have a stream. Change-Id: I4bb106f2097842ab8606dd0890bb2c39ac4575a8 Signed-off-by: Bob Stasyszyn <Bob.Stasyszyn@securekey.com>
1 parent 3e7d1d1 commit 9f82f8e

File tree

6 files changed

+190
-87
lines changed

6 files changed

+190
-87
lines changed

‎pkg/fab/comm/connection.go‎

Lines changed: 6 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ import (
2222
"github.com/hyperledger/fabric-sdk-go/pkg/core/config/endpoint"
2323
"google.golang.org/grpc"
2424
"google.golang.org/grpc/credentials"
25-
"google.golang.org/grpc/peer"
2625
)
2726

2827
var logger = logging.NewLogger("fabsdk/fab")
@@ -33,22 +32,18 @@ const (
3332
maxCallSendMsgSize = 100 * 1024 * 1024
3433
)
3534

36-
// StreamProvider creates a GRPC stream
37-
type StreamProvider func(conn *grpc.ClientConn) (grpc.ClientStream, error)
38-
3935
// GRPCConnection manages the GRPC connection and client stream
4036
type GRPCConnection struct {
4137
context fabcontext.Client
4238
chConfig fab.ChannelCfg
4339
conn *grpc.ClientConn
44-
stream grpc.ClientStream
4540
commManager fab.CommManager
4641
tlsCertHash []byte
4742
done int32
4843
}
4944

5045
// NewConnection creates a new connection
51-
func NewConnection(ctx fabcontext.Client, chConfig fab.ChannelCfg, streamProvider StreamProvider, url string, opts ...options.Opt) (*GRPCConnection, error) {
46+
func NewConnection(ctx fabcontext.Client, chConfig fab.ChannelCfg, url string, opts ...options.Opt) (*GRPCConnection, error) {
5247
if url == "" {
5348
return nil, errors.New("server URL not specified")
5449
}
@@ -74,38 +69,11 @@ func NewConnection(ctx fabcontext.Client, chConfig fab.ChannelCfg, streamProvide
7469
return nil, errors.Wrapf(err, "could not connect to %s", url)
7570
}
7671

77-
stream, err := streamProvider(grpcconn)
78-
if err != nil {
79-
commManager.ReleaseConn(grpcconn)
80-
return nil, errors.Wrapf(err, "could not create stream to %s", url)
81-
}
82-
83-
if stream == nil {
84-
return nil, errors.New("unexpected nil stream received from provider")
85-
}
86-
peer, ok := peer.FromContext(stream.Context())
87-
if !ok || peer == nil {
88-
//return error - certificate is not available
89-
return nil, errors.Wrapf(err, "No peer cert in GRPC stream")
90-
91-
}
92-
if peer.AuthInfo != nil {
93-
tlsInfo := peer.AuthInfo.(credentials.TLSInfo)
94-
for _, peercert := range tlsInfo.State.PeerCertificates {
95-
err := verifier.ValidateCertificateDates(peercert)
96-
if err != nil {
97-
//log and return error
98-
logger.Error(err)
99-
return nil, errors.Wrapf(err, "Got error while validating certificate dates for [%v]", peercert.Subject)
100-
}
101-
}
102-
}
10372
return &GRPCConnection{
10473
context: ctx,
10574
chConfig: chConfig,
10675
commManager: commManager,
10776
conn: grpcconn,
108-
stream: stream,
10977
tlsCertHash: comm.TLSCertHash(ctx.EndpointConfig()),
11078
}, nil
11179
}
@@ -115,18 +83,18 @@ func (c *GRPCConnection) ChannelConfig() fab.ChannelCfg {
11583
return c.chConfig
11684
}
11785

86+
// ClientConn returns the underlying GRPC connection
87+
func (c *GRPCConnection) ClientConn() *grpc.ClientConn {
88+
return c.conn
89+
}
90+
11891
// Close closes the connection
11992
func (c *GRPCConnection) Close() {
12093
if !c.setClosed() {
12194
logger.Debugf("Already closed")
12295
return
12396
}
12497

125-
logger.Debug("Closing stream....")
126-
if err := c.stream.CloseSend(); err != nil {
127-
logger.Warnf("error closing GRPC stream: %s", err)
128-
}
129-
13098
logger.Debug("Releasing connection....")
13199
c.commManager.ReleaseConn(c.conn)
132100

@@ -142,11 +110,6 @@ func (c *GRPCConnection) setClosed() bool {
142110
return atomic.CompareAndSwapInt32(&c.done, 0, 1)
143111
}
144112

145-
// Stream returns the GRPC stream
146-
func (c *GRPCConnection) Stream() grpc.Stream {
147-
return c.stream
148-
}
149-
150113
// TLSCertHash returns the hash of the TLS cert
151114
func (c *GRPCConnection) TLSCertHash() []byte {
152115
return c.tlsCertHash

‎pkg/fab/comm/connection_test.go‎

Lines changed: 2 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -7,65 +7,31 @@ SPDX-License-Identifier: Apache-2.0
77
package comm
88

99
import (
10-
"context"
1110
"testing"
1211
"time"
1312

14-
"google.golang.org/grpc/keepalive"
15-
1613
eventmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/mocks"
1714
fabmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
1815
mspmocks "github.com/hyperledger/fabric-sdk-go/pkg/msp/test/mockmsp"
19-
20-
pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
21-
"github.com/pkg/errors"
22-
"google.golang.org/grpc"
2316
)
2417

25-
var testStream = func(grpcconn *grpc.ClientConn) (grpc.ClientStream, error) {
26-
return pb.NewDeliverClient(grpcconn).Deliver(context.Background())
27-
}
28-
29-
var invalidStream = func(grpcconn *grpc.ClientConn) (grpc.ClientStream, error) {
30-
return nil, errors.New("simulated error creating stream")
31-
}
32-
3318
func TestConnection(t *testing.T) {
3419
channelID := "testchannel"
3520

3621
context := newMockContext()
3722
chConfig := fabmocks.NewMockChannelCfg(channelID)
3823

39-
_, err := NewConnection(context, chConfig, testStream, "")
24+
_, err := NewConnection(context, chConfig, "")
4025
if err == nil {
4126
t.Fatalf("expected error creating new connection with empty URL")
4227
}
43-
_, err = NewConnection(context, chConfig, testStream, "invalidhost:0000",
44-
WithFailFast(true),
45-
WithCertificate(nil),
46-
WithInsecure(),
47-
WithHostOverride(""),
48-
WithKeepAliveParams(keepalive.ClientParameters{}),
49-
WithConnectTimeout(3*time.Second),
50-
)
51-
if err == nil {
52-
t.Fatalf("expected error creating new connection with invalid URL")
53-
}
54-
_, err = NewConnection(context, chConfig, invalidStream, peerURL)
55-
if err == nil {
56-
t.Fatalf("expected error creating new connection with invalid stream but got none")
57-
}
58-
59-
conn, err := NewConnection(context, chConfig, testStream, peerURL)
28+
conn, err := NewConnection(context, chConfig, peerURL)
6029
if err != nil {
6130
t.Fatalf("error creating new connection: %s", err)
6231
}
6332
if conn.Closed() {
6433
t.Fatalf("expected connection to be open")
6534
}
66-
if conn.Stream() == nil {
67-
t.Fatalf("got invalid stream")
68-
}
6935
if _, err := context.Serialize(); err != nil {
7036
t.Fatalf("error getting identity")
7137
}

‎pkg/fab/comm/streamconnection.go‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
/*
2+
Copyright SecureKey Technologies Inc. All Rights Reserved.
3+
4+
SPDX-License-Identifier: Apache-2.0
5+
*/
6+
7+
package comm
8+
9+
import (
10+
"sync"
11+
12+
"github.com/pkg/errors"
13+
14+
"github.com/hyperledger/fabric-sdk-go/pkg/client/common/verifier"
15+
"github.com/hyperledger/fabric-sdk-go/pkg/common/options"
16+
fabcontext "github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
17+
"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
18+
"google.golang.org/grpc"
19+
"google.golang.org/grpc/credentials"
20+
"google.golang.org/grpc/peer"
21+
)
22+
23+
// StreamProvider creates a GRPC stream
24+
type StreamProvider func(conn *grpc.ClientConn) (grpc.ClientStream, error)
25+
26+
// StreamConnection manages the GRPC connection and client stream
27+
type StreamConnection struct {
28+
*GRPCConnection
29+
stream grpc.ClientStream
30+
lock sync.Mutex
31+
}
32+
33+
// NewStreamConnection creates a new connection with stream
34+
func NewStreamConnection(ctx fabcontext.Client, chConfig fab.ChannelCfg, streamProvider StreamProvider, url string, opts ...options.Opt) (*StreamConnection, error) {
35+
conn, err := NewConnection(ctx, chConfig, url, opts...)
36+
if err != nil {
37+
return nil, err
38+
}
39+
40+
stream, err := streamProvider(conn.conn)
41+
if err != nil {
42+
conn.commManager.ReleaseConn(conn.conn)
43+
return nil, errors.Wrapf(err, "could not create stream to %s", url)
44+
}
45+
46+
if stream == nil {
47+
return nil, errors.New("unexpected nil stream received from provider")
48+
}
49+
50+
peer, ok := peer.FromContext(stream.Context())
51+
if !ok || peer == nil {
52+
//return error - certificate is not available
53+
return nil, errors.Wrapf(err, "No peer cert in GRPC stream")
54+
55+
}
56+
57+
if peer.AuthInfo != nil {
58+
tlsInfo := peer.AuthInfo.(credentials.TLSInfo)
59+
for _, peercert := range tlsInfo.State.PeerCertificates {
60+
err := verifier.ValidateCertificateDates(peercert)
61+
if err != nil {
62+
logger.Error(err)
63+
return nil, errors.Wrapf(err, "error validating certificate dates for [%v]", peercert.Subject)
64+
}
65+
}
66+
}
67+
68+
return &StreamConnection{
69+
GRPCConnection: conn,
70+
stream: stream,
71+
}, nil
72+
}
73+
74+
// Close closes the connection
75+
func (c *StreamConnection) Close() {
76+
c.lock.Lock()
77+
defer c.lock.Unlock()
78+
79+
if c.Closed() {
80+
return
81+
}
82+
83+
logger.Debug("Closing stream....")
84+
if err := c.stream.CloseSend(); err != nil {
85+
logger.Warnf("error closing GRPC stream: %s", err)
86+
}
87+
88+
c.GRPCConnection.Close()
89+
}
90+
91+
// Stream returns the GRPC stream
92+
func (c *StreamConnection) Stream() grpc.Stream {
93+
return c.stream
94+
}

‎pkg/fab/comm/streamconnection_test.go‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
/*
2+
Copyright SecureKey Technologies Inc. All Rights Reserved.
3+
4+
SPDX-License-Identifier: Apache-2.0
5+
*/
6+
7+
package comm
8+
9+
import (
10+
"context"
11+
"testing"
12+
"time"
13+
14+
"google.golang.org/grpc/keepalive"
15+
16+
fabmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
17+
18+
pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
19+
"github.com/pkg/errors"
20+
"google.golang.org/grpc"
21+
)
22+
23+
var testStream = func(grpcconn *grpc.ClientConn) (grpc.ClientStream, error) {
24+
return pb.NewDeliverClient(grpcconn).Deliver(context.Background())
25+
}
26+
27+
var invalidStream = func(grpcconn *grpc.ClientConn) (grpc.ClientStream, error) {
28+
return nil, errors.New("simulated error creating stream")
29+
}
30+
31+
func TestStreamConnection(t *testing.T) {
32+
channelID := "testchannel"
33+
34+
context := newMockContext()
35+
chConfig := fabmocks.NewMockChannelCfg(channelID)
36+
37+
_, err := NewStreamConnection(context, chConfig, testStream, "")
38+
if err == nil {
39+
t.Fatalf("expected error creating new connection with empty URL")
40+
}
41+
_, err = NewStreamConnection(context, chConfig, testStream, "invalidhost:0000",
42+
WithFailFast(true),
43+
WithCertificate(nil),
44+
WithInsecure(),
45+
WithHostOverride(""),
46+
WithKeepAliveParams(keepalive.ClientParameters{}),
47+
WithConnectTimeout(3*time.Second),
48+
)
49+
if err == nil {
50+
t.Fatalf("expected error creating new connection with invalid URL")
51+
}
52+
_, err = NewStreamConnection(context, chConfig, invalidStream, peerURL)
53+
if err == nil {
54+
t.Fatalf("expected error creating new connection with invalid stream but got none")
55+
}
56+
57+
conn, err := NewStreamConnection(context, chConfig, testStream, peerURL)
58+
if err != nil {
59+
t.Fatalf("error creating new connection: %s", err)
60+
}
61+
if conn.Closed() {
62+
t.Fatalf("expected connection to be open")
63+
}
64+
if conn.Stream() == nil {
65+
t.Fatalf("got invalid stream")
66+
}
67+
if _, err := context.Serialize(); err != nil {
68+
t.Fatalf("error getting identity")
69+
}
70+
71+
time.Sleep(1 * time.Second)
72+
73+
conn.Close()
74+
if !conn.Closed() {
75+
t.Fatalf("expected connection to be closed")
76+
}
77+
78+
// Calling close again should be ignored
79+
conn.Close()
80+
}

‎pkg/fab/events/deliverclient/connection/connection.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ type deliverStream interface {
3939

4040
// DeliverConnection manages the connection to the deliver server
4141
type DeliverConnection struct {
42-
comm.GRPCConnection
42+
*comm.StreamConnection
4343
url string
4444
}
4545

@@ -61,7 +61,7 @@ var (
6161
// New returns a new Deliver Server connection
6262
func New(ctx fabcontext.Client, chConfig fab.ChannelCfg, streamProvider StreamProvider, url string, opts ...options.Opt) (*DeliverConnection, error) {
6363
logger.Debugf("Connecting to %s...", url)
64-
connect, err := comm.NewConnection(
64+
connect, err := comm.NewStreamConnection(
6565
ctx, chConfig,
6666
func(grpcconn *grpc.ClientConn) (grpc.ClientStream, error) {
6767
return streamProvider(pb.NewDeliverClient(grpcconn))
@@ -73,8 +73,8 @@ func New(ctx fabcontext.Client, chConfig fab.ChannelCfg, streamProvider StreamPr
7373
}
7474

7575
return &DeliverConnection{
76-
GRPCConnection: *connect,
77-
url: url,
76+
StreamConnection: connect,
77+
url: url,
7878
}, nil
7979
}
8080

0 commit comments

Comments
 (0)
⚡