.

go/neo/client_test.go

@@ -39,6 +39,7 @@ import (
 	"lab.nexedi.com/kirr/go123/xnet"
 	"lab.nexedi.com/kirr/go123/xsync"
 
+	"lab.nexedi.com/kirr/neo/go/neo/neonet"
 	bsqlite "lab.nexedi.com/kirr/neo/go/neo/storage/sqlite"
 )
 
@@ -192,6 +193,10 @@ type NEOGoSrv struct {
 	cancel      func()           // to stop spawned nodes
 	serveWG     *xsync.WorkGroup // services are spawned under serveWG
 
+	// network access-point for all nodes
+	// (this is functional test, lonet is not injected)
+	net xnet.Networker
+
 	Ml    xnet.Listener    // M listens here
 	Sl    xnet.Listener    // S listens here
 	M     *Master          // M service
@@ -221,7 +226,15 @@ func StartNEOGoSrv(opt NEOSrvOptions) (_ *NEOGoSrv, err error) {
 
 	// FIXME tune glog to write logs into workdir
 
-	net := xnet.NetPlain("tcp") // FIXME TLS on SSL
+	net, err := neonet.Join(ctx, neonet.Config{
+		CA:   opt.CA(),
+		Cert: opt.Cert(),
+		Key:  opt.Key(),
+	})
+	if err != nil {
+		return nil, err
+	}
+	n.net = net
 
 	n.Ml, err = net.Listen(ctx, "");  if err != nil { return nil, err }
 	n.Sl, err = net.Listen(ctx, "");  if err != nil { return nil, err }
@@ -288,6 +301,11 @@ func (n *NEOGoSrv) Close() (err error) {
 		__ := n.Sback.Close() // XXX should be in Storage?
 		err = xerr.First(err, __)
 	}
+
+	if n.net != nil {
+		__ := n.net.Close()
+		err = xerr.First(err, __)
+	}
 	return err
 }
 

go/neo/cmd/neo/misc.go

@@ -30,13 +30,10 @@ import (
 
 	"github.com/soheilhy/cmux"
 
-	"lab.nexedi.com/kirr/go123/xerr"
 	"lab.nexedi.com/kirr/go123/xnet"
-	"lab.nexedi.com/kirr/go123/xnet/lonet"
-	"lab.nexedi.com/kirr/go123/xstrings"
 	"lab.nexedi.com/kirr/go123/xsync"
 	"lab.nexedi.com/kirr/neo/go/internal/log"
-	"lab.nexedi.com/kirr/neo/go/neo/internal/xtls"
+	"lab.nexedi.com/kirr/neo/go/neo/neonet"
 
 
 	// for http://.../debug/pprof
@@ -53,54 +50,14 @@ func netFlags(flags *flag.FlagSet) (netSetup func(context.Context) (xnet.Network
 
 	flonode := flags.String("lonode", "", "<net>/<host> for this node on lonet network")
 
-	return func(ctx context.Context) (_ xnet.Networker, err error) {
-		defer xerr.Contextf(&err, "network setup")
+	return func(ctx context.Context) (xnet.Networker, error) {
+		return neonet.Join(ctx, neonet.Config{
+			CA:   *fca,
+			Cert: *fcert,
+			Key:  *fkey,
 
-		ca   := *fca
-		cert := *fcert
-		key  := *fkey
-		ssl := (ca != "" || cert != "" || key != "")
-		if !ssl && !(ca != "" && cert != "" && key != "") {
-			return nil, fmt.Errorf("incomplete ca/cert/key provided")
-		}
-
-		var net xnet.Networker
-		defer func() {
-			if err != nil && net != nil {
-				net.Close() // ignore err
-			}
-		}()
-		lonode := *flonode
-		if lonode == "" {
-			net = xnet.NetPlain("tcp") // TODO not only "tcp" ?
-		} else {
-			defer xerr.Contextf(&err, "lonode %s", lonode)
-			netname, hostname, err := xstrings.HeadTail(lonode, "/")
-			if err != nil {
-				return nil, fmt.Errorf("invalid lonode")
-			}
-			network, err := lonet.Join(ctx, netname)
-			if err != nil {
-				return nil, err
-			}
-			host, err := network.NewHost(ctx, hostname)
-			if err != nil {
-				network.Close() // ignore err
-				return nil, err
-			}
-
-			net = host // XXX also need to close network on host close
-		}
-
-		if ssl {
-			tlsCfg, err := xtls.ConfigForP2P(ca, cert, key)
-			if err != nil {
-				return nil, err
-			}
-			net = xnet.NetTLS(net, tlsCfg)
-		}
-
-		return net, nil
+			LoNode: *flonode,
+		})
 	}
 }
 

go/neo/neonet/net.go

+// Copyright (C) 2020  Nexedi SA and Contributors.
+//                     Kirill Smelkov <kirr@nexedi.com>
+//
+// This program is free software: you can Use, Study, Modify and Redistribute
+// it under the terms of the GNU General Public License version 3, or (at your
+// option) any later version, as published by the Free Software Foundation.
+//
+// You can also Link and Combine this program with other software covered by
+// the terms of any of the Free Software licenses or any of the Open Source
+// Initiative approved licenses and Convey the resulting work. Corresponding
+// source of such a combination shall include the source code for all other
+// software used.
+//
+// This program is distributed WITHOUT ANY WARRANTY; without even the implied
+// warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+//
+// See COPYING file for full licensing terms.
+// See https://www.nexedi.com/licensing for rationale and options.
+
+package neonet
+// joining raw underlying network.
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"lab.nexedi.com/kirr/go123/xerr"
+	"lab.nexedi.com/kirr/go123/xnet"
+	"lab.nexedi.com/kirr/go123/xnet/lonet"
+	"lab.nexedi.com/kirr/go123/xstrings"
+
+	"lab.nexedi.com/kirr/neo/go/neo/internal/xtls"
+)
+
+
+// Config specifies network configuration for Join.
+//
+// Empty config means to use plain TCP networking for inter-node exchange.
+type Config struct {
+	// CA/Cert/Key, if non-empty, tells to use TLS in peer-to-peer mode
+	// with CA certificate and public/private node keys taken from specified files.
+	CA, Cert, Key string
+
+	// LoNode, if non-empty, tells to use lonet as underlying network.
+	// Network and hostname should be specified as "<net>/<host>"
+	LoNode string
+}
+
+
+// Join returns network access-point suitable for interoperating with nodes in
+// a NEO cluster according to config.
+func Join(ctx context.Context, cfg Config) (net xnet.Networker, err error) {
+	optv := []string{}
+	ssl := (cfg.CA != "" || cfg.Cert != "" || cfg.Key != "")
+	if ssl {
+		optv = append(optv, "ssl")
+	}
+	if cfg.LoNode != "" {
+		optv = append(optv, fmt.Sprintf("lonode=%q", cfg.LoNode))
+	}
+	defer xerr.Contextf(&err, "neonet join [%s]", strings.Join(optv, " "))
+
+	if ssl && !(cfg.CA != "" && cfg.Cert != "" && cfg.Key != "") {
+		return nil, fmt.Errorf("incomplete ca/cert/key provided")
+	}
+
+	defer func() {
+		if err != nil && net != nil {
+			net.Close() // ignore err
+		}
+	}()
+
+	if cfg.LoNode == "" {
+		net = xnet.NetPlain("tcp") // TODO not only "tcp" ?
+	} else {
+		netname, hostname, err := xstrings.HeadTail(cfg.LoNode, "/")
+		if err != nil {
+			return nil, fmt.Errorf("invalid lonode")
+		}
+		network, err := lonet.Join(ctx, netname)
+		if err != nil {
+			return nil, err
+		}
+		host, err := network.NewHost(ctx, hostname)
+		if err != nil {
+			network.Close() // ignore err
+			return nil, err
+		}
+		network.AutoClose()
+
+		net = host
+	}
+
+	if ssl {
+		tlsCfg, err := xtls.ConfigForP2P(cfg.CA, cfg.Cert, cfg.Key)
+		if err != nil {
+			return nil, err
+		}
+		net = xnet.NetTLS(net, tlsCfg)
+	}
+
+	return net, nil
+}