Add MQTT-over-WebSocket tests and client option

Author: Alexander Menzhinsky

cmd/iothub-device/main.go

@@ -18,7 +18,7 @@ import (
 
 var transports = map[string]func() (transport.Transport, error){
 	"mqtt": func() (transport.Transport, error) {
-		return mqtt.New(), nil
+		return mqtt.New(mqtt.WithWebSocket(wsFlag)), nil
 	},
 	"amqp": func() (transport.Transport, error) {
 		return nil, errors.New("not implemented")
@@ -29,6 +29,7 @@ var transports = map[string]func() (transport.Transport, error){
 }
 
 var (
+	wsFlag        bool
 	debugFlag     bool
 	formatFlag    string
 	quiteFlag     bool
@@ -63,6 +64,7 @@ $IOTHUB_DEVICE_CONNECTION_STRING environment variable is required unless you use
 func run() error {
 	ctx := context.Background()
 	return internal.New(help, func(f *flag.FlagSet) {
+		f.BoolVar(&wsFlag, "ws", false, "enable MQTT-over-WebSocket transport")
 		f.BoolVar(&debugFlag, "debug", false, "enable debug mode")
 		f.StringVar(&formatFlag, "format", "json-pretty", "data output format <json|json-pretty>")
 		f.StringVar(&transportFlag, "transport", "mqtt", "transport to use <mqtt|amqp|http>")

iotdevice/transport/mqtt/mqtt.go

@@ -49,9 +49,9 @@ func WithClientOptionsConfig(fn func(opts *mqtt.ClientOptions)) TransportOption
 
 // WithWebSocket makes the mqtt client use MQTT over WebSockets on port 443,
 // which is great if e.g. port 8883 is blocked.
-func WithWebSocket() TransportOption {
+func WithWebSocket(enable bool) TransportOption {
 	return func(tr *Transport) {
-		tr.webSocket = true
+		tr.webSocket = enable
 	}
 }
 
@@ -82,7 +82,7 @@ type Transport struct {
 
 	logger logger.Logger
 	cocfg  func(opts *mqtt.ClientOptions)
-	
+
 	webSocket bool
 }
 
@@ -105,7 +105,7 @@ func (tr *Transport) Connect(ctx context.Context, creds transport.Credentials) e
 	}
 
 	tlsCfg := &tls.Config{
-		RootCAs: common.RootCAs(),
+		RootCAs:       common.RootCAs(),
 		Renegotiation: tls.RenegotiateOnceAsClient,
 	}
 	if crt := creds.GetCertificate(); crt != nil {

tests/end2end_test.go

@@ -66,7 +66,8 @@ func TestEnd2End(t *testing.T) {
 	}
 
 	for name, mktransport := range map[string]func() transport.Transport{
-		"mqtt": func() transport.Transport { return mqtt.New() },
+		"mqtt":    func() transport.Transport { return mqtt.New() },
+		"mqtt-ws": func() transport.Transport { return mqtt.New(mqtt.WithWebSocket(true)) },
 		// TODO: "amqp": func() transport.Transport { return amqp.New() },
 		// TODO: "http": func() transport.Transport { return http.New() },
 	} {