Create stream channel before performing master election (#27)

* Create stream channel before performing master election

* fix

VERSION

@@ -1 +1 @@
-1.1.8-dev
+1.1.8

pkg/p4rtclient/client.go

@@ -42,8 +42,8 @@ func (c *client) ClientConn() *grpc.ClientConn {
 	return c.grpcClient
 }
 
-func (c *client) PerformMasterArbitration(role *p4api.Role) (*p4api.StreamMessageResponse_Arbitration, error) {
-	resp, err := c.streamClient.PerformMasterArbitration(role)
+func (c *client) PerformMasterArbitration(ctx context.Context, role *p4api.Role) (*p4api.StreamMessageResponse_Arbitration, error) {
+	resp, err := c.streamClient.PerformMasterArbitration(ctx, role)
 	return resp, errors.FromGRPC(err)
 }
 

pkg/p4rtclient/conn_manager.go

@@ -162,12 +162,6 @@ func (m *connManager) Connect(ctx context.Context, destination *Destination) (Cl
 	}
 
 	m.targets[targetID] = p4rtClient
-	streamChannel, err := p4rtClient.p4runtimeClient.StreamChannel(context.Background())
-	if err != nil {
-		log.Errorw("Cannot open a p4rt stream for connection", "targetID", destination.TargetID, "error", err)
-		return nil, err
-	}
-	p4rtClient.streamClient.streamChannel = streamChannel
 	go func() {
 		var conn Conn
 		state := clientConn.GetState()

pkg/p4rtclient/conn_manager_test.go

@@ -381,12 +381,12 @@ func TestClient_SetMasterArbitration(t *testing.T) {
 	assert.NoError(t, err)
 
 	role := p4utils.NewStratumRole("test1", 0, []byte{}, false, true)
-	resp, err := conn1.PerformMasterArbitration(role)
+	resp, err := conn1.PerformMasterArbitration(ctx, role)
 	assert.NoError(t, err)
 	assert.Equal(t, "test1", resp.Arbitration.Role.Name)
 
 	role2 := p4utils.NewStratumRole("test2", 0, []byte{}, false, true)
-	resp, err = conn2.PerformMasterArbitration(role2)
+	resp, err = conn2.PerformMasterArbitration(ctx, role2)
 	assert.Equal(t, "test2", resp.Arbitration.Role.Name)
 
 	assert.NoError(t, err)

pkg/p4rtclient/stream.go

@@ -5,24 +5,28 @@
 package p4rtclient
 
 import (
-	"github.com/onosproject/onos-lib-go/pkg/errors"
+	"context"
 	"github.com/onosproject/onos-net-lib/pkg/p4utils"
 	p4api "github.com/p4lang/p4runtime/go/p4/v1"
 )
 
 // StreamClient p4runtime master stream client
 type StreamClient interface {
-	PerformMasterArbitration(role *p4api.Role) (*p4api.StreamMessageResponse_Arbitration, error)
+	PerformMasterArbitration(ctx context.Context, role *p4api.Role) (*p4api.StreamMessageResponse_Arbitration, error)
 }
 
 type streamClient struct {
 	p4runtimeClient p4api.P4RuntimeClient
-	streamChannel   p4api.P4Runtime_StreamChannelClient
 	deviceID        uint64
 }
 
-func (s *streamClient) PerformMasterArbitration(role *p4api.Role) (*p4api.StreamMessageResponse_Arbitration, error) {
+func (s *streamClient) PerformMasterArbitration(ctx context.Context, role *p4api.Role) (*p4api.StreamMessageResponse_Arbitration, error) {
 	electionID := p4utils.TimeBasedElectionID()
+	channel, err := s.p4runtimeClient.StreamChannel(ctx)
+	if err != nil {
+		return nil, err
+	}
+
 	request := &p4api.StreamMessageRequest{
 		Update: &p4api.StreamMessageRequest_Arbitration{Arbitration: &p4api.MasterArbitrationUpdate{
 			DeviceId:   s.deviceID,
@@ -31,26 +35,26 @@ func (s *streamClient) PerformMasterArbitration(role *p4api.Role) (*p4api.Stream
 		}},
 	}
 	log.Infow("Sending master arbitration request", "request", request)
-	err := s.streamChannel.Send(request)
+	err = channel.Send(request)
 	if err != nil {
 		return nil, err
 	}
 
-	in, err := s.streamChannel.Recv()
-	if err != nil {
-		return nil, err
-	}
-
-	switch v := in.Update.(type) {
-	case *p4api.StreamMessageResponse_Arbitration:
-		log.Infow("Received arbitration response", "response", v)
+	for {
+		in, err := channel.Recv()
 		if err != nil {
 			return nil, err
 		}
-		return v, nil
 
+		switch v := in.Update.(type) {
+		case *p4api.StreamMessageResponse_Arbitration:
+			log.Infow("Received arbitration response", "response", v)
+			if err != nil {
+				return nil, err
+			}
+			return v, nil
+		}
 	}
-	return nil, errors.NewNotSupported("not an arbitration response message")
 
 }