@@ -27,11 +27,14 @@ import (
27
27
config "github.com/hyperledger/fabric-sdk-go/config"
28
28
consumer "github.com/hyperledger/fabric/events/consumer"
29
29
ehpb "github.com/hyperledger/fabric/protos/peer"
30
+ logging "github.com/op/go-logging"
30
31
"golang.org/x/net/context"
31
32
"google.golang.org/grpc"
32
33
"google.golang.org/grpc/credentials"
33
34
)
34
35
36
+ var logger = logging .MustGetLogger ("fabric_sdk_go" )
37
+
35
38
const defaultTimeout = time .Second * 3
36
39
37
40
//EventsClient holds the stream and adapter for consumer to work with
@@ -51,6 +54,7 @@ type eventsClient struct {
51
54
adapter consumer.EventAdapter
52
55
TLSCertificate string
53
56
TLSServerHostOverride string
57
+ clientConn * grpc.ClientConn
54
58
}
55
59
56
60
//NewEventsClient Returns a new grpc.ClientConn to the configured local PEER.
@@ -63,7 +67,7 @@ func NewEventsClient(peerAddress string, certificate string, serverhostoverride
63
67
regTimeout = 60 * time .Second
64
68
err = fmt .Errorf ("regTimeout > 60, setting to 60 sec" )
65
69
}
66
- return & eventsClient {sync.RWMutex {}, peerAddress , regTimeout , nil , adapter , certificate , serverhostoverride }, err
70
+ return & eventsClient {sync.RWMutex {}, peerAddress , regTimeout , nil , adapter , certificate , serverhostoverride , nil }, err
67
71
}
68
72
69
73
//newEventsClientConnectionWithAddress Returns a new grpc.ClientConn to the configured local PEER.
@@ -140,7 +144,7 @@ func (ec *eventsClient) UnregisterAsync(ies []*ehpb.Interest) error {
140
144
emsg := & ehpb.Event {Event : & ehpb.Event_Unregister {Unregister : & ehpb.Unregister {Events : ies }}}
141
145
var err error
142
146
if err = ec .send (emsg ); err != nil {
143
- err = fmt .Errorf ("error on unregister send %s\n " , err )
147
+ err = fmt .Errorf ("error on unregister send %s" , err )
144
148
}
145
149
146
150
return err
@@ -181,7 +185,7 @@ func (ec *eventsClient) unregister(ies []*ehpb.Interest) error {
181
185
func (ec * eventsClient ) Recv () (* ehpb.Event , error ) {
182
186
in , err := ec .stream .Recv ()
183
187
if err == io .EOF {
184
- // read done.
188
+ // read done
185
189
if ec .adapter != nil {
186
190
ec .adapter .Disconnected (nil )
187
191
}
@@ -227,6 +231,7 @@ func (ec *eventsClient) Start() error {
227
231
if err != nil {
228
232
return fmt .Errorf ("Could not create client conn to %s (%v)" , ec .peerAddress , err )
229
233
}
234
+ ec .clientConn = conn
230
235
231
236
ies , err := ec .adapter .GetInterestedEvents ()
232
237
if err != nil {
@@ -258,5 +263,18 @@ func (ec *eventsClient) Stop() error {
258
263
// in case the stream/chat server has not been established earlier, we assume that it's closed, successfully
259
264
return nil
260
265
}
261
- return ec .stream .CloseSend ()
266
+ //this closes only sending direction of the stream; event is still there
267
+ //read will not return an error
268
+ err := ec .stream .CloseSend ()
269
+ if err != nil {
270
+ return err
271
+ }
272
+ //close client connection
273
+ if ec .clientConn != nil {
274
+ err := ec .clientConn .Close ()
275
+ if err != nil {
276
+ return err
277
+ }
278
+ }
279
+ return nil
262
280
}
0 commit comments