diff --git a/cmd/di.go b/cmd/di.go index e07d6fd48a..125da55653 100644 --- a/cmd/di.go +++ b/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 { diff --git a/cmd/flags_network.go b/cmd/flags_network.go index c0e06a2035..8ee8db4944 100644 --- a/cmd/flags_network.go +++ b/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{ diff --git a/communication/nats/discovery/address.go b/communication/nats/discovery/address.go index 3bc19adada..ec9e891e1d 100644 --- a/communication/nats/discovery/address.go +++ b/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 diff --git a/communication/nats/discovery/address_test.go b/communication/nats/discovery/address_test.go index b370e2f06c..c64034b872 100644 --- a/communication/nats/discovery/address_test.go +++ b/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) { diff --git a/core/service/manager.go b/core/service/manager.go index e040c716a6..f97791dcef 100644 --- a/core/service/manager.go +++ b/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 diff --git a/metadata/network.go b/metadata/network.go index ce0c0cb815..e32f9c31a6 100644 --- a/metadata/network.go +++ b/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"), diff --git a/server/interface.go b/server/interface.go index 31531dd181..253084d0ed 100644 --- a/server/interface.go +++ b/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) } diff --git a/server/mysterium_api_fake.go b/server/mysterium_api_fake.go index c11a827ab0..c7a06d84bf 100644 --- a/server/mysterium_api_fake.go +++ b/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 }