add dmsghttp access

cmd/config-bootstrapper/commands/root.go

@@ -4,12 +4,17 @@ package commands
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"log"
 	"os"
 
 	cc "github.com/ivanpirog/coloredcobra"
+	"github.com/skycoin/dmsg/pkg/direct"
+	"github.com/skycoin/dmsg/pkg/dmsg"
+	"github.com/skycoin/dmsg/pkg/dmsghttp"
 	"github.com/skycoin/skywire-utilities/pkg/buildinfo"
+	"github.com/skycoin/skywire-utilities/pkg/cipher"
 	"github.com/skycoin/skywire-utilities/pkg/cmdutil"
 	"github.com/skycoin/skywire-utilities/pkg/logging"
 	"github.com/skycoin/skywire-utilities/pkg/tcpproxy"
@@ -24,13 +29,19 @@ var (
 	tag      string
 	stunPath string
 	domain   string
+	dmsgDisc string
+	sk       cipher.SecKey
+	dmsgPort uint16
 )
 
 func init() {
 	RootCmd.Flags().StringVarP(&addr, "addr", "a", ":9082", "address to bind to\033[0m")
 	RootCmd.Flags().StringVar(&tag, "tag", "address_resolver", "logging tag\033[0m")
 	RootCmd.Flags().StringVarP(&stunPath, "config", "c", "./config.json", "stun server list file location\033[0m")
 	RootCmd.Flags().StringVarP(&domain, "domain", "d", "skywire.skycoin.com", "the domain of the endpoints\033[0m")
+	RootCmd.Flags().StringVar(&dmsgDisc, "dmsg-disc", "http://dmsgd.skywire.skycoin.com", "url of dmsg-discovery\033[0m")
+	RootCmd.Flags().Var(&sk, "sk", "dmsg secret key\r")
+	RootCmd.Flags().Uint16Var(&dmsgPort, "dmsgPort", dmsg.DefaultDmsgHTTPPort, "dmsg port value\r")
 	var helpflag bool
 	RootCmd.SetUsageTemplate(help)
 	RootCmd.PersistentFlags().BoolVarP(&helpflag, "help", "h", false, "help for config-bootstrapper")
@@ -59,14 +70,52 @@ var RootCmd = &cobra.Command{
 		logger := logging.MustGetLogger(tag)
 		config := readConfig(logger, stunPath)
 
-		conAPI := api.New(logger, config, domain)
+		pk, err := sk.PubKey()
+		if err != nil {
+			logger.WithError(err).Warn("No SecKey found. Skipping serving on dmsghttp.")
+		}
+
+		var dmsgAddr string
+		if !pk.Null() {
+			dmsgAddr = fmt.Sprintf("%s:%d", pk.Hex(), dmsgPort)
+		}
+
+		conAPI := api.New(logger, config, domain, dmsgAddr)
 		if logger != nil {
 			logger.Infof("Listening on %s", addr)
 		}
 
 		ctx, cancel := cmdutil.SignalContext(context.Background(), logger)
 		defer cancel()
 
+		if !pk.Null() {
+			servers := dmsghttp.GetServers(ctx, dmsgDisc, logger)
+
+			var keys cipher.PubKeys
+			keys = append(keys, pk)
+			dClient := direct.NewClient(direct.GetAllEntries(keys, servers), logger)
+			config := &dmsg.Config{
+				MinSessions:    0, // listen on all available servers
+				UpdateInterval: dmsg.DefaultUpdateInterval,
+			}
+
+			dmsgDC, closeDmsgDC, err := direct.StartDmsg(ctx, logger, pk, sk, dClient, config)
+			if err != nil {
+				logger.WithError(err).Fatal("failed to start direct dmsg client.")
+			}
+
+			defer closeDmsgDC()
+
+			go dmsghttp.UpdateServers(ctx, dClient, dmsgDisc, dmsgDC, logger)
+
+			go func() {
+				if err := dmsghttp.ListenAndServe(ctx, sk, conAPI, dClient, dmsg.DefaultDmsgHTTPPort, dmsgDC, logger); err != nil {
+					logger.Errorf("dmsghttp.ListenAndServe: %v", err)
+					cancel()
+				}
+			}()
+		}
+
 		go func() {
 			if err := tcpproxy.ListenAndServe(addr, conAPI); err != nil {
 				logger.Errorf("conAPI.ListenAndServe: %v", err)

pkg/config-bootstrapper/api/api.go

@@ -36,12 +36,15 @@ type API struct {
 
 	closeOnce sync.Once
 	closeC    chan struct{}
+
+	dmsgAddr string
 }
 
 // HealthCheckResponse is struct of /health endpoint
 type HealthCheckResponse struct {
 	BuildInfo *buildinfo.Info `json:"build_info,omitempty"`
 	StartedAt time.Time       `json:"started_at"`
+	DmsgAddr  string          `json:"dmsg_address,omitempty"`
 }
 
 // Error is the object returned to the client when there's an error.
@@ -58,7 +61,7 @@ type Config struct {
 }
 
 // New creates a new api.
-func New(log *logging.Logger, conf Config, domain string) *API {
+func New(log *logging.Logger, conf Config, domain, dmsgAddr string) *API {
 
 	sd := strings.Replace(skyenv.ServiceDiscAddr, "skycoin.com", domain, -1)
 	if domain == "skywire.skycoin.com" {
@@ -85,6 +88,7 @@ func New(log *logging.Logger, conf Config, domain string) *API {
 		services:       services,
 		dmsghttpConfTs: time.Now().Add(-5 * time.Minute),
 		closeC:         make(chan struct{}),
+		dmsgAddr:       dmsgAddr,
 	}
 
 	r := chi.NewRouter()
@@ -119,6 +123,7 @@ func (a *API) health(w http.ResponseWriter, r *http.Request) {
 	a.writeJSON(w, r, http.StatusOK, HealthCheckResponse{
 		BuildInfo: info,
 		StartedAt: a.startedAt,
+		DmsgAddr:  a.dmsgAddr,
 	})
 }