[FAB-4533] Use context cancellations in deliveryservice

This addresses FAB-4503 and FAB-4489
The reason for the CI failures is that apparently,
closing the underlying gRPC connection doesn't always make Recv() return.

The solution is to use context.WithCancel and use the cancel function
when Disconnect or Close are called and not only close the gRPC stream.

Change-Id: Ia255467345dc6178764a2eb4f1e51167b914eebd
Signed-off-by: Yacov Manevich <[email protected]>

Author: yacovm

core/deliverservice/client.go

@@ -139,13 +139,14 @@ func (bc *broadcastClient) connect() error {
 		logger.Error("Failed obtaining connection:", err)
 		return err
 	}
-	abc, err := bc.createClient(conn).Deliver(context.Background())
+	ctx, cf := context.WithCancel(context.Background())
+	abc, err := bc.createClient(conn).Deliver(ctx)
 	if err != nil {
 		logger.Error("Connection to ", endpoint, "established but was unable to create gRPC stream:", err)
 		conn.Close()
 		return err
 	}
-	err = bc.afterConnect(conn, abc)
+	err = bc.afterConnect(conn, abc, cf)
 	if err == nil {
 		return nil
 	}
@@ -155,9 +156,9 @@ func (bc *broadcastClient) connect() error {
 	return err
 }
 
-func (bc *broadcastClient) afterConnect(conn *grpc.ClientConn, abc orderer.AtomicBroadcast_DeliverClient) error {
+func (bc *broadcastClient) afterConnect(conn *grpc.ClientConn, abc orderer.AtomicBroadcast_DeliverClient, cf context.CancelFunc) error {
 	bc.Lock()
-	bc.conn = &connection{ClientConn: conn}
+	bc.conn = &connection{ClientConn: conn, cancel: cf}
 	bc.BlocksDeliverer = abc
 	if bc.shouldStop() {
 		bc.Unlock()
@@ -219,13 +220,15 @@ func (bc *broadcastClient) Disconnect() {
 }
 
 type connection struct {
-	*grpc.ClientConn
 	sync.Once
+	*grpc.ClientConn
+	cancel context.CancelFunc
 }
 
 func (c *connection) Close() error {
 	var err error
 	c.Once.Do(func() {
+		c.cancel()
 		err = c.ClientConn.Close()
 	})
 	return err

core/deliverservice/client_test.go

@@ -471,7 +471,6 @@ func testGreenPath(t *testing.T, bdc blocksDelivererConsumer) {
 }
 
 func TestCloseWhileRecv(t *testing.T) {
-	t.Skip()
 	// Scenario: Recv is being called and after a while,
 	// the connection is closed.
 	// The Recv should return immediately in such a case
@@ -600,7 +599,6 @@ func TestProductionUsage(t *testing.T) {
 }
 
 func TestDisconnect(t *testing.T) {
-	t.Skip()
 	// Scenario: spawn 2 ordering service instances
 	// and a client.
 	// Have the client try to Recv() from one of them,