unistack-org/micro
4
0
Fork 3
Code Issues 18 Pull Requests 1 Actions 1 Packages Projects Releases 265 Wiki Activity

Fix the mqtt broker connect. Add to default brokers in cmd

Browse Source
This commit is contained in:
Asim 2016-04-23 00:38:45 +01:00
parent d74cf95278
commit eecdeaae99
2 changed files with 8 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
broker/mqtt/mqtt.go
View File

@ -27,10 +27,8 @@ import (
"strings" "strings"
"time" "time"
"github.com/micro/go-micro/broker"
"github.com/micro/go-micro/cmd"
"github.com/eclipse/paho.mqtt.golang" "github.com/eclipse/paho.mqtt.golang"
"github.com/micro/go-micro/broker"
) )
type mqttBroker struct { type mqttBroker struct {
@ -41,7 +39,6 @@ type mqttBroker struct {
func init() { func init() {
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
cmd.DefaultBrokers["mqtt"] = NewBroker
} }
func setAddrs(addrs []string) []string { func setAddrs(addrs []string) []string {
@ -160,8 +157,10 @@ func (m *mqttBroker) Address() string {
} }
func (m *mqttBroker) Connect() error { func (m *mqttBroker) Connect() error {
t := m.client.Connect() if t := m.client.Connect(); t.Wait() && t.Error() != nil {
return t.Error() return t.Error()
}
return nil
} }
func (m *mqttBroker) Disconnect() error { func (m *mqttBroker) Disconnect() error {
@ -201,7 +200,7 @@ func (m *mqttBroker) Subscribe(topic string, h broker.Handler, opts ...broker.Su
} }
}) })
if t.Error() != nil { if t.Wait() && t.Error() != nil {
return nil, t.Error() return nil, t.Error()
} }

2
cmd/cmd.go
View File

@ -10,6 +10,7 @@ import (
"github.com/micro/cli" "github.com/micro/cli"
"github.com/micro/go-micro/broker" "github.com/micro/go-micro/broker"
"github.com/micro/go-micro/broker/mqtt"
"github.com/micro/go-micro/client" "github.com/micro/go-micro/client"
"github.com/micro/go-micro/registry" "github.com/micro/go-micro/registry"
"github.com/micro/go-micro/selector" "github.com/micro/go-micro/selector"
@ -118,6 +119,7 @@ var (
DefaultBrokers = map[string]func(...broker.Option) broker.Broker{ DefaultBrokers = map[string]func(...broker.Option) broker.Broker{
"http": broker.NewBroker, "http": broker.NewBroker,
"mqtt": mqtt.NewBroker,
} }
DefaultRegistries = map[string]func(...registry.Option) registry.Registry{ DefaultRegistries = map[string]func(...registry.Option) registry.Registry{