Network CLI

api/api.go

@@ -2,7 +2,6 @@ package api
 
 import (
 	"context"
-
 	"github.com/libp2p/go-libp2p-core/peer"
 )
 
@@ -34,9 +33,9 @@ type API interface {
 
 	// network
 
-	// // peers
+	NetPeers(context.Context) ([]peer.AddrInfo, error) // TODO: check serialization
+	NetConnect(context.Context, peer.AddrInfo) error
 	// // ping
-	// // connect
 
 	// Struct
 

api/struct.go

@@ -11,9 +11,20 @@ type Struct struct {
 	Internal struct {
 		ID      func(context.Context) (peer.ID, error)
 		Version func(context.Context) (Version, error)
+
+		NetPeers   func(context.Context) ([]peer.AddrInfo, error)
+		NetConnect func(context.Context, peer.AddrInfo) error
 	}
 }
 
+func (c *Struct) NetPeers(ctx context.Context) ([]peer.AddrInfo, error) {
+	return c.Internal.NetPeers(ctx)
+}
+
+func (c *Struct) NetConnect(ctx context.Context, p peer.AddrInfo) error {
+	return c.Internal.NetConnect(ctx, p)
+}
+
 // ID implements API.ID
 func (c *Struct) ID(ctx context.Context) (peer.ID, error) {
 	return c.Internal.ID(ctx)

cli/cmd.go

@@ -1,10 +1,18 @@
 package cli
 
 import (
+	"github.com/filecoin-project/go-lotus/api"
 	"gopkg.in/urfave/cli.v2"
 )
 
+type ApiConnector func() api.API
+
+func getApi(ctx *cli.Context) api.API {
+	return ctx.App.Metadata["api"].(ApiConnector)()
+}
+
 // Commands is the root group of CLI commands
 var Commands = []*cli.Command{
+	netCmd,
 	versionCmd,
 }

cli/net.go

@@ -0,0 +1,134 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+	ma "github.com/multiformats/go-multiaddr"
+	madns "github.com/multiformats/go-multiaddr-dns"
+	"gopkg.in/urfave/cli.v2"
+)
+
+var netCmd = &cli.Command{
+	Name:  "net",
+	Usage: "Manage P2P Network",
+	Subcommands: []*cli.Command{
+		netPeers,
+		netConnect,
+	},
+}
+
+var netPeers = &cli.Command{
+	Name:  "peers",
+	Usage: "Print peers",
+	Action: func(ctx *cli.Context) error {
+		api := getApi(ctx)
+		fmt.Println(api.NetPeers(context.Background()))
+
+		return nil
+	},
+}
+
+var netConnect = &cli.Command{
+	Name:  "connect",
+	Usage: "Connect to a peer",
+	Action: func(ctx *cli.Context) error {
+		pis, err := parseAddresses(context.Background(), ctx.Args().Slice())
+		if err != nil {
+			return err
+		}
+
+		api := getApi(ctx)
+		for _, pi := range pis {
+			fmt.Printf("connect %s", pi.ID.Pretty())
+			err := api.NetConnect(context.Background(), pi)
+			if err != nil {
+				fmt.Println("failure")
+				return err
+			}
+			fmt.Println("success")
+		}
+
+		return nil
+	},
+}
+
+// parseAddresses is a function that takes in a slice of string peer addresses
+// (multiaddr + peerid) and returns a slice of properly constructed peers
+func parseAddresses(ctx context.Context, addrs []string) ([]peer.AddrInfo, error) {
+	// resolve addresses
+	maddrs, err := resolveAddresses(ctx, addrs)
+	if err != nil {
+		return nil, err
+	}
+
+	return peer.AddrInfosFromP2pAddrs(maddrs...)
+}
+
+const (
+	dnsResolveTimeout = 10 * time.Second
+)
+
+// resolveAddresses resolves addresses parallelly
+func resolveAddresses(ctx context.Context, addrs []string) ([]ma.Multiaddr, error) {
+	ctx, cancel := context.WithTimeout(ctx, dnsResolveTimeout)
+	defer cancel()
+
+	var maddrs []ma.Multiaddr
+	var wg sync.WaitGroup
+	resolveErrC := make(chan error, len(addrs))
+
+	maddrC := make(chan ma.Multiaddr)
+
+	for _, addr := range addrs {
+		maddr, err := ma.NewMultiaddr(addr)
+		if err != nil {
+			return nil, err
+		}
+
+		// check whether address ends in `ipfs/Qm...`
+		if _, last := ma.SplitLast(maddr); last.Protocol().Code == ma.P_IPFS {
+			maddrs = append(maddrs, maddr)
+			continue
+		}
+		wg.Add(1)
+		go func(maddr ma.Multiaddr) {
+			defer wg.Done()
+			raddrs, err := madns.Resolve(ctx, maddr)
+			if err != nil {
+				resolveErrC <- err
+				return
+			}
+			// filter out addresses that still doesn't end in `ipfs/Qm...`
+			found := 0
+			for _, raddr := range raddrs {
+				if _, last := ma.SplitLast(raddr); last != nil && last.Protocol().Code == ma.P_IPFS {
+					maddrC <- raddr
+					found++
+				}
+			}
+			if found == 0 {
+				resolveErrC <- fmt.Errorf("found no ipfs peers at %s", maddr)
+			}
+		}(maddr)
+	}
+	go func() {
+		wg.Wait()
+		close(maddrC)
+	}()
+
+	for maddr := range maddrC {
+		maddrs = append(maddrs, maddr)
+	}
+
+	select {
+	case err := <-resolveErrC:
+		return nil, err
+	default:
+	}
+
+	return maddrs, nil
+}

cmd/lotus/main.go

@@ -1,6 +1,8 @@
 package main
 
 import (
+	"github.com/filecoin-project/go-lotus/api"
+	"github.com/filecoin-project/go-lotus/api/client"
 	"log"
 	"os"
 
@@ -20,6 +22,12 @@ func main() {
 		Name:    "lotus",
 		Usage:   "Filecoin decentralized storage network client",
 		Version: build.Version,
+		Metadata: map[string]interface{}{
+			"api": lcli.ApiConnector(func() api.API {
+				// TODO: get this from repo
+				return client.NewRPC("http://127.0.0.1:1234/rpc/v0")
+			}),
+		},
 
 		Commands: append(local, lcli.Commands...),
 	}

daemon/cmd.go

@@ -13,6 +13,12 @@ import (
 var Cmd = &cli.Command{
 	Name:  "daemon",
 	Usage: "Start a lotus daemon process",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:  "api",
+			Value: ":1234",
+		},
+	},
 	Action: func(cctx *cli.Context) error {
 		ctx := context.Background()
 
@@ -26,6 +32,6 @@ var Cmd = &cli.Command{
 			return err
 		}
 
-		return serveRPC(api)
+		return serveRPC(api, cctx.String("api"))
 	},
 }

daemon/rpc.go

@@ -7,9 +7,9 @@ import (
 	"github.com/filecoin-project/go-lotus/lib/jsonrpc"
 )
 
-func serveRPC(api api.API) error {
+func serveRPC(api api.API, addr string) error {
 	rpcServer := jsonrpc.NewServer()
 	rpcServer.Register("Filecoin", api)
 	http.Handle("/rpc/v0", rpcServer)
-	return http.ListenAndServe(":1234", http.DefaultServeMux)
+	return http.ListenAndServe(addr, http.DefaultServeMux)
 }

go.mod

@@ -38,6 +38,7 @@ require (
 	github.com/libp2p/go-maddr-filter v0.0.4
 	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1
 	github.com/multiformats/go-multiaddr v0.0.4
+	github.com/multiformats/go-multiaddr-dns v0.0.2
 	github.com/multiformats/go-multihash v0.0.5
 	github.com/pkg/errors v0.8.1
 	github.com/polydawn/refmt v0.0.0-20190408063855-01bf1e26dd14

lib/jsonrpc/rpc_client.go

@@ -6,11 +6,18 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"reflect"
 	"sync/atomic"
+
+	logging "github.com/ipfs/go-log"
 )
 
+var log = logging.Logger("rpc")
+
+const clientDebug = true
+
 var (
 	errorType   = reflect.TypeOf(new(error)).Elem()
 	contextType = reflect.TypeOf(new(context.Context)).Elem()
@@ -148,6 +155,20 @@ func NewClient(addr string, namespace string, handler interface{}) ClientCloser
 
 			// process response
 
+			if clientDebug {
+				rsp, err := ioutil.ReadAll(httpResp.Body)
+				if err != nil {
+					return processError(err)
+				}
+				if err := httpResp.Body.Close(); err != nil {
+					return processError(err)
+				}
+
+				log.Warnw("rpc response", "body", string(rsp))
+
+				httpResp.Body = ioutil.NopCloser(bytes.NewReader(rsp))
+			}
+
 			var resp clientResponse
 			if valOut != -1 {
 				resp.Result = result(reflect.New(ftyp.Out(valOut)))

node/api.go

@@ -0,0 +1,85 @@
+package node
+
+import (
+	"context"
+	"errors"
+	"reflect"
+
+	"github.com/filecoin-project/go-lotus/api"
+	"github.com/filecoin-project/go-lotus/build"
+
+	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/multiformats/go-multiaddr"
+	"go.uber.org/fx"
+)
+
+var errTyp = reflect.TypeOf(new(error)).Elem()
+
+// TODO: type checking, this isn't JS
+func provideApi(f interface{}, toProvide interface{}) fx.Option {
+	rf := reflect.ValueOf(f)
+	tp := reflect.ValueOf(toProvide).Elem()
+
+	ins := make([]reflect.Type, rf.Type().NumIn())
+	for i := range ins {
+		ins[i] = rf.Type().In(i)
+	}
+
+	ctyp := reflect.FuncOf(ins, []reflect.Type{errTyp}, rf.Type().IsVariadic())
+
+	return fx.Invoke(reflect.MakeFunc(ctyp, func(args []reflect.Value) (results []reflect.Value) {
+		provided := rf.Call(args)
+		tp.Set(provided[0].Elem().Convert(tp.Type()))
+		return []reflect.Value{reflect.ValueOf(new(error)).Elem()}
+	}).Interface())
+}
+
+func apiOption(resAPI *api.Struct) fx.Option {
+	in := &resAPI.Internal
+
+	return fx.Options(
+		provideApi(versionAPI, &in.Version),
+		provideApi(idAPI, &in.ID),
+		provideApi(netPeersAPI, &in.NetPeers),
+		provideApi(netConnectAPI, &in.NetConnect),
+	)
+}
+
+func idAPI(id peer.ID) interface{} {
+	return func(ctx context.Context) (peer.ID, error) {
+		return id, nil
+	}
+}
+
+func versionAPI() interface{} {
+	return func(context.Context) (api.Version, error) {
+		return api.Version{
+			Version: build.Version,
+		}, nil
+	}
+}
+
+func netPeersAPI(h host.Host) interface{} {
+	return func(ctx context.Context) ([]peer.AddrInfo, error) {
+		conns := h.Network().Conns()
+		out := make([]peer.AddrInfo, len(conns))
+
+		for i, conn := range conns {
+			out[i] = peer.AddrInfo{
+				ID: conn.RemotePeer(),
+				Addrs: []multiaddr.Multiaddr{
+					conn.RemoteMultiaddr(),
+				},
+			}
+		}
+
+		return out, nil
+	}
+}
+
+func netConnectAPI(h host.Host) interface{} {
+	return func(ctx context.Context, p peer.AddrInfo) error {
+		return errors.New("nope")
+	}
+}