Added QoS and retained flags

controller/endpoint/endpoint.go

@@ -53,6 +53,8 @@ type Endpoint struct {
 		Host      string
 		Port      int
 		QueueName string
+		Qos       byte
+		Retained  bool
 	}
 }
 
@@ -328,6 +330,40 @@ func parseEndpoint(s string) (Endpoint, error) {
 			}
 		}
 
+		// Parsing additional params
+		if len(sqp) > 1 {
+			m, err := url.ParseQuery(sqp[1])
+			if err != nil {
+				return endpoint, errors.New("invalid MQTT url")
+			}
+			for key, val := range m {
+				if len(val) == 0 {
+					continue
+				}
+				switch key {
+				case "qos":
+					n, err := strconv.ParseUint(val[0], 10, 8)
+					if err != nil {
+						return endpoint, errors.New("invalid MQTT qos value")
+					}
+					endpoint.MQTT.Qos = byte(n)
+				case "retained":
+					n, err := strconv.ParseUint(val[0], 10, 8)
+					if err != nil {
+						return endpoint, errors.New("invalid MQTT retained value")
+					}
+
+					if n != 1 && n != 0 {
+						return endpoint, errors.New("invalid MQTT retained, should be [0, 1]")
+					}
+
+					if n == 1 {
+						endpoint.MQTT.Retained = true
+					}
+				}
+			}
+		}
+
 		// Throw error if we not provide any queue name
 		if endpoint.MQTT.QueueName == "" {
 			return endpoint, errors.New("missing MQTT topic name")

controller/endpoint/mqtt.go

@@ -63,7 +63,7 @@ func (conn *MQTTEndpointConn) Send(msg string) error {
 		conn.conn = c
 	}
 
-	t := conn.conn.Publish(conn.ep.MQTT.QueueName, 0, false, msg)
+	t := conn.conn.Publish(conn.ep.MQTT.QueueName, conn.ep.MQTT.Qos, conn.ep.MQTT.Retained, msg)
 	t.Wait()
 
 	if t.Error() != nil {