Commit Graph

9 Commits

Author SHA1 Message Date
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
415a7071a7 Tweaks to mqttProcessConnect()
The test TestMQTTPersistedSession() flapped once on GA. It turns
out that when the server was sending CONNACK the test was immediately
using a NATS publisher to send a message that was not received by
the MQTT subscription for the recovered session.

Sending the CONNACK before restoring subscriptions allowed for a
window where a different connection could publish and messages
would be missed. It is technically ok, I think, and test could
have been easily fixed to ensure that we don't NATS publish before
the session is fully restored.

However, I have changed the order to first restore subscriptions
then send the CONNACK. The way locking happens with MQTT subscriptions
we are sure that the CONNACK will be sent first because even if
there are inflight messages, the MQTT callbacks will have to wait
for the session lock under which the subscriptions are restored
and the CONNACK sent.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-12-03 17:57:51 -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
41fac39f8e Split createClient() into versions for normal, WS and MQTT clients.
This duplicate quite a bit of code, but reduces the conditionals
in the createClient() function.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-12-02 13:50:50 -07:00
Ivan Kozlovic
4fc04d3f55 Revert changes to processSub()
Based on how the MQTT callback operates, it is safe to finish setup
of the MQTT subscriptions after processSub() returns. So I have
reverted the changes to processSub() which will minimize changes
to non-MQTT related code.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
2020-12-01 15:38:47 -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
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