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
 

docs/load-test.md

@@ -0,0 +1,23 @@
+## Test scenarios
+
+Testing environment to be determined.
+
+### Message publishing
+
+In this scenario, large number of requests are sent to HTTP adapter service 
+every second. This test checks how much time HTTP adapter took to response to 
+each request.
+
+#### Results
+
+TBD
+
+### Create and get client
+
+In this scenario, large number of requests are sent to manager service to create 
+client, and than to retrieve its data. This test checks how much time manager
+service took to response to each request.
+
+#### Results
+
+TBD

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/.gitignore

@@ -0,0 +1,8 @@
+target/
+.classpath
+.cache-tests
+.cache-main
+.settings/
+.project
+*.class
+bin/

load-test/README.md

@@ -0,0 +1,49 @@
+# Load Test
+
+This SBT project contains load tests written for mainflux platform.
+
+## Setup
+
+In order to run load tests you must have [openjdk8](http://openjdk.java.net/install/) and [sbt](https://www.scala-sbt.org/1.0/docs/Setup.html) installed on your machine.
+
+## Configuration
+
+Tests are configured to use variables from `JAVA_OPTS` presented in the
+following table. Note that any unset variables will be replaced with their
+default values.
+
+| Variable | Description                              | Default               |
+|----------|------------------------------------------|-----------------------|
+| manager  | Manager service URL                      | http://localhost:8180 |
+| http     | HTTP adapter service URL                 | http://localhost:8182 |
+| requests | Number of requests to be sent per second | 100                   |
+
+## Usage
+
+This project contains two simulations:
+
+- `PublishSimulation`
+- `CreateAndRetrieveClientSimulation`
+
+To run all tests you will have to run following commands:
+
+```bash
+cd <path_to_mainflux_project>/load-test
+sbt gatling:test
+```
+
+### Publish Simulation
+
+`PublishSimulation` contains load tests for publishing messages. To run this test use following command:
+
+```bash
+sbt "gatling:testOnly com.mainflux.loadtest.simulations.PublishSimulation"
+```
+
+### Create And Retrieve Client Simulation
+
+`CreateAndRetrieveClientSimulation` contains load tests for creating and retrieving clients. To run this test use following command:
+
+```bash
+sbt "gatling:testOnly com.mainflux.loadtest.simulations.CreateAndRetrieveClientSimulation"
+```

load-test/build.sbt

@@ -0,0 +1,18 @@
+enablePlugins(GatlingPlugin)
+
+name := "load-test"
+version := "0.2.3"
+
+scalaVersion := "2.12.4"
+
+val gatlingVersion = "2.3.1"
+val circeVersion = "0.9.3"
+
+libraryDependencies ++= Seq(
+  "io.gatling.highcharts" %  "gatling-charts-highcharts" % gatlingVersion,
+  "io.gatling"            %  "gatling-test-framework"    % gatlingVersion,
+  "org.scalaj"            %% "scalaj-http"               % "2.3.0",
+  "io.circe"              %% "circe-core"                % circeVersion,
+  "io.circe"              %% "circe-generic"             % circeVersion,
+  "io.circe"              %% "circe-parser"              % circeVersion
+)

load-test/project/build.properties

@@ -0,0 +1 @@
+sbt.version=1.1.2

load-test/project/plugins.sbt

@@ -0,0 +1 @@
+addSbtPlugin("io.gatling" % "gatling-sbt" % "2.2.2")

load-test/src/test/resources/logback.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+	<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+		<encoder>
+			<pattern>%d{HH:mm:ss.SSS} [%-5level] %logger{15} - %msg%n%rEx</pattern>
+		</encoder>
+        <immediateFlush>false</immediateFlush>
+	</appender>
+	<appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">
+		<appender-ref ref="CONSOLE" />
+	</appender>
+	<root level="WARN">
+		<appender-ref ref="ASYNC" />
+	</root>
+</configuration>

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

@@ -0,0 +1,7 @@
+package com.mainflux.loadtest.simulations
+
+object Constants {
+  val ManagerUrl = System.getProperty("manager", "http://localhost:8180")
+  val HttpAdapterUrl = System.getProperty("http", "http://localhost:8182")
+  val RequestsPerSecond = Integer.getInteger("requests", 100)
+}

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

@@ -0,0 +1,65 @@
+package com.mainflux.loadtest.simulations
+
+import scala.concurrent.duration._
+import scalaj.http.Http
+import io.gatling.core.Predef._
+import io.gatling.http.Predef._
+import io.gatling.jdbc.Predef._
+import io.circe._
+import io.circe.generic.auto._
+import io.circe.parser._
+import io.circe.syntax._
+import CreateAndRetrieveClientSimulation._
+import io.gatling.http.protocol.HttpProtocolBuilder.toHttpProtocol
+import io.gatling.http.request.builder.HttpRequestBuilder.toActionBuilder
+import com.mainflux.loadtest.simulations.Constants._
+
+class CreateAndRetrieveClientSimulation extends Simulation {
+
+  // Register user
+  Http(s"${ManagerUrl}/users")
+    .postData(User)
+    .header(HttpHeaderNames.ContentType, ContentType)
+    .asString
+
+  // Login user
+  val tokenRes = Http(s"${ManagerUrl}/tokens")
+    .postData(User)
+    .header(HttpHeaderNames.ContentType, ContentType)
+    .asString
+    .body
+
+  val tokenCursor = parse(tokenRes).getOrElse(Json.Null).hcursor
+  val token = tokenCursor.downField("token").as[String].getOrElse("")
+
+  // Prepare testing scenario
+  val httpProtocol = http
+    .baseURL(ManagerUrl)
+    .inferHtmlResources()
+    .acceptHeader("*/*")
+    .contentTypeHeader(ContentType)
+    .userAgentHeader("curl/7.54.0")
+
+  val scn = scenario("CreateAndGetClient")
+    .exec(http("CreateClientRequest")
+      .post("/clients")
+      .header(HttpHeaderNames.ContentType, ContentType)
+      .header(HttpHeaderNames.Authorization, token)
+      .body(StringBody(Client))
+      .check(status.is(201))
+      .check(headerRegex(HttpHeaderNames.Location, "(.*)").saveAs("location")))
+    .exec(http("GetClientRequest")
+      .get("${location}")
+      .header(HttpHeaderNames.Authorization, token)
+      .check(status.is(200)))
+
+  setUp(
+    scn.inject(
+      constantUsersPerSec(RequestsPerSecond.toDouble) during (15 second))).protocols(httpProtocol)
+}
+
+object CreateAndRetrieveClientSimulation {
+  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

@@ -0,0 +1,98 @@
+package com.mainflux.loadtest.simulations
+
+import scala.concurrent.duration._
+import scalaj.http.Http
+import io.gatling.core.Predef._
+import io.gatling.http.Predef._
+import io.gatling.jdbc.Predef._
+import io.circe._
+import io.circe.generic.auto._
+import io.circe.parser._
+import io.circe.syntax._
+import PublishSimulation._
+import io.gatling.http.protocol.HttpProtocolBuilder.toHttpProtocol
+import io.gatling.http.request.builder.HttpRequestBuilder.toActionBuilder
+import com.mainflux.loadtest.simulations.Constants._
+
+class PublishSimulation extends Simulation {
+
+  // Register user
+  Http(s"${ManagerUrl}/users")
+    .postData(User)
+    .header(HttpHeaderNames.ContentType, ContentType)
+    .asString
+
+  // Login user
+  val tokenRes = Http(s"${ManagerUrl}/tokens")
+    .postData(User)
+    .header(HttpHeaderNames.ContentType, ContentType)
+    .asString
+    .body
+
+  val tokenCursor = parse(tokenRes).getOrElse(Json.Null).hcursor
+  val token = tokenCursor.downField("token").as[String].getOrElse("")
+
+  // Register client
+  val clientLocation = Http(s"${ManagerUrl}/clients")
+    .postData(Client)
+    .header(HttpHeaderNames.Authorization, token)
+    .header(HttpHeaderNames.ContentType, ContentType)
+    .asString
+    .headers.get("Location").get(0)
+
+  val clientId = clientLocation.split("/")(2)
+
+  // Get client key
+  val clientRes = Http(s"${ManagerUrl}/clients/${clientId}")
+    .header(HttpHeaderNames.Authorization, token)
+    .header(HttpHeaderNames.ContentType, ContentType)
+    .asString
+    .body
+
+  val clientCursor = parse(clientRes).getOrElse(Json.Null).hcursor
+  val clientKey = clientCursor.downField("key").as[String].getOrElse("")
+
+  // Register channel
+  val chanLocation = Http(s"${ManagerUrl}/channels")
+    .postData(Channel)
+    .header(HttpHeaderNames.Authorization, token)
+    .header(HttpHeaderNames.ContentType, ContentType)
+    .asString
+    .headers.get("Location").get(0)
+
+  val chanId = chanLocation.split("/")(2)
+
+  // Connect client to channel
+  Http(s"${ManagerUrl}/channels/${chanId}/clients/${clientId}")
+    .method("PUT")
+    .header(HttpHeaderNames.Authorization, token)
+    .asString
+
+  // Prepare testing scenario
+  val httpProtocol = http
+    .baseURL(HttpAdapterUrl)
+    .inferHtmlResources()
+    .acceptHeader("*/*")
+    .contentTypeHeader("application/json; charset=utf-8")
+    .userAgentHeader("curl/7.54.0")
+
+  val scn = scenario("PublishMessage")
+    .exec(http("PublishMessageRequest")
+      .post(s"/channels/${chanId}/messages")
+      .header(HttpHeaderNames.ContentType, "application/senml+json")
+      .header(HttpHeaderNames.Authorization, clientKey)
+      .body(StringBody(Message))
+      .check(status.is(202)))
+
+  setUp(
+    scn.inject(
+      constantUsersPerSec(RequestsPerSecond.toDouble) during (15 second))).protocols(httpProtocol)
+}
+
+object PublishSimulation {
+  val ContentType = "application/json"
+  val User = """{"email":"john.doe@email.com", "password":"123"}"""
+  val Client = """{"type":"device", "name":"weio"}"""
+  val Channel = """{"name":"mychan"}"""
+  val Message = """[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]"""
+}

manager/api/endpoint_test.go

@@ -316,28 +316,35 @@ func TestListClients(t *testing.T) {
 		client.Key = id
 		clients = append(clients, client)
 	}
-
+	clientURL := fmt.Sprintf("%s/clients", ts.URL)
 	cases := []struct {
 		desc   string
 		auth   string
 		status int
-		offset int
-		limit  int
+		url    string
 		res    []manager.Client
 	}{
-		{"get a list of clients", user.Email, http.StatusOK, 1, 5, clients[1:6]},
-		{"get a list of clients with invalid token", invalidEmail, http.StatusForbidden, 0, 1, nil},
-		{"get a list of clients with invalid offset", user.Email, http.StatusBadRequest, -1, 5, nil},
-		{"get a list of clients with invalid limit", user.Email, http.StatusBadRequest, 1, -5, nil},
-		{"get a list of clients with zero limit", user.Email, http.StatusBadRequest, 1, 0, nil},
-		{"get a list of clients with limit greater than max", user.Email, http.StatusBadRequest, 0, 110, nil},
+		{"get a list of clients", user.Email, http.StatusOK, fmt.Sprintf("%s?offset=%d&limit=%d", clientURL, 0, 5), clients[0:5]},
+		{"get a list of clients with invalid token", invalidEmail, http.StatusForbidden, fmt.Sprintf("%s?offset=%d&limit=%d", clientURL, 0, 1), nil},
+		{"get a list of clients with invalid offset", user.Email, http.StatusBadRequest, fmt.Sprintf("%s?offset=%d&limit=%d", clientURL, -1, 5), nil},
+		{"get a list of clients with invalid limit", user.Email, http.StatusBadRequest, fmt.Sprintf("%s?offset=%d&limit=%d", clientURL, 1, -5), nil},
+		{"get a list of clients with zero limit", user.Email, http.StatusBadRequest, fmt.Sprintf("%s?offset=%d&limit=%d", clientURL, 1, 0), nil},
+		{"get a list of clients with no offset provided", user.Email, http.StatusOK, fmt.Sprintf("%s?limit=%d", clientURL, 5), clients[0:5]},
+		{"get a list of clients with no limit provided", user.Email, http.StatusOK, fmt.Sprintf("%s?offset=%d", clientURL, 1), clients[1:11]},
+		{"get a list of clients with redundant query params", user.Email, http.StatusOK, fmt.Sprintf("%s?offset=%d&limit=%d&value=something", clientURL, 0, 5), clients[0:5]},
+		{"get a list of clients with limit greater than max", user.Email, http.StatusBadRequest, fmt.Sprintf("%s?offset=%d&limit=%d", clientURL, 0, 110), nil},
+		{"get a list of clients with default URL", user.Email, http.StatusOK, fmt.Sprintf("%s%s", clientURL, ""), clients[0:10]},
+		{"get a list of clients with invalid URL", user.Email, http.StatusBadRequest, fmt.Sprintf("%s%s", clientURL, "?%%"), nil},
+		{"get a list of clients with invalid number of params", user.Email, http.StatusBadRequest, fmt.Sprintf("%s%s", clientURL, "?offset=4&limit=4&limit=5&offset=5"), nil},
+		{"get a list of clients with invalid offset", user.Email, http.StatusBadRequest, fmt.Sprintf("%s%s", clientURL, "?offset=e&limit=5"), nil},
+		{"get a list of clients with invalid limit", user.Email, http.StatusBadRequest, fmt.Sprintf("%s%s", clientURL, "?offset=5&limit=e"), nil},
 	}
 
 	for _, tc := range cases {
 		req := testRequest{
 			client: cli,
 			method: http.MethodGet,
-			url:    fmt.Sprintf("%s/clients?offset=%d&limit=%d", ts.URL, tc.offset, tc.limit),
+			url:    tc.url,
 			token:  tc.auth,
 		}
 		res, err := req.make()
@@ -347,19 +354,6 @@ func TestListClients(t *testing.T) {
 		assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
 		assert.ElementsMatch(t, tc.res, data["clients"], fmt.Sprintf("%s: expected body %s got %s", tc.desc, tc.res, data["clients"]))
 	}
-	req := testRequest{
-		client: cli,
-		method: http.MethodGet,
-		url:    fmt.Sprintf("%s/clients", ts.URL),
-		token:  user.Email,
-	}
-	defaults := "get a list of clients with no limit and offset params"
-	res, err := req.make()
-	assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", defaults, err))
-	var data map[string][]manager.Client
-	json.NewDecoder(res.Body).Decode(&data)
-	assert.Equal(t, http.StatusOK, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", defaults, http.StatusOK, res.StatusCode))
-	assert.ElementsMatch(t, clients[0:10], data["clients"], fmt.Sprintf("%s: expected body %s got %s", defaults, clients[0:10], data["clients"]))
 }
 
 func TestRemoveClient(t *testing.T) {
@@ -538,28 +532,36 @@ func TestListChannels(t *testing.T) {
 		channel.ID = id
 		channels = append(channels, channel)
 	}
+	channelURL := fmt.Sprintf("%s/channels", ts.URL)
 
 	cases := []struct {
 		desc   string
 		auth   string
 		status int
-		offset int
-		limit  int
+		url    string
 		res    []manager.Channel
 	}{
-		{"get a list of channels", user.Email, http.StatusOK, 1, 5, channels[1:6]},
-		{"get a list of channels with invalid token", invalidEmail, http.StatusForbidden, 0, 1, nil},
-		{"get a list of channels with invalid offset", user.Email, http.StatusBadRequest, -1, 5, nil},
-		{"get a list of channels with invalid limit", user.Email, http.StatusBadRequest, 1, -5, nil},
-		{"get a list of channels with zero limit", user.Email, http.StatusBadRequest, 1, 0, nil},
-		{"get a list of channels with limit greater than max", user.Email, http.StatusBadRequest, 0, 110, nil},
+		{"get a list of channels", user.Email, http.StatusOK, fmt.Sprintf("%s?offset=%d&limit=%d", channelURL, 0, 6), channels[0:6]},
+		{"get a list of channels with invalid token", invalidEmail, http.StatusForbidden, fmt.Sprintf("%s?offset=%d&limit=%d", channelURL, 0, 1), nil},
+		{"get a list of channels with invalid offset", user.Email, http.StatusBadRequest, fmt.Sprintf("%s?offset=%d&limit=%d", channelURL, -1, 5), nil},
+		{"get a list of channels with invalid limit", user.Email, http.StatusBadRequest, fmt.Sprintf("%s?offset=%d&limit=%d", channelURL, -1, 5), nil},
+		{"get a list of channels with zero limit", user.Email, http.StatusBadRequest, fmt.Sprintf("%s?offset=%d&limit=%d", channelURL, 1, 0), nil},
+		{"get a list of channels with no offset provided", user.Email, http.StatusOK, fmt.Sprintf("%s?limit=%d", channelURL, 5), channels[0:5]},
+		{"get a list of channels with no limit provided", user.Email, http.StatusOK, fmt.Sprintf("%s?offset=%d", channelURL, 1), channels[1:11]},
+		{"get a list of channels with redundant query params", user.Email, http.StatusOK, fmt.Sprintf("%s?offset=%d&limit=%d&value=something", channelURL, 0, 5), channels[0:5]},
+		{"get a list of channels with limit greater than max", user.Email, http.StatusBadRequest, fmt.Sprintf("%s?offset=%d&limit=%d", channelURL, 0, 110), nil},
+		{"get a list of channels with default URL", user.Email, http.StatusOK, fmt.Sprintf("%s%s", channelURL, ""), channels[0:10]},
+		{"get a list of channels with invalid URL", user.Email, http.StatusBadRequest, fmt.Sprintf("%s%s", channelURL, "?%%"), nil},
+		{"get a list of channels with invalid number of params", user.Email, http.StatusBadRequest, fmt.Sprintf("%s%s", channelURL, "?offset=4&limit=4&limit=5&offset=5"), nil},
+		{"get a list of channels with invalid offset", user.Email, http.StatusBadRequest, fmt.Sprintf("%s%s", channelURL, "?offset=e&limit=5"), nil},
+		{"get a list of channels with invalid limit", user.Email, http.StatusBadRequest, fmt.Sprintf("%s%s", channelURL, "?offset=5&limit=e"), nil},
 	}
 
 	for _, tc := range cases {
 		req := testRequest{
 			client: client,
 			method: http.MethodGet,
-			url:    fmt.Sprintf("%s/channels?offset=%d&limit=%d", ts.URL, tc.offset, tc.limit),
+			url:    tc.url,
 			token:  tc.auth,
 		}
 		res, err := req.make()
@@ -569,19 +571,6 @@ func TestListChannels(t *testing.T) {
 		assert.Equal(t, tc.status, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", tc.desc, tc.status, res.StatusCode))
 		assert.ElementsMatch(t, tc.res, body["channels"], fmt.Sprintf("%s: expected body %s got %s", tc.desc, tc.res, body["channels"]))
 	}
-	req := testRequest{
-		client: client,
-		method: http.MethodGet,
-		url:    fmt.Sprintf("%s/channels", ts.URL),
-		token:  user.Email,
-	}
-	defaults := "get a list of channels with no limit and offset params"
-	res, err := req.make()
-	assert.Nil(t, err, fmt.Sprintf("%s: unexpected error %s", defaults, err))
-	var data map[string][]manager.Channel
-	json.NewDecoder(res.Body).Decode(&data)
-	assert.Equal(t, http.StatusOK, res.StatusCode, fmt.Sprintf("%s: expected status code %d got %d", defaults, http.StatusOK, res.StatusCode))
-	assert.ElementsMatch(t, channels[0:10], data["channels"], fmt.Sprintf("%s: expected body %s got %s", defaults, channels[0:10], data["channels"]))
 }
 
 func TestRemoveChannel(t *testing.T) {

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
@@ -254,28 +254,13 @@ func decodeView(_ context.Context, r *http.Request) (interface{}, error) {
 }
 
 func decodeList(_ context.Context, r *http.Request) (interface{}, error) {
-
 	q, err := url.ParseQuery(r.URL.RawQuery)
 	if err != nil {
-		return nil, err
+		return nil, errInvalidQueryParams
 	}
 	offset := 0
 	limit := 10
 
-	n := len(q)
-	if n == 0 {
-		req := listResourcesReq{
-			key:    r.Header.Get("Authorization"),
-			offset: offset,
-			limit:  limit,
-		}
-		return req, nil
-	}
-
-	if n > 2 {
-		return nil, errInvalidQueryParams
-	}
-
 	off, lmt := q["offset"], q["limit"]
 
 	if len(off) > 1 || len(lmt) > 1 {
@@ -285,14 +270,14 @@ func decodeList(_ context.Context, r *http.Request) (interface{}, error) {
 	if len(off) == 1 {
 		offset, err = strconv.Atoi(off[0])
 		if err != nil {
-			return nil, err
+			return nil, errInvalidQueryParams
 		}
 	}
 
 	if len(lmt) == 1 {
 		limit, err = strconv.Atoi(lmt[0])
 		if err != nil {
-			return nil, err
+			return nil, errInvalidQueryParams
 		}
 	}
 	req := listResourcesReq{
@@ -346,6 +331,8 @@ func encodeError(_ context.Context, err error, w http.ResponseWriter) {
 		w.WriteHeader(http.StatusConflict)
 	case errUnsupportedContentType:
 		w.WriteHeader(http.StatusUnsupportedMediaType)
+	case errInvalidQueryParams:
+		w.WriteHeader(http.StatusBadRequest)
 	case io.ErrUnexpectedEOF:
 		w.WriteHeader(http.StatusBadRequest)
 	case io.EOF:

manager/mocks/channels.go

@@ -71,11 +71,12 @@ func (crm *channelRepositoryMock) All(owner string, offset, limit int) []manager
 		return channels
 	}
 
-	first := fmt.Sprintf("%s%012d", chanId, offset)
-	last := fmt.Sprintf("%s%012d", chanId, offset+limit)
+	// Since IDs starts from 1, shift everything by one.
+	first := fmt.Sprintf("%s%012d", chanId, offset+1)
+	last := fmt.Sprintf("%s%012d", chanId, offset+limit+1)
 
 	for k, v := range crm.channels {
-		if strings.HasPrefix(k, prefix) && v.ID > first && v.ID <= last {
+		if strings.HasPrefix(k, prefix) && v.ID >= first && v.ID < last {
 			channels = append(channels, v)
 		}
 	}

manager/mocks/clients.go

@@ -75,11 +75,12 @@ func (crm *clientRepositoryMock) All(owner string, offset, limit int) []manager.
 		return clients
 	}
 
-	first := fmt.Sprintf("%s%012d", cliId, offset)
-	last := fmt.Sprintf("%s%012d", cliId, offset+limit)
+	// Since IDs start from 1, shift everything by one.
+	first := fmt.Sprintf("%s%012d", cliId, offset+1)
+	last := fmt.Sprintf("%s%012d", cliId, offset+limit+1)
 
 	for k, v := range crm.clients {
-		if strings.HasPrefix(k, prefix) && v.ID > first && v.ID <= last {
+		if strings.HasPrefix(k, prefix) && v.ID >= first && v.ID < last {
 			clients = append(clients, v)
 		}
 	}

mkdocs.yml

@@ -41,3 +41,4 @@ pages:
     - License: LICENSE.txt
 - Architecture: architecture.md
 - Getting started: getting-started.md
+- Load test: load-test.md

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.1"
+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
 }