MF-596 - Add subtopic to RawMessage

cli/message.go

@@ -14,7 +14,7 @@ const contentTypeSenml = "application/senml+json"
 var cmdMessages = []cobra.Command{
 	cobra.Command{
 		Use:   "send",
-		Short: "send <channel_id> <JSON_string> <thing_key>",
+		Short: "send <channel_id>[.<subtopic>...] <JSON_string> <thing_key>",
 		Long:  `Sends message on the channel`,
 		Run: func(cmd *cobra.Command, args []string) {
 			if len(args) != 3 {
@@ -32,7 +32,7 @@ var cmdMessages = []cobra.Command{
 	},
 	cobra.Command{
 		Use:   "read",
-		Short: "read <channel_id> <thing_key>",
+		Short: "read <channel_id>[.<subtopic>...] <thing_key>",
 		Long:  `Reads all channel messages`,
 		Run: func(cmd *cobra.Command, args []string) {
 			if len(args) != 2 {

http/api/logging.go

@@ -31,7 +31,11 @@ func LoggingMiddleware(svc mainflux.MessagePublisher, logger log.Logger) mainflu
 
 func (lm *loggingMiddleware) Publish(msg mainflux.RawMessage) (err error) {
 	defer func(begin time.Time) {
-		message := fmt.Sprintf("Method publish took %s to complete", time.Since(begin))
+		destChannel := msg.Channel
+		if msg.Subtopic != "" {
+			destChannel = fmt.Sprintf("%s.%s", destChannel, msg.Subtopic)
+		}
+		message := fmt.Sprintf("Method publish to channel %s took %s to complete", destChannel, time.Since(begin))
 		if err != nil {
 			lm.logger.Warn(fmt.Sprintf("%s with error: %s.", message, err))
 			return

http/api/transport.go

@@ -13,6 +13,8 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"regexp"
+	"strings"
 	"time"
 
 	kithttp "github.com/go-kit/kit/transport/http"
@@ -27,8 +29,9 @@ import (
 const protocol = "http"
 
 var (
-	errMalformedData = errors.New("malformed request data")
-	auth             mainflux.ThingsServiceClient
+	errMalformedData  = errors.New("malformed request data")
+	auth              mainflux.ThingsServiceClient
+	channelPartRegExp = regexp.MustCompile(`^/channels/([\w\-]+)/messages((/[\w\-]+)*)*\??.*$`)
 )
 
 // MakeHandler returns a HTTP handler for API endpoints.
@@ -41,7 +44,7 @@ func MakeHandler(svc mainflux.MessagePublisher, tc mainflux.ThingsServiceClient)
 
 	r := bone.New()
 
-	r.Post("/channels/:id/messages", kithttp.NewServer(
+	r.Post("/channels/*", kithttp.NewServer(
 		sendMessageEndpoint(svc),
 		decodeRequest,
 		encodeResponse,
@@ -55,7 +58,19 @@ func MakeHandler(svc mainflux.MessagePublisher, tc mainflux.ThingsServiceClient)
 }
 
 func decodeRequest(_ context.Context, r *http.Request) (interface{}, error) {
-	publisher, err := authorize(r)
+	channelParts := channelPartRegExp.FindStringSubmatch(r.RequestURI)
+	if len(channelParts) < 2 {
+		return nil, errMalformedData
+	}
+
+	chanID := channelParts[1]
+	subtopic := strings.Replace(channelParts[2], "/", ".", -1)
+	if subtopic != "" {
+		// channelParts[2] contains the subtopic parts starting with char /
+		subtopic = subtopic[1:]
+	}
+
+	publisher, err := authorize(r, chanID)
 	if err != nil {
 		return nil, err
 	}
@@ -65,35 +80,25 @@ func decodeRequest(_ context.Context, r *http.Request) (interface{}, error) {
 		return nil, err
 	}
 
-	channel := bone.GetValue(r, "id")
-	if channel == "" {
-		return nil, errMalformedData
-	}
-
 	msg := mainflux.RawMessage{
 		Publisher:   publisher,
 		Protocol:    protocol,
 		ContentType: r.Header.Get("Content-Type"),
-		Channel:     channel,
+		Channel:     chanID,
+		Subtopic:    subtopic,
 		Payload:     payload,
 	}
 
 	return msg, nil
 }
 
-func authorize(r *http.Request) (string, error) {
+func authorize(r *http.Request, chanID string) (string, error) {
 	apiKey := r.Header.Get("Authorization")
 
 	if apiKey == "" {
 		return "", things.ErrUnauthorizedAccess
 	}
 
-	// extract ID from /channels/:id/messages
-	chanID := bone.GetValue(r, "id")
-	if chanID == "" {
-		return "", errMalformedData
-	}
-
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 

http/nats/publisher.go

@@ -16,6 +16,8 @@ import (
 	broker "github.com/nats-io/go-nats"
 )
 
+const prefix = "channel"
+
 var _ mainflux.MessagePublisher = (*natsPublisher)(nil)
 
 type natsPublisher struct {
@@ -33,6 +35,9 @@ func (pub *natsPublisher) Publish(msg mainflux.RawMessage) error {
 		return err
 	}
 
-	subject := fmt.Sprintf("channel.%s", msg.Channel)
+	subject := fmt.Sprintf("%s.%s", prefix, msg.Channel)
+	if msg.Subtopic != "" {
+		subject = fmt.Sprintf("%s.%s", subject, msg.Subtopic)
+	}
 	return pub.nc.Publish(subject, data)
 }