fix: #35 peer Registration Race when client connects to the signal server

signal/client.go

@@ -109,8 +109,6 @@ func (c *Client) connect(key string, msgHandler func(msg *proto.Message) error)
 	// add key fingerprint to the request header to be identified on the server side
 	md := metadata.New(map[string]string{proto.HeaderId: key})
 	ctx := metadata.NewOutgoingContext(c.ctx, md)
-	//ctx, cancel := context.WithCancel(ctx)
-	//defer cancel()
 
 	stream, err := c.realClient.ConnectStream(ctx)
 
@@ -118,6 +116,15 @@ func (c *Client) connect(key string, msgHandler func(msg *proto.Message) error)
 	if err != nil {
 		return err
 	}
+	// blocks
+	header, err := c.stream.Header()
+	if err != nil {
+		return err
+	}
+	registered := header.Get(proto.HeaderRegistered)
+	if len(registered) == 0 {
+		return fmt.Errorf("didn't receive a registration header from the Signal server whille connecting to the streams")
+	}
 	//connection established we are good to use the stream
 	c.connWg.Done()
 

signal/proto/constants.go

@@ -2,3 +2,4 @@ package proto
 
 // protocol constants, field names that can be used by both client and server
 const HeaderId = "x-wiretrustee-peer-id"
+const HeaderRegistered = "x-wiretrustee-peer-registered"

signal/signal.go

@@ -52,6 +52,13 @@ func (s *Server) ConnectStream(stream proto.SignalExchange_ConnectStreamServer)
 		return err
 	}
 
+	//needed to confirm that the peer has been registered so that the client can proceed
+	header := metadata.Pairs(proto.HeaderRegistered, "1")
+	err = stream.SendHeader(header)
+	if err != nil {
+		return err
+	}
+
 	log.Infof("peer [%s] has successfully connected", p.Id)
 	for {
 		msg, err := stream.Recv()