Merge pull request #140 from guggero/monitor-rest

loopd+looprpc: add REST compatible status calls

cmd/loop/main.go

@@ -2,18 +2,18 @@ package main
 
 import (
 	"bytes"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
 	"strconv"
 	"time"
 
-	"github.com/golang/protobuf/jsonpb"
-	"github.com/golang/protobuf/proto"
 	"github.com/lightninglabs/loop"
 	"github.com/lightninglabs/loop/looprpc"
 	"github.com/lightninglabs/loop/swap"
+	"github.com/lightninglabs/protobuf-hex-display/json"
+	"github.com/lightninglabs/protobuf-hex-display/jsonpb"
+	"github.com/lightninglabs/protobuf-hex-display/proto"
 
 	"github.com/btcsuite/btcutil"
 
@@ -30,6 +30,10 @@ var (
 	maxRoutingFeeRate = int64(20000)
 
 	defaultSwapWaitTime = 30 * time.Minute
+
+	// maxMsgRecvSize is the largest message our client will receive. We
+	// set this to 200MiB atm.
+	maxMsgRecvSize = grpc.MaxCallRecvMsgSize(1 * 1024 * 1024 * 200)
 )
 
 func printJSON(resp interface{}) {
@@ -49,6 +53,7 @@ func printJSON(resp interface{}) {
 
 func printRespJSON(resp proto.Message) {
 	jsonMarshaler := &jsonpb.Marshaler{
+		OrigName:     true,
 		EmitDefaults: true,
 		Indent:       "    ",
 	}
@@ -83,6 +88,7 @@ func main() {
 	app.Commands = []cli.Command{
 		loopOutCommand, loopInCommand, termsCommand,
 		monitorCommand, quoteCommand, listAuthCommand,
+		listSwapsCommand, swapInfoCommand,
 	}
 
 	err := app.Run(os.Args)
@@ -232,6 +238,7 @@ func logSwap(swap *looprpc.SwapStatus) {
 func getClientConn(address string) (*grpc.ClientConn, error) {
 	opts := []grpc.DialOption{
 		grpc.WithInsecure(),
+		grpc.WithDefaultCallOptions(maxMsgRecvSize),
 	}
 
 	conn, err := grpc.Dial(address, opts...)

cmd/loop/swaps.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"context"
+	"encoding/hex"
+	"fmt"
+
+	"github.com/lightninglabs/loop/looprpc"
+	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/urfave/cli"
+)
+
+var listSwapsCommand = cli.Command{
+	Name:  "listswaps",
+	Usage: "list all swaps in the local database",
+	Description: "Allows the user to get a list of all swaps that are " +
+		"currently stored in the database",
+	Action: listSwaps,
+}
+
+func listSwaps(ctx *cli.Context) error {
+	client, cleanup, err := getClient(ctx)
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	resp, err := client.ListSwaps(
+		context.Background(), &looprpc.ListSwapsRequest{},
+	)
+	if err != nil {
+		return err
+	}
+
+	printRespJSON(resp)
+	return nil
+}
+
+var swapInfoCommand = cli.Command{
+	Name:      "swapinfo",
+	Usage:     "show the status of a swap",
+	ArgsUsage: "id",
+	Description: "Allows the user to get the status of a single swap " +
+		"currently stored in the database",
+	Flags: []cli.Flag{
+		cli.Uint64Flag{
+			Name:  "id",
+			Usage: "the ID of the swap",
+		},
+	},
+	Action: swapInfo,
+}
+
+func swapInfo(ctx *cli.Context) error {
+	args := ctx.Args()
+
+	var id string
+	switch {
+	case ctx.IsSet("id"):
+		id = ctx.String("id")
+	case ctx.NArg() > 0:
+		id = args[0]
+		args = args.Tail()
+	default:
+		// Show command help if no arguments and flags were provided.
+		return cli.ShowCommandHelp(ctx, "swapinfo")
+	}
+
+	if len(id) != hex.EncodedLen(lntypes.HashSize) {
+		return fmt.Errorf("invalid swap ID")
+	}
+	idBytes, err := hex.DecodeString(id)
+	if err != nil {
+		return fmt.Errorf("cannot hex decode id: %v", err)
+	}
+
+	client, cleanup, err := getClient(ctx)
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	resp, err := client.SwapInfo(
+		context.Background(), &looprpc.SwapInfoRequest{Id: idBytes},
+	)
+	if err != nil {
+		return err
+	}
+
+	printRespJSON(resp)
+	return nil
+}

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/google/go-cmp v0.3.1 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.10.0
 	github.com/jessevdk/go-flags v1.4.0
+	github.com/lightninglabs/protobuf-hex-display v1.3.3-0.20191212020323-b444784ce75d
 	github.com/lightningnetwork/lnd v0.9.0-beta-rc3.0.20200121213302-a2977c4438b5
 	github.com/lightningnetwork/lnd/queue v1.0.2
 	github.com/urfave/cli v1.20.0

go.sum

@@ -136,6 +136,7 @@ github.com/lightninglabs/gozmq v0.0.0-20191113021534-d20a764486bf h1:HZKvJUHlcXI
 github.com/lightninglabs/gozmq v0.0.0-20191113021534-d20a764486bf/go.mod h1:vxmQPeIQxPf6Jf9rM8R+B4rKBqLA2AjttNxkFBL2Plk=
 github.com/lightninglabs/neutrino v0.11.0 h1:lPpYFCtsfJX2W5zI4pWycPmbbBdr7zU+BafYdLoD6k0=
 github.com/lightninglabs/neutrino v0.11.0/go.mod h1:CuhF0iuzg9Sp2HO6ZgXgayviFTn1QHdSTJlMncK80wg=
+github.com/lightninglabs/protobuf-hex-display v1.3.3-0.20191212020323-b444784ce75d h1:QWD/5MPnaZfUVP7P8wLa4M8Td2DI7XXHXt2vhVtUgGI=
 github.com/lightninglabs/protobuf-hex-display v1.3.3-0.20191212020323-b444784ce75d/go.mod h1:KDb67YMzoh4eudnzClmvs2FbiLG9vxISmLApUkCa4uI=
 github.com/lightningnetwork/lightning-onion v1.0.1 h1:qChGgS5+aPxFeR6JiUsGvanei1bn6WJpYbvosw/1604=
 github.com/lightningnetwork/lightning-onion v1.0.1/go.mod h1:rigfi6Af/KqsF7Za0hOgcyq2PNH4AN70AaMRxcJkff4=

loopd/daemon.go

@@ -16,9 +16,16 @@ import (
 	"github.com/lightninglabs/loop"
 	"github.com/lightninglabs/loop/lndclient"
 	"github.com/lightninglabs/loop/looprpc"
+	"github.com/lightningnetwork/lnd/lntypes"
 	"google.golang.org/grpc"
 )
 
+var (
+	// maxMsgRecvSize is the largest message our REST proxy will receive. We
+	// set this to 200MiB atm.
+	maxMsgRecvSize = grpc.MaxCallRecvMsgSize(1 * 1024 * 1024 * 200)
+)
+
 // listenerCfg holds closures used to retrieve listeners for the gRPC services.
 type listenerCfg struct {
 	// grpcListener returns a listener to use for the gRPC server.
@@ -68,14 +75,18 @@ func daemon(config *config, lisCfg *listenerCfg) error {
 		return err
 	}
 
+	swaps := make(map[lntypes.Hash]loop.SwapInfo)
 	for _, s := range swapsList {
 		swaps[s.SwapHash] = *s
 	}
 
 	// Instantiate the loopd gRPC server.
 	server := swapClientServer{
-		impl: swapClient,
-		lnd:  &lnd.LndServices,
+		impl:        swapClient,
+		lnd:         &lnd.LndServices,
+		swaps:       swaps,
+		subscribers: make(map[int]chan<- interface{}),
+		statusChan:  make(chan loop.SwapInfo),
 	}
 
 	serverOpts := []grpc.ServerOption{}
@@ -92,12 +103,26 @@ func daemon(config *config, lisCfg *listenerCfg) error {
 	}
 	defer grpcListener.Close()
 
+	// The default JSON marshaler of the REST proxy only sets OrigName to
+	// true, which instructs it to use the same field names as specified in
+	// the proto file and not switch to camel case. What we also want is
+	// that the marshaler prints all values, even if they are falsey.
+	customMarshalerOption := proxy.WithMarshalerOption(
+		proxy.MIMEWildcard, &proxy.JSONPb{
+			OrigName:     true,
+			EmitDefaults: true,
+		},
+	)
+
 	// We'll also create and start an accompanying proxy to serve clients
 	// through REST.
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	mux := proxy.NewServeMux()
-	proxyOpts := []grpc.DialOption{grpc.WithInsecure()}
+	mux := proxy.NewServeMux(customMarshalerOption)
+	proxyOpts := []grpc.DialOption{
+		grpc.WithInsecure(),
+		grpc.WithDefaultCallOptions(maxMsgRecvSize),
+	}
 	err = looprpc.RegisterSwapClientHandlerFromEndpoint(
 		ctx, mux, config.RPCListen, proxyOpts,
 	)
@@ -130,8 +155,6 @@ func daemon(config *config, lisCfg *listenerCfg) error {
 		log.Infof("REST proxy disabled")
 	}
 
-	statusChan := make(chan loop.SwapInfo)
-
 	mainCtx, cancel := context.WithCancel(context.Background())
 	var wg sync.WaitGroup
 
@@ -141,7 +164,7 @@ func daemon(config *config, lisCfg *listenerCfg) error {
 		defer wg.Done()
 
 		log.Infof("Starting swap client")
-		err := swapClient.Run(mainCtx, statusChan)
+		err := swapClient.Run(mainCtx, server.statusChan)
 		if err != nil {
 			log.Error(err)
 		}
@@ -159,25 +182,7 @@ func daemon(config *config, lisCfg *listenerCfg) error {
 		defer wg.Done()
 
 		log.Infof("Waiting for updates")
-		for {
-			select {
-			case swap := <-statusChan:
-				swapsLock.Lock()
-				swaps[swap.SwapHash] = swap
-
-				for _, subscriber := range subscribers {
-					select {
-					case subscriber <- swap:
-					case <-mainCtx.Done():
-						return
-					}
-				}
-
-				swapsLock.Unlock()
-			case <-mainCtx.Done():
-				return
-			}
-		}
+		server.processStatusUpdates(mainCtx)
 	}()
 
 	// Start the grpc server.

loopd/start.go

@@ -7,26 +7,16 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 
 	"github.com/jessevdk/go-flags"
 	"github.com/lightninglabs/loop"
 	"github.com/lightninglabs/loop/lndclient"
 	"github.com/lightningnetwork/lnd/build"
-	"github.com/lightningnetwork/lnd/lntypes"
 )
 
 const (
-	defaultConfTarget = int32(6)
-)
-
-var (
+	defaultConfTarget     = int32(6)
 	defaultConfigFilename = "loopd.conf"
-
-	swaps            = make(map[lntypes.Hash]loop.SwapInfo)
-	subscribers      = make(map[int]chan<- interface{})
-	nextSubscriberID int
-	swapsLock        sync.Mutex
 )
 
 // RPCConfig holds optional options that can be used to make the loop daemon