Commit Graph

13 Commits

Author SHA1 Message Date
Ivan Kozlovic
61bd1b8d86 MQTT clustering
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2021-02-19 08:50:00 -07:00
Derek Collison
c16f6e193d Move JetStream direct APIs to private.
Signed-off-by: Derek Collison <derek@nats.io>
2021-02-07 15:19:22 -08:00
Ivan Kozlovic
af57f55738 Fixing some flappers (leafnode and mqtt)
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2021-01-26 14:23:49 -07:00
Ángel
07c3214473 Added tests for MQTT bearer token authentication
Signed-off-by: Ángel <angelo.fly1@gmail.com>
2021-01-25 18:44:28 +01:00
Derek Collison
f0cdf89c61 JetStream Clustering WIP
Signed-off-by: Derek Collison <derek@nats.io>
2021-01-14 01:14:52 -08:00
Ivan Kozlovic
1dd485531f Updates to MQTT
- Keep track of published topic to avoid conversion in some case
- Rework tracing of some MQTT protocols
- Rework topic-to-subject conversion
- Update some tests

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-12-10 10:55:41 -07:00
Ivan Kozlovic
1d7c4712a5 Increase Pub performance
Essentially make publish a zero alloc. Use c.mqtt.pp as the parser
publish packet structure. Messages were initially copied because
MQTT messages don't have CR_LF but was adding it so that it worked
for NATS pub/subs and MQTT pub/subs.
Now an MQTT producer sending to NATS sub will queue CR_LF after
payload.

Here is result of benchcmp for MQTT pub runs only:
```
benchmark                                     old ns/op     new ns/op     delta
BenchmarkMQTT_QoS0_Pub_______0b_Payload-8     157           55.6          -64.59%
BenchmarkMQTT_QoS0_Pub_______8b_Payload-8     167           61.0          -63.47%
BenchmarkMQTT_QoS0_Pub______32b_Payload-8     181           65.3          -63.92%
BenchmarkMQTT_QoS0_Pub_____128b_Payload-8     243           78.1          -67.86%
BenchmarkMQTT_QoS0_Pub_____256b_Payload-8     298           95.0          -68.12%
BenchmarkMQTT_QoS0_Pub_______1K_Payload-8     604           224           -62.91%
BenchmarkMQTT_QoS1_Pub_______0b_Payload-8     1713          1314          -23.29%
BenchmarkMQTT_QoS1_Pub_______8b_Payload-8     1703          1311          -23.02%
BenchmarkMQTT_QoS1_Pub______32b_Payload-8     1722          1345          -21.89%
BenchmarkMQTT_QoS1_Pub_____128b_Payload-8     2105          1432          -31.97%
BenchmarkMQTT_QoS1_Pub_____256b_Payload-8     2148          1503          -30.03%
BenchmarkMQTT_QoS1_Pub_______1K_Payload-8     3024          1889          -37.53%

benchmark                                     old MB/s     new MB/s     speedup
BenchmarkMQTT_QoS0_Pub_______0b_Payload-8     31.76        89.91        2.83x
BenchmarkMQTT_QoS0_Pub_______8b_Payload-8     77.79        213.01       2.74x
BenchmarkMQTT_QoS0_Pub______32b_Payload-8     204.52       566.26       2.77x
BenchmarkMQTT_QoS0_Pub_____128b_Payload-8     550.65       1715.96      3.12x
BenchmarkMQTT_QoS0_Pub_____256b_Payload-8     877.77       2757.16      3.14x
BenchmarkMQTT_QoS0_Pub_______1K_Payload-8     1705.02      4607.72      2.70x
BenchmarkMQTT_QoS1_Pub_______0b_Payload-8     6.42         8.37         1.30x
BenchmarkMQTT_QoS1_Pub_______8b_Payload-8     11.16        14.49        1.30x
BenchmarkMQTT_QoS1_Pub______32b_Payload-8     24.97        31.97        1.28x
BenchmarkMQTT_QoS1_Pub_____128b_Payload-8     66.52        97.74        1.47x
BenchmarkMQTT_QoS1_Pub_____256b_Payload-8     124.78       178.27       1.43x
BenchmarkMQTT_QoS1_Pub_______1K_Payload-8     342.64       548.32       1.60x

benchmark                                     old allocs     new allocs     delta
BenchmarkMQTT_QoS0_Pub_______0b_Payload-8     3              0              -100.00%
BenchmarkMQTT_QoS0_Pub_______8b_Payload-8     3              0              -100.00%
BenchmarkMQTT_QoS0_Pub______32b_Payload-8     3              0              -100.00%
BenchmarkMQTT_QoS0_Pub_____128b_Payload-8     4              0              -100.00%
BenchmarkMQTT_QoS0_Pub_____256b_Payload-8     4              0              -100.00%
BenchmarkMQTT_QoS0_Pub_______1K_Payload-8     4              0              -100.00%
BenchmarkMQTT_QoS1_Pub_______0b_Payload-8     5              2              -60.00%
BenchmarkMQTT_QoS1_Pub_______8b_Payload-8     5              2              -60.00%
BenchmarkMQTT_QoS1_Pub______32b_Payload-8     5              2              -60.00%
BenchmarkMQTT_QoS1_Pub_____128b_Payload-8     7              3              -57.14%
BenchmarkMQTT_QoS1_Pub_____256b_Payload-8     7              3              -57.14%
BenchmarkMQTT_QoS1_Pub_______1K_Payload-8     7              3              -57.14%

benchmark                                     old bytes     new bytes     delta
BenchmarkMQTT_QoS0_Pub_______0b_Payload-8     80            0             -100.00%
BenchmarkMQTT_QoS0_Pub_______8b_Payload-8     88            0             -100.00%
BenchmarkMQTT_QoS0_Pub______32b_Payload-8     120           0             -100.00%
BenchmarkMQTT_QoS0_Pub_____128b_Payload-8     224           0             -100.00%
BenchmarkMQTT_QoS0_Pub_____256b_Payload-8     369           1             -99.73%
BenchmarkMQTT_QoS0_Pub_______1K_Payload-8     1250          31            -97.52%
BenchmarkMQTT_QoS1_Pub_______0b_Payload-8     106           28            -73.58%
BenchmarkMQTT_QoS1_Pub_______8b_Payload-8     122           28            -77.05%
BenchmarkMQTT_QoS1_Pub______32b_Payload-8     154           28            -81.82%
BenchmarkMQTT_QoS1_Pub_____128b_Payload-8     381           157           -58.79%
BenchmarkMQTT_QoS1_Pub_____256b_Payload-8     655           287           -56.18%
BenchmarkMQTT_QoS1_Pub_______1K_Payload-8     2312          1078          -53.37%
```

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-12-04 14:42:37 -07:00
Ivan Kozlovic
cf9ba928ca Fixed some MQTT tests
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-12-02 17:00:47 -07:00
Ivan Kozlovic
67425d23c8 Add c.isMqtt() and c.isWebsocket()
This hides the check on "c.mqtt != nil" or "c.ws != nil".
Added some tests.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-12-02 15:52:06 -07:00
Ivan Kozlovic
3e91ef75ab Some updates based on code review
- Added non-public stream and consumer configuration options to
achieve the "no subject" and "no interest" capabilities. Had
to implement custom FileStreamInfo and FileConsumerInfo marshal/
unmarshal methods so that those non public fields can be
persisted/recovered properly.
- Restored some of JS original code (since now can use config
instead of passing booleans to the functions).
- Use RLock for deliveryFormsCycle() check (unrelated to MQTT).
- Removed restriction on creating streams with MQTT prefix.
- Preventing API deletion of internal streams and their consumers.
- Added comment on Sublist's ReverseMatch method.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-12-01 14:05:54 -07:00
Ivan Kozlovic
718c995914 Allow "nats" utility to display internal MQTT streams
MQTT streams are special in that we do not set subjects in the config
since they capture all subjects. Otherwise, we would have been forced
to create a stream on say "MQTT.>" but then all publishes would have
to be prefixed with "MQTT." in order for them to be captured.

However, if one uses the "nats" tool to inspect those streams, the tool
would fail with:

```
server response is not a valid "io.nats.jetstream.api.v1.stream_info_response" message:
(root): Must validate one and only one schema (oneOf)
config: subjects is required
config: Must validate all the schemas (allOf)
```

To solve that, if we detect that user asks for the MQTT streams, we
artificially set the returned config's subject to ">".

Alternatively, we may want to not return those streams at all, although
there may be value to see the info for mqtt streams/consumers.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-11-30 20:08:44 -07:00
Ivan Kozlovic
ac4890acba Fixed flapper
Tests dealing with MQTT "will" needed to wait for the server to
process the MQTT client close of the connection. Only then we
have the guarantee that the server produced the "will" message.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-11-30 20:08:44 -07:00
Ivan Kozlovic
1dba6418ed [ADDED] MQTT Support
This PR introduces native support for MQTT clients. It requires use
of accounts with JetStream enabled. Since as of now clustering is
not available, MQTT will be limited to single instance.

Only QoS 0 and 1 are supported at the moment. MQTT clients can
exchange messages with NATS clients and vice-versa.

Since JetStream is required, accounts with JetStream enabled must
exist in order for an MQTT client to connect to the NATS Server.
The administrator can limit the users that can use MQTT with the
allowed_connection_types option in the user section. For instance:
```
accounts {
  mqtt {
    users [
      {user: all, password: pwd, allowed_connection_types: ["STANDARD", "WEBSOCKET", "MQTT"]}
      {user: mqtt_only, password: pwd, allowed_connection_types: "MQTT"}
    ]
    jetstream: enabled
  }
}
```
The "mqtt_only" can only be used for MQTT connections, which the user
"all" accepts standard, websocket and MQTT clients.

Here is what a configuration to enable MQTT looks like:
```
mqtt {
  # Specify a host and port to listen for websocket connections
  #
  # listen: "host:port"

  # It can also be configured with individual parameters,
  # namely host and port.
  #
  # host: "hostname"
  port: 1883

  # TLS configuration section
  #
  # tls {
  #  cert_file: "/path/to/cert.pem"
  #  key_file: "/path/to/key.pem"
  #  ca_file: "/path/to/ca.pem"
  #
  #  # Time allowed for the TLS handshake to complete
  #  timeout: 2.0
  #
  #  # Takes the user name from the certificate
  #  #
  #  # verify_an_map: true
  #}

  # Authentication override. Here are possible options.
  #
  # authorization {
  #   # Simple username/password
  #   #
  #   user: "some_user_name"
  #   password: "some_password"
  #
  #   # Token. The server will check the MQTT's password in the connect
  #   # protocol against this token.
  #   #
  #   # token: "some_token"
  #
  #   # Time allowed for the client to send the MQTT connect protocol
  #   # after the TCP connection is established.
  #   #
  #   timeout: 2.0
  #}

  # If an MQTT client connects and does not provide a username/password and
  # this option is set, the server will use this client (and therefore account).
  #
  # no_auth_user: "some_user_name"

  # This is the time after which the server will redeliver a QoS 1 message
  # sent to a subscription that has not acknowledged (PUBACK) the message.
  # The default is 30 seconds.
  #
  # ack_wait: "1m"

  # This limits the number of QoS1 messages sent to a session without receiving
  # acknowledgement (PUBACK) from that session. MQTT specification defines
  # a packet identifier as an unsigned int 16, which means that the maximum
  # value is 65535. The default value is 1024.
  #
  # max_ack_pending: 100
}
```

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-11-30 20:08:44 -07:00