NOISSUE - Raise code coverage in ws adapter (#242)

* Fix failed subscription handling in ws adapter

Fix unsubscribing bug in ws adapter.

Signed-off-by: Aleksandar Novakovic <anovakovic01@gmail.com>

* Add subscription fail and publish fail test cases

Update mock implementation to receive publish error in order to
support these test cases.

Signed-off-by: Aleksandar Novakovic <anovakovic01@gmail.com>

* Update mainflux version to 0.2.3

Update project version and load tests version to 0.2.3.

Signed-off-by: Aleksandar Novakovic <anovakovic01@gmail.com>

* Update version endpoint response format

Signed-off-by: Aleksandar Novakovic <anovakovic01@gmail.com>

docs/getting-started.md

@@ -19,7 +19,7 @@ docker-compose -f docker/docker-compose.yml up -d
 Use the Mainflux API to create user account:
 
 ```
-curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json; charset=utf-8" https://localhost/users -d '{"email":"john.doe@email.com", "password":"123"}'
+curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json" https://localhost/users -d '{"email":"john.doe@email.com", "password":"123"}'
 ```
 
 Note that when using official `docker-compose`, all services are behind `nginx`
@@ -31,7 +31,7 @@ In order for this user to be able to authenticate to the system, you will have
 to create an authorization token for him:
 
 ```
-curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json; charset=utf-8" https://localhost/tokens -d '{"email":"john.doe@email.com", "password":"123"}'
+curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json" https://localhost/tokens -d '{"email":"john.doe@email.com", "password":"123"}'
 ```
 
 Response should look like this:
@@ -54,7 +54,7 @@ Devices are provisioned by executing request `POST /clients`, with a
 that belong to this particular user.
 
 ```
-curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json; charset=utf-8" -H "Authorization: <user_auth_token>" https://localhost/clients -d '{"type":"device", "name":"weio"}'
+curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json" -H "Authorization: <user_auth_token>" https://localhost/clients -d '{"type":"device", "name":"weio"}'
 ```
 
 Response will contain `Location` header whose value represents path to newly
@@ -62,7 +62,7 @@ created client:
 
 ```
 HTTP/1.1 201 Created
-Content-Type: application/json; charset=utf-8
+Content-Type: application/json
 Location: /clients/81380742-7116-4f6f-9800-14fe464f6773
 Date: Tue, 10 Apr 2018 10:02:59 GMT
 Content-Length: 0
@@ -74,7 +74,7 @@ Applications are provisioned by executing HTTP request `POST /clients`, with
 `"type":"app"` specified in JSON payload.
 
 ```
-curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json; charset=utf-8" -H "Authorization: <user_auth_token>" https://localhost/clients -d '{"type":"app", "name":"myapp"}'
+curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json" -H "Authorization: <user_auth_token>" https://localhost/clients -d '{"type":"app", "name":"myapp"}'
 ```
 
 Response will contain `Location` header whose value represents path to newly
@@ -82,7 +82,7 @@ created client (same as for devices):
 
 ```
 HTTP/1.1 201 Created
-Content-Type: application/json; charset=utf-8
+Content-Type: application/json
 Location: /clients/cb63f852-2d48-44f0-a0cf-e450496c6c92
 Date: Tue, 10 Apr 2018 10:33:17 GMT
 Content-Length: 0
@@ -102,7 +102,7 @@ Notice that you will receive only those clients that were provisioned by
 
 ```
 HTTP/1.1 200 OK
-Content-Type: application/json; charset=utf-8
+Content-Type: application/json
 Date: Tue, 10 Apr 2018 10:50:12 GMT
 Content-Length: 1105
 
@@ -148,7 +148,7 @@ curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X DELETE
 Channels are provisioned by executing request `POST /channels`:
 
 ```
-curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json; charset=utf-8" -H "Authorization: <user_auth_token>" https://localhost/channels -d '{"name":"mychan"}'
+curl -s -S -i --cacert docker/ssl/certs/mainflux-server.crt --insecure -X POST -H "Content-Type: application/json" -H "Authorization: <user_auth_token>" https://localhost/channels -d '{"name":"mychan"}'
 ```
 
 After sending request you should receive response with `Location` header that
@@ -156,7 +156,7 @@ contains path to newly created channel:
 
 ```
 HTTP/1.1 201 Created
-Content-Type: application/json; charset=utf-8
+Content-Type: application/json
 Location: /channels/19daa7a8-a489-4571-8714-ef1a214ed914
 Date: Tue, 10 Apr 2018 11:30:07 GMT
 Content-Length: 0
@@ -176,7 +176,7 @@ token's owner.
 
 ```
 HTTP/1.1 200 OK
-Content-Type: application/json; charset=utf-8
+Content-Type: application/json
 Date: Tue, 10 Apr 2018 11:38:06 GMT
 Content-Length: 139
 

http/api/transport.go

@@ -41,7 +41,7 @@ func MakeHandler(svc mainflux.MessagePublisher, mc manager.ManagerClient) http.H
 		opts...,
 	))
 
-	r.GetFunc("/version", mainflux.Version())
+	r.GetFunc("/version", mainflux.Version("http"))
 	r.Handle("/metrics", promhttp.Handler())
 
 	return r

load-test/build.sbt

@@ -1,7 +1,7 @@
 enablePlugins(GatlingPlugin)
 
 name := "load-test"
-version := "0.2.2"
+version := "0.2.3"
 
 scalaVersion := "2.12.4"
 

load-test/src/test/scala/com/mainflux/loadtest/simulations/CreateAndRetrieveClientSimulation.scala

@@ -59,7 +59,7 @@ class CreateAndRetrieveClientSimulation extends Simulation {
 }
 
 object CreateAndRetrieveClientSimulation {
-  val ContentType = "application/json; charset=utf-8"
+  val ContentType = "application/json"
   val User = """{"email":"john.doe@email.com", "password":"123"}"""
   val Client = """{"type":"device", "name":"weio"}"""
 }

load-test/src/test/scala/com/mainflux/loadtest/simulations/PublishSimulation.scala

@@ -90,7 +90,7 @@ class PublishSimulation extends Simulation {
 }
 
 object PublishSimulation {
-  val ContentType = "application/json; charset=utf-8"
+  val ContentType = "application/json"
   val User = """{"email":"john.doe@email.com", "password":"123"}"""
   val Client = """{"type":"device", "name":"weio"}"""
   val Channel = """{"name":"mychan"}"""

manager/api/transport.go

@@ -143,7 +143,7 @@ func MakeHandler(svc manager.Service) http.Handler {
 		opts...,
 	))
 
-	r.GetFunc("/version", mainflux.Version())
+	r.GetFunc("/version", mainflux.Version("manager"))
 	r.Handle("/metrics", promhttp.Handler())
 
 	return r

normalizer/api.go

@@ -11,7 +11,7 @@ import (
 // MakeHandler returns a HTTP handler for API endpoints.
 func MakeHandler() http.Handler {
 	r := bone.New()
-	r.GetFunc("/version", mainflux.Version())
+	r.GetFunc("/version", mainflux.Version("normalizer"))
 	r.Handle("/metrics", promhttp.Handler())
 
 	return r

version.go

@@ -5,16 +5,17 @@ import (
 	"net/http"
 )
 
-const version string = "0.2.2"
+const version string = "0.2.3"
 
 type response struct {
-	Version string
+	Service string `json:"service"`
+	Version string `json:"version"`
 }
 
 // Version exposes an HTTP handler for retrieving service version.
-func Version() http.HandlerFunc {
+func Version(service string) http.HandlerFunc {
 	return http.HandlerFunc(func(rw http.ResponseWriter, _ *http.Request) {
-		res := response{Version: version}
+		res := response{service, version}
 
 		data, _ := json.Marshal(res)
 

ws/adapter_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/mainflux/mainflux/ws"
 	"github.com/mainflux/mainflux/ws/mocks"
+	broker "github.com/nats-io/go-nats"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/mainflux/mainflux"
@@ -29,7 +30,7 @@ var (
 
 func newService() ws.Service {
 	subs := map[string]ws.Channel{chanID: channel}
-	pubsub := mocks.NewService(subs)
+	pubsub := mocks.NewService(subs, broker.ErrInvalidMsg)
 	return ws.New(pubsub)
 }
 

ws/api/transport.go

@@ -38,7 +38,7 @@ func MakeHandler(svc ws.Service, mc manager.ManagerClient, l log.Logger) http.Ha
 
 	mux := bone.New()
 	mux.GetFunc("/channels/:id/messages", handshake(svc))
-	mux.GetFunc("/version", mainflux.Version())
+	mux.GetFunc("/version", mainflux.Version("websocket"))
 	mux.Handle("/metrics", promhttp.Handler())
 
 	return mux
@@ -69,9 +69,9 @@ func handshake(svc ws.Service) http.HandlerFunc {
 		// Subscribe to channel
 		channel := ws.Channel{make(chan mainflux.RawMessage), make(chan bool)}
 		sub.channel = channel
-		if err = svc.Subscribe(sub.chanID, sub.channel); err != nil {
+		if err := svc.Subscribe(sub.chanID, sub.channel); err != nil {
 			logger.Warn(fmt.Sprintf("Failed to subscribe to NATS subject: %s", err))
-			w.WriteHeader(http.StatusExpectationFailed)
+			conn.Close()
 			return
 		}
 		go sub.listen()
@@ -137,6 +137,7 @@ func (sub subscription) broadcast(svc ws.Service) {
 		if err := svc.Publish(msg); err != nil {
 			logger.Warn(fmt.Sprintf("Failed to publish message to NATS: %s", err))
 			if err == ws.ErrFailedConnection {
+				sub.conn.Close()
 				sub.channel.Closed <- true
 				return
 			}

ws/api/transport_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/mainflux/mainflux/ws"
 	"github.com/mainflux/mainflux/ws/api"
 	"github.com/mainflux/mainflux/ws/mocks"
+	broker "github.com/nats-io/go-nats"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -31,7 +32,7 @@ var (
 
 func newService() ws.Service {
 	subs := map[string]ws.Channel{chanID: channel}
-	pubsub := mocks.NewService(subs)
+	pubsub := mocks.NewService(subs, broker.ErrConnectionClosed)
 	return ws.New(pubsub)
 }
 
@@ -88,11 +89,14 @@ func TestHandshake(t *testing.T) {
 		header bool
 		token  string
 		status int
+		msg    []byte
 	}{
-		{"connect and send message", chanID, true, token, http.StatusSwitchingProtocols},
-		{"connect with invalid token", chanID, true, "", http.StatusForbidden},
-		{"connect with invalid channel id", "1", true, token, http.StatusNotFound},
-		{"connect and send message with token as query parameter", chanID, false, token, http.StatusSwitchingProtocols},
+		{"connect and send message", chanID, true, token, http.StatusSwitchingProtocols, msg},
+		{"connect to non-existent channel", "123e4567-e89b-12d3-a456-000000000042", true, token, http.StatusSwitchingProtocols, []byte{}},
+		{"connect with invalid token", chanID, true, "", http.StatusForbidden, []byte{}},
+		{"connect with invalid channel id", "1", true, token, http.StatusNotFound, []byte{}},
+		{"connect and send message with token as query parameter", chanID, false, token, http.StatusSwitchingProtocols, msg},
+		{"connect and send message that cannot be published", chanID, true, token, http.StatusSwitchingProtocols, []byte{}},
 	}
 
 	for _, tc := range cases {
@@ -101,7 +105,7 @@ func TestHandshake(t *testing.T) {
 		if err != nil {
 			continue
 		}
-		err = conn.WriteMessage(websocket.TextMessage, msg)
+		err = conn.WriteMessage(websocket.TextMessage, tc.msg)
 		assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s\n", tc.desc, err))
 	}
 }

ws/channel_test.go

@@ -0,0 +1,18 @@
+package ws_test
+
+import (
+	"testing"
+
+	"github.com/mainflux/mainflux"
+	"github.com/mainflux/mainflux/ws"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClose(t *testing.T) {
+	channel := ws.Channel{make(chan mainflux.RawMessage), make(chan bool)}
+	channel.Close()
+	_, closed := <-channel.Closed
+	_, messagesClosed := <-channel.Messages
+	assert.False(t, closed, "channel closed stayed open")
+	assert.False(t, messagesClosed, "channel messages stayed open")
+}

ws/mocks/messages.go

@@ -3,23 +3,23 @@ package mocks
 import (
 	"github.com/mainflux/mainflux"
 	"github.com/mainflux/mainflux/ws"
-	broker "github.com/nats-io/go-nats"
 )
 
 var _ ws.Service = (*mockService)(nil)
 
 type mockService struct {
 	subscriptions map[string]ws.Channel
+	pubError      error
 }
 
 // NewService returns mock message publisher.
-func NewService(subs map[string]ws.Channel) ws.Service {
-	return mockService{subs}
+func NewService(subs map[string]ws.Channel, pubError error) ws.Service {
+	return mockService{subs, pubError}
 }
 
 func (svc mockService) Publish(msg mainflux.RawMessage) error {
 	if len(msg.Payload) == 0 {
-		return broker.ErrInvalidMsg
+		return svc.pubError
 	}
 	svc.subscriptions[msg.Channel].Messages <- msg
 	return nil

ws/nats/pubsub.go

@@ -59,13 +59,20 @@ func (pubsub *natsPubSub) Subscribe(chanID string, channel ws.Channel) error {
 			return
 		}
 
+		// Prevents sending message to closed channel
 		select {
 		case channel.Messages <- rawMsg:
 		case <-channel.Closed:
 			sub.Unsubscribe()
-			channel.Close()
 		}
 	})
 
+	// Check if subscription should be closed
+	go func() {
+		<-channel.Closed
+		sub.Unsubscribe()
+		channel.Close()
+	}()
+
 	return err
 }