[ADDED] MQTT Support

[kozlovic]

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

[kozlovic]

@derekcollison Obviously too much to review, but if you could focus on areas that existed before but have been modified for MQTT. Namely consumer.go, jetstream.go. I also needed a reverse-match for sublist, so also have a look at that.

[derekcollison]

Will take a look.

[derekcollison]

In general LGTM. Did not look too closely at mqtt code and tests but integration points.

[kozlovic]

@derekcollison I have addressed some of the comments, namely using non public stream/consumer config options. I am now going to look at others, like createClient() and if I really need the subscription to be created before calling processSub, etc..

[kozlovic]

@derekcollison Good news, I was able to revert changes to processSub() and still maintain code correctness/no race with mqtt subscriptions. So now moving to other remarks such as createClient(), etc..

[derekcollison]

Just ping me when you want me to take another look. We are very close IMO..

[derekcollison]

LGTM - minor comments.

[derekcollison]

Maybe make more generic?

switch c.clientType() {
  case NATS:
  case MQTT:
  case WS:
}

Or something like that?

[kozlovic]

Please check commit 2ca0794 when you get a chance. I am not sold on that and I am afraid that it gets mis-used.. meaning you still want to check for c.kind == CLIENT when you deal with routing of messages for instance. So you would need extendedKind() only in specific situations (like the places I use it in this commit).
I can easily revert this commit or have another approach.

[derekcollison]

The enum for what gets returned should include a NON_CLIENT option as well.

[kozlovic]

I guess you miss the fact that this is extendedKind(), not Type() which means that it can be invoked on any type of connection and will return say ROUTER, LEAF, but for a client, it may return CLIENT (if regular) or the new CLIENT_MQTT or CLIENT_WS.
What you suggest (and I do have a stashed version of that) is to have a clientType() that should be invoked only for c.kind == CLIENT and that will then return NATS/MQTT/WS or NON_CLIENT if incorrectly invoked on a non CLIENT connection.

[derekcollison]

Yes the second option is what I had in mind, but minor so..

[kozlovic]

Ok, replaced that commit with 035cffa, which I believe does what you asked.

[derekcollison]

LGTM

[kozlovic]

@derekcollison I could have merged and then work on optimization, but I decided to include this one in the main PR. If could could review 1d7c471 when you have time. I still have some work to do with conversion of topic/subject for delivery of messages to subscriptions. If I am not done soon with that one, we could merge this PR and will continue on perf improvement later.

[derekcollison]

LGTM