.

go/neo/xneo/xneo.go

@@ -33,6 +33,7 @@ import (
 
 	"lab.nexedi.com/kirr/neo/go/internal/log"
 	"lab.nexedi.com/kirr/neo/go/internal/task"
+	"lab.nexedi.com/kirr/neo/go/internal/xcontext"
 	"lab.nexedi.com/kirr/neo/go/internal/xio"
 
 	"lab.nexedi.com/kirr/neo/go/neo/neonet"
@@ -89,7 +90,8 @@ func NewNodeApp(net xnet.Networker, typ proto.NodeType, clusterName, masterAddr
 // Dial does not update .NodeTab or its node entries in any way.
 // For establishing links to peers present in .NodeTab use Node.Dial.
 //
-// XXX unexport after NodeApp += talkMaster
+// XXX unexport after NodeApp += talkMaster  <- used only to dial to M
+// <- dialing to other nodes always go through node.Dial
 func (app *NodeApp) Dial(ctx context.Context, peerType proto.NodeType, addr string) (_ *neonet.NodeLink, _ *proto.AcceptIdentification, err error) {
 	defer task.Runningf(&ctx, "dial %v (%v)", addr, peerType)(&err)
 
@@ -100,7 +102,7 @@ func (app *NodeApp) Dial(ctx context.Context, peerType proto.NodeType, addr stri
 
 	log.Info(ctx, "dialed ok; requesting identification...")
 	defer xerr.Contextf(&err, "%s: request identification", link)
-	// close link on error or FIXME: ctx cancel
+	// close link on error or FIXME: ctx cancel	XXX -> xcontext.WithCloseOnErrCancel
 	//cleanup := xio.CloseWhenDone(ctx, link)
 	defer func() {
 		if err != nil {
@@ -183,7 +185,7 @@ type listener struct {
 	l neonet.LinkListener
 }
 
-func (l *listener) Accept(ctx context.Context) (_ *neonet.Request, _ *proto.RequestIdentification, err error) {
+func (l *listener) Accept(ctx context.Context) (_ *neonet.Request, msgID *proto.RequestIdentification, err error) {
 	link, err := l.l.Accept(ctx)
 	if err != nil {
 		return nil, nil, err
@@ -191,20 +193,29 @@ func (l *listener) Accept(ctx context.Context) (_ *neonet.Request, _ *proto.Requ
 
 	// identify peer
 	// the first conn must come with RequestIdentification packet
-	defer xerr.Context(&err, "identify")	// XXX -> task.ErrContext?
-	req, err := link.Recv1(/*XXX ctx*/)
+	defer xerr.Context(&err, "identify")
+	var req neonet.Request
+	err = xcontext.WithCloseOnErrCancel(ctx, link, func() error {
+		var err error
+		req, err = link.Recv1(/*XXX ctx*/)
+		if err != nil {
+			return err
+		}
+
+		switch msg := req.Msg.(type) {
+		case *proto.RequestIdentification:
+			msgID = msg
+			return nil
+		}
+
+		emsg := &proto.Error{proto.PROTOCOL_ERROR, fmt.Sprintf("unexpected message %T", req.Msg)}
+		req.Reply(emsg)	// ignore err
+		return emsg
+	})
 	if err != nil {
 		return nil, nil, err
 	}
-
-	switch msg := req.Msg.(type) {
-	case *proto.RequestIdentification:
-		return &req, msg, nil
-	}
-
-	emsg := &proto.Error{proto.PROTOCOL_ERROR, fmt.Sprintf("unexpected message %T", req.Msg)}
-	req.Reply(emsg)	// XXX err
-	return nil, nil, emsg
+	return &req, msgID, err
 }
 
 func (l *listener) Close() error   { return l.l.Close() }