Make broker flag configurable

cmd/di.go

@@ -244,12 +244,17 @@ func (di *Dependencies) bootstrapServiceComponents(nodeOptions node.Options) {
 
 	discoveryService := discovery.NewService(di.IdentityRegistry, di.IdentityRegistration, di.MysteriumClient, di.SignerFactory)
 
-	newDialogWaiter := func(providerID identity.Identity) communication.DialogWaiter {
+	newDialogWaiter := func(providerID identity.Identity) (communication.DialogWaiter, error) {
+		address, err := nats_discovery.NewAddressFromHostAndID(di.NetworkDefinition.BrokerAddress, providerID)
+		if err != nil {
+			return nil, err
+		}
+
 		return nats_dialog.NewDialogWaiter(
-			nats_discovery.NewAddressGenerate(di.NetworkDefinition.BrokerAddress, providerID),
+			address,
 			di.SignerFactory(providerID),
 			di.IdentityRegistry,
-		)
+		), nil
 	}
 	newDialogHandler := func(proposal dto_discovery.ServiceProposal, configProvider session.ConfigProvider) communication.DialogHandler {
 		promiseHandler := func(dialog communication.Dialog) session.PromiseProcessor {

cmd/flags_network.go

@@ -45,18 +45,18 @@ var (
 
 	discoveryAddressFlag = cli.StringFlag{
 		Name:  "discovery-address",
-		Usage: "Address (URL form) of discovery service",
+		Usage: "`URL` of discovery service",
 		Value: metadata.DefaultNetwork.DiscoveryAPIAddress,
 	}
 	brokerAddressFlag = cli.StringFlag{
 		Name:  "broker-address",
-		Usage: "Address (IP or domain name) of message broker",
+		Usage: "`URI` of message broker",
 		Value: metadata.DefaultNetwork.BrokerAddress,
 	}
 
 	etherRpcFlag = cli.StringFlag{
 		Name:  "ether.client.rpc",
-		Usage: "Url or IPC socket to connect to ethereum node, anything what ethereum client accepts - works",
+		Usage: "URL or IPC socket to connect to ethereum node, anything what ethereum client accepts - works",
 		Value: metadata.DefaultNetwork.EtherClientRPC,
 	}
 	etherContractPaymentsFlag = cli.StringFlag{

communication/nats/discovery/address.go

@@ -19,6 +19,8 @@ package discovery
 
 import (
 	"fmt"
+	"net/url"
+	"strings"
 
 	"github.com/mysteriumnetwork/node/communication/nats"
 	"github.com/mysteriumnetwork/node/identity"
@@ -34,11 +36,26 @@ func NewAddress(topic string, addresses ...string) *AddressNATS {
 	}
 }
 
-// NewAddressGenerate generates NATS address for current node
-func NewAddressGenerate(brokerIP string, myID identity.Identity) *AddressNATS {
-	address := fmt.Sprintf("nats://%s:%d", brokerIP, BrokerPort)
+// NewAddressFromHostAndID generates NATS address for current node
+func NewAddressFromHostAndID(uri string, myID identity.Identity) (*AddressNATS, error) {
+	// Add scheme first otherwise url.Parse() fails.
+	var rawurl string
+	if strings.HasPrefix(uri, "nats:") {
+		rawurl = uri
+	} else {
+		rawurl = fmt.Sprintf("nats://%s", uri)
+	}
+
+	url, err := url.Parse(rawurl)
+	if err != nil {
+		return nil, err
+	}
+
+	if url.Port() == "" {
+		url.Host = fmt.Sprintf("%s:%d", url.Host, BrokerPort)
+	}
 
-	return NewAddress(myID.Address, address)
+	return NewAddress(myID.Address, url.String()), nil
 }
 
 // NewAddressForContact extracts NATS address from given contact structure

communication/nats/discovery/address_test.go

@@ -39,19 +39,34 @@ func TestNewAddress(t *testing.T) {
 	)
 }
 
-func TestNewAddressGenerate(t *testing.T) {
+func TestNewAddressFromHostAndID(t *testing.T) {
+	var tests = []struct {
+		uri  string
+		want string
+	}{
+		{"127.0.0.1", "nats://127.0.0.1:4222"},
+		{"nats://127.0.0.1", "nats://127.0.0.1:4222"},
+		{"127.0.0.1:4222", "nats://127.0.0.1:4222"},
+		{"nats://127.0.0.1:4222", "nats://127.0.0.1:4222"},
+
+		{"nats://127.0.0.1:4333", "nats://127.0.0.1:4333"},
+		{"nats://example.com:4333", "nats://example.com:4333"},
+	}
+
 	myID := identity.FromAddress("provider1")
-	brokerIP := "127.0.0.1"
-	address := NewAddressGenerate(brokerIP, myID)
+	for _, tc := range tests {
+		address, err := NewAddressFromHostAndID(tc.uri, myID)
+		assert.NoError(t, err)
+		assert.Equal(
+			t,
+			&AddressNATS{
+				servers: []string{tc.want},
+				topic:   "provider1",
+			},
+			address,
+		)
 
-	assert.Equal(
-		t,
-		&AddressNATS{
-			servers: []string{"nats://" + brokerIP + ":4222"},
-			topic:   "provider1",
-		},
-		address,
-	)
+	}
 }
 
 func TestNewAddressForContact(t *testing.T) {

core/service/manager.go

@@ -49,7 +49,7 @@ type Service interface {
 }
 
 // DialogWaiterFactory initiates communication channel which waits for incoming dialogs
-type DialogWaiterFactory func(providerID identity.Identity) communication.DialogWaiter
+type DialogWaiterFactory func(providerID identity.Identity) (communication.DialogWaiter, error)
 
 // DialogHandlerFactory initiates instance which is able to handle incoming dialogs
 type DialogHandlerFactory func(dto_discovery.ServiceProposal, session.ConfigProvider) communication.DialogHandler
@@ -75,7 +75,7 @@ func NewManager(
 type Manager struct {
 	identityHandler identity_selector.Handler
 
-	dialogWaiterFactory  func(identity identity.Identity) communication.DialogWaiter
+	dialogWaiterFactory  func(identity identity.Identity) (communication.DialogWaiter, error)
 	dialogWaiter         communication.DialogWaiter
 	dialogHandlerFactory DialogHandlerFactory
 
@@ -102,7 +102,10 @@ func (manager *Manager) Start(options Options) (err error) {
 		return err
 	}
 
-	manager.dialogWaiter = manager.dialogWaiterFactory(providerID)
+	manager.dialogWaiter, err = manager.dialogWaiterFactory(providerID)
+	if err != nil {
+		return err
+	}
 	providerContact, err := manager.dialogWaiter.Start()
 	if err != nil {
 		return err

metadata/network.go

@@ -33,7 +33,7 @@ type NetworkDefinition struct {
 // TestnetDefinition defines parameters for test network (currently default network)
 var TestnetDefinition = NetworkDefinition{
 	"https://testnet-api.mysterium.network/v1",
-	"testnet-broker.mysterium.network",
+	"nats://testnet-broker.mysterium.network",
 	"https://ropsten.infura.io",
 	"https://testnet-morqa.mysterium.network/api/v1",
 	common.HexToAddress("0xbe5F9CCea12Df756bF4a5Baf4c29A10c3ee7C83B"),

server/interface.go

@@ -33,5 +33,5 @@ type Client interface {
 	UnregisterProposal(proposal dto_discovery.ServiceProposal, signer identity.Signer) (err error)
 	PingProposal(proposal dto_discovery.ServiceProposal, signer identity.Signer) (err error)
 
-	SendSessionStats(sessionId session.ID, sessionStats dto.SessionStats, signer identity.Signer) (err error)
+	SendSessionStats(sessionID session.ID, sessionStats dto.SessionStats, signer identity.Signer) (err error)
 }

server/mysterium_api_fake.go

@@ -96,8 +96,8 @@ func (client *ClientFake) FindProposals(providerID string) (proposals []dto_disc
 }
 
 // SendSessionStats heartbeats that session is still active + session upload and download amounts
-func (client *ClientFake) SendSessionStats(sessionId session.ID, sessionStats dto.SessionStats, signer identity.Signer) (err error) {
-	log.Info(mysteriumAPILogPrefix, "Fake session stats sent: ", sessionId)
+func (client *ClientFake) SendSessionStats(sessionID session.ID, sessionStats dto.SessionStats, signer identity.Signer) (err error) {
+	log.Info(mysteriumAPILogPrefix, "Fake session stats sent: ", sessionID)
 
 	return nil
 }