Merge pull request #16 from seqsense/add-wss-support

Add mqtt over wss protocol

awsiotprotocol/config.go

@@ -19,5 +19,5 @@ type Config struct {
 	CertPath string
 	CaPath   string
 	ClientId string
-	Host     string
+	Url      string
 }

awsiotprotocol/loader.go

@@ -16,6 +16,7 @@ package awsiotprotocol
 
 import (
 	"fmt"
+	"net/url"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 )
@@ -31,12 +32,18 @@ type Protocol interface {
 
 func init() {
 	registerProtocol(Mqtts{})
+	registerProtocol(Wss{})
 }
 
-func ByName(name string) (Protocol, error) {
-	p, ok := protocols[name]
+func ByUrl(u string) (Protocol, error) {
+	url, err := url.Parse(u)
+	if err != nil {
+		return nil, err
+	}
+
+	p, ok := protocols[url.Scheme]
 	if !ok {
-		return nil, fmt.Errorf("Protocol \"%s\" is not supported", name)
+		return nil, fmt.Errorf("Protocol \"%s\" is not supported", url.Scheme)
 	}
 	return p, nil
 }

awsiotprotocol/mqtts.go

@@ -17,8 +17,11 @@ package awsiotprotocol
 import (
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"io/ioutil"
+	"net/url"
+	"strconv"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 )
@@ -31,15 +34,32 @@ func (s Mqtts) Name() string {
 }
 
 func (s Mqtts) NewClientOptions(opt *Config) (*mqtt.ClientOptions, error) {
+	url, err := url.Parse(opt.Url)
+	if err != nil {
+		return nil, err
+	}
+	host := url.Hostname()
+	if host == "" {
+		return nil, errors.New("Hostname is not provided in the URL")
+	}
+	var port int
+	if url.Port() != "" {
+		port, err = strconv.Atoi(url.Port())
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		port = 8883
+	}
 
-	tlsconfig, err := newTLSConfig(opt.Host, opt.CaPath, opt.CertPath, opt.KeyPath)
+	tlsconfig, err := newTLSConfig(host, opt.CaPath, opt.CertPath, opt.KeyPath)
 	if err != nil {
 		return nil, err
 	}
 
 	opts := mqtt.NewClientOptions()
 	opts.AddBroker(
-		fmt.Sprintf("ssl://%s:8883", opt.Host))
+		fmt.Sprintf("ssl://%s:%d", host, port))
 	opts.SetClientID(opt.ClientId)
 	opts.SetTLSConfig(tlsconfig)
 

awsiotprotocol/wss.go

@@ -0,0 +1,35 @@
+// Copyright 2018 SEQSENSE, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package awsiotprotocol
+
+import (
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+type Wss struct {
+}
+
+func (s Wss) Name() string {
+	return "wss"
+}
+
+func (s Wss) NewClientOptions(opt *Config) (*mqtt.ClientOptions, error) {
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(opt.Url)
+	opts.SetClientID(opt.ClientId)
+	opts.SetAutoReconnect(false) // use custom reconnection algorithm with offline queueing
+
+	return opts, nil
+}

device.go

@@ -50,7 +50,11 @@ type DeviceClient struct {
 }
 
 func New(opt *Options) *DeviceClient {
-	p, err := awsiotprotocol.ByName(opt.Protocol)
+	if opt.Protocol != "" || opt.Host != "" {
+		log.Printf("Options.Protocol and Options.Host is deprecated. Use Options.URL instead.")
+		opt.Url = opt.Protocol + "://" + opt.Host
+	}
+	p, err := awsiotprotocol.ByUrl(opt.Url)
 	if err != nil {
 		panic(err)
 	}
@@ -60,7 +64,7 @@ func New(opt *Options) *DeviceClient {
 			CertPath: opt.CertPath,
 			CaPath:   opt.CaPath,
 			ClientId: opt.ClientId,
-			Host:     opt.Host,
+			Url:      opt.Url,
 		},
 	)
 	if err != nil {

options.go

@@ -33,8 +33,9 @@ type Options struct {
 	MaximumReconnectTime     time.Duration
 	MinimumConnectionTime    time.Duration
 	Keepalive                time.Duration
-	Protocol                 string
-	Host                     string
+	Url                      string
+	Protocol                 string // [deprecated] use Url
+	Host                     string // [deprecated] use Url
 	Debug                    bool
 	Qos                      byte
 	Retain                   bool

sample/main.go

@@ -45,7 +45,7 @@ var (
 	caPath      = flag.String("ca-path", "CAfile.pem", "Path to CA certificate pem file")
 	thingName   = flag.String("thing-name", "sample", "Thing name")
 	region      = flag.String("region", "ap-northeast-1", "AWS region")
-	host        = flag.String("host", "hoge.iot.ap-northeast-1.amazonaws.com", "AWS IoT host")
+	url         = flag.String("url", "mqtts://hoge.iot.ap-northeast-1.amazonaws.com", "AWS IoT endpoint")
 )
 
 func main() {
@@ -61,8 +61,7 @@ func main() {
 		MaximumReconnectTime:     time.Second * 2,
 		MinimumConnectionTime:    time.Second * 2,
 		Keepalive:                time.Second * 2,
-		Protocol:                 "mqtts",
-		Host:                     *host,
+		Url:                      *url,
 		Debug:                    false,
 		Qos:                      1,
 		Retain:                   false,