nmqtt.nim

## Native Nim MQTT client library and binaries
##
## zevv (https://github.com/zevv) & ThomasTJdev (https://github.com/ThomasTJdev) & python36 (https://github.com/python36)

import std/[strutils,
            asyncnet,
            net,
            asyncdispatch,
            tables,
            monotimes,
            times]

when defined(broker):
  import
    cligen,
    sequtils,
    times,
    random,
    nmqtt/utils/passwords,
    nmqtt/utils/version
  from parsecfg import loadConfig, getSectionValue
  from os import fileExists

type
  QoS* = range[0..2]

  PubCallback = object
    cb: proc(topic: string, message: string)
    qos: QoS

  TopicCb = tuple[templ: seq[string], cb: PubCallback]

  MqttCtx* = ref object
    host: string
    port: Port
    sslOn: bool
    sslCert: string
    sslKey: string
    verbosity: int
    beenConnected: bool
    username: string
    password: string
    state: State
    clientId: string
    s: AsyncSocket
    ssl: SslContext
    msgIdSeq: MsgId
    workQueue: OrderedTable[MsgId, Work]
    pubCallbacks: Table[string, TopicCb]
    inWork: bool
    hasNewWorks: bool
    keepAlive: Duration
    keepAliveTimeout: Duration
    keepAliveMonitorRunned: bool
    pingRunned: bool
    lastSend: MonoTime
    awaitingPingResp: bool
    timeoutPingRespTime: MonoTime
    pingRespTimeout: Duration
    maxInflightMessages: int
    willFlag: bool
    willQoS: QoS
    willRetain: bool
    willTopic: string
    willMsg: string

    #when defined(broker):
    proto: string
    version: uint8
    connFlags: string
    subscribed: Table[string, QoS] # Topic, QoS
    lastAction: MonoTime # Check keepAlive

  #when defined(broker):
  ConnAckFlag = enum
    ConnAcc               = 0x00
    ConnRefProtocol       = 0x01
    ConnRefRejected       = 0x02
    ConnRefUnavailable    = 0x03
    ConnRefBadUserPwd     = 0x04
    ConnRefNotAuthorized  = 0x05

  State = enum
    Disabled, Disconnected, Connecting, Connected, Disconnecting, Error

  MsgId = uint16

  PktType = enum
    Notype      =  0
    Connect     =  1
    ConnAck     =  2
    Publish     =  3
    PubAck      =  4
    PubRec      =  5
    PubRel      =  6
    PubComp     =  7
    Subscribe   =  8
    SubAck      =  9
    Unsubscribe = 10
    Unsuback    = 11
    PingReq     = 12
    PingResp    = 13
    Disconnect  = 14

  ConnectFlag = enum
    WillQoS0     = 0x00
    CleanSession = 0x02
    WillFlag     = 0x04
    WillQoS1     = 0x08
    WillQoS2     = 0x10
    WillRetain   = 0x20
    PasswordFlag = 0x40
    UserNameFlag = 0x80

  Pkt = object
    typ: PktType
    flags: uint8
    data: seq[uint8]

  WorkKind = enum
    PubWork, SubWork

  WorkState = enum
    WorkNew, WorkSent, WorkAcked

  Work = ref object
    state: WorkState
    msgId: MsgId
    topic: string
    qos: QoS
    typ: PktType
    flags: uint16 #when defined(broker)
    case wk: WorkKind
    of PubWork:
      retain: bool
      message: string
    of SubWork:
      discard

when defined(broker):
  type
    TopicCtx = tuple[templ: seq[string], ctxs: seq[MqttCtx]]

    MqttBroker* = ref object
      host: string
      port: Port
      sslOn: bool
      sslCert: string
      sslKey: string
      verbosity: int
      connections: Table[string, MqttCtx]
      retained: Table[string, RetainedMsg] # Topic, RetaindMsg
      subscribers: Table[string, TopicCtx]
      version: uint8
      clientIdMaxLen: int
      clientKickOld: bool
      emptyClientId: bool
      spacesInClientId: bool
      passClientId: bool
      maxConnections: int
      passwords: Table[string, string]

    RetainedMsg = object
      msg: string
      qos: QoS
      time: float
      clientid: string

when defined(broker):
  var
    mqttbroker = MqttBroker()
    r = initRand(toInt(epochTime()))

#
# Packet helpers
#

proc put(pkt: var Pkt, v: uint16) =
  pkt.data.add((v.int /%  256).uint8)
  pkt.data.add((v.int mod 256).uint8)

proc put(pkt: var Pkt, v: uint8) =
  pkt.data.add(v)

proc put(pkt: var Pkt, data: string, withLen: bool) =
  if withLen:
    pkt.put(data.len.uint16)
  for c in data:
    pkt.put(c.uint8)

proc getu8(pkt: Pkt, offset: int): (uint8, int) =
  let val = pkt.data[offset]
  result = (val, offset+1)

proc getQoS(pkt: Pkt, offset: int): (QoS, int) =
  let val = pkt.data[offset]
  result = (QoS(val), offset+1)

proc getu16(pkt: Pkt, offset: int): (uint16, int) =
  let val = (pkt.data[offset].int*256 + pkt.data[offset+1].int).uint16
  result = (val, offset+2)

proc getstring(pkt: Pkt, offset: int, withLen: bool): (string, int) =
  var val: string
  if withLen:
    var (len, offset2) = pkt.getu16(offset)
    for i in 0..<len.int:
      val.add(pkt.data[offset+i+2].char)
    result = (val, offset2+len.int)
  else:
    for i in offset..<pkt.data.len:
      val.add(pkt.data[i].char)
    result = (val, pkt.data.len)

when defined(broker):
  proc getstring(pkt: Pkt, offset: int, len: int): (string, int) =
    var val: string
    for i in offset..<len+offset:
      val.add(pkt.data[i].char)
    result = (val, len+offset)

when defined(broker):
  proc getbin(pkt: Pkt, b: int): (string, int) =
    result = (toBin(parseBiggestInt($pkt.data[b]), 8), b+1)

proc `$`(pkt: Pkt): string =
  result.add $pkt.typ & "(" & $pkt.flags.toHex & "): "
  for b in pkt.data:
    result.add(b.toHex)
    result.add(" ")

proc newPkt(typ: PktType = NOTYPE, flags: uint8 = 0): Pkt =
  result.typ = typ
  result.flags = flags

#
# Debug
#
proc dmp(ctx: MqttCtx, s: string) =
  when not defined(broker):
    if defined(dev) or ctx.verbosity >= 2:
      stderr.write("\e[1;30m" & s & "\e[0m\n")
  when defined(broker):
    if defined(dev) or mqttbroker.verbosity >= 2:
      stderr.write("\e[1;30m" & s & "\e[0m\n")
  when defined(test):
    let s = split(s, " ")
    testDmp.add(@[$(s[0] & " " & s[1]), $join(s[2..s.len-1], " ")])

proc dbg(ctx: MqttCtx, s: string) =
  stderr.write("\e[37m" & s & "\e[0m\n")

proc verbose(s: string) =
  stderr.write("\e[37m" & s & "\e[0m\n")

when defined(broker):
  proc verbose(e: string, s: Table) =
    var output: string
    for t, c in s:
      if output != "":
        output.add(", ")
      when c is RetainedMsg:
        output.add("{" & t & "}")
      else:
        output.add("{" & t & ": " & $c.ctxs.len & "}")
    stderr.write("\e[37m" & e & " >> " & output & "\e[0m\n")

  proc verbose(ctx: auto) =
    var output: string
    for t, c in fieldPairs(ctx[]):
      when c is AsyncSocket:
        let conn = if c.isClosed(): "..disconnected.." else: "..connected.."
        output.add("  " & t & ": " & conn & "\n")
      when c is Table:
        output.add("  " & t & ": " & $c.len & "\n")
      when c is MqttCtx:
        output.add("  " & t & ": " & $c.clientid & "\n")
      when c is string or c is bool or c is int or c is uint8 or c is uint16:
        output.add("  " & t & ": " & $c & "\n")

    when ctx is MqttBroker:
      stderr.write("\e[37m" & "Broker      >>\n" & output & "\e[0m\n")
    when ctx is MqttCtx:
      stderr.write("\e[37m" & "Client      >> " & ctx.clientid & "\n" & output & "\e[0m\n")

proc wrn(ctx: MqttCtx, s: string) =
  stderr.write("\e[1;31mWarning >> " & s & "\e[0m\n")

proc wrn(s: string) =
  stderr.write("\e[1;31mWarning >> " & s & "\e[0m\n")

#
# Subscribers
#

proc matchTopic(templ: seq[string], topic: string): bool {.inline.} =
  var i = 0
  for t in templ:
    if t == "#":
      return i == 0 or i >= topic.len or topic[i] == '/'

    elif i > topic.len:
      return false

    else:
      if i > 0:
        while i < topic.len:
          if topic[i] == '/':
            break
          inc i

      if i + t.len > topic.len or t != topic[i ..< i + t.len]:
        return false
      i += t.len

    if i == topic.len:
      return true

proc splitTopic(topic: string): seq[string] =
  if topic == "#":
    @["#"]
  elif topic[^1] == '#':
    topic[0..^3].split('+') & @["#"]
  else:
    topic.split('+')

when defined(broker):
  proc retainMsg(topic: string, message: string, qos: QoS, clientid: string) =
    mqttbroker.retained[topic] = RetainedMsg(msg: message, qos: qos, clientid: clientid)

when defined(broker):
  proc addSubscriber*(ctx: MqttCtx, topic: string) {.async.} =
    ## Adds a subscriber to MqttBroker
    try:
      if mqttbroker.subscribers.hasKey(topic):
        for c in mqttbroker.subscribers[topic].ctxs:
          if c.clientId == ctx.clientId:
            return
        mqttbroker.subscribers[topic].ctxs.insert(ctx)
      else:
        mqttbroker.subscribers[topic] = (templ: splitTopic(topic), ctxs: @[ctx])
    except:
      wrn("Crash when adding a new subcriber")

when defined(broker):
  proc removeSubscriber*(ctx: MqttCtx, topic: string) {.async.} =
    ## Removes a subscriber from specific topic
    try:
      if mqttbroker.subscribers.hasKey(topic):
        mqttbroker.subscribers[topic].ctxs = filter(mqttbroker.subscribers[topic].ctxs, proc(x: MqttCtx): bool = x != ctx)
    except:
      wrn("Crash when removing subscriber with specific topic")

when defined(broker):
  proc removeSubscriber*(ctx: MqttCtx) {.async.} =
    ## Removes a subscriber without knowing the topics
    var delTop: seq[string]
    for t, c in mqttbroker.subscribers.mpairs():
      if ctx in c.ctxs:
        c.ctxs = filter(c.ctxs, proc(x: MqttCtx): bool = x != ctx)

        if c.ctxs.len() == 0:
          delTop.add(t)

    for t in delTop:
      mqttbroker.subscribers.del(t)

when defined(broker):
  proc qosAlign(qP, qS: QoS): QoS =
    ## Aligns the QOS for publisher and subscriber.
    if qP == qS:
      result = qP
    elif qP > qS:
      result = qS
    else:
      result = qP

when defined(broker):
  proc keepAliveMonitor(ctx: MqttCtx) {.async.}

#
# MQTT context
#

proc nextMsgId(ctx: MqttCtx): MsgId =
  inc ctx.msgIdSeq
  return ctx.msgIdSeq

proc hasInflightSlots(ctx: MqttCtx): bool =
  var cnt = 0
  for w in ctx.workQueue.values():
    if w.qos in {1, 2} and (w.typ != Publish or w.state == WorkSent):
      inc cnt
      if cnt == ctx.maxInflightMessages:
        return false
  true

proc sendDisconnect(ctx: MqttCtx): Future[bool] {.async.}

proc close(ctx: MqttCtx, reason: string) {.async.} =
  if ctx.state in {Connecting, Connected}:
    ctx.state = Disconnecting
    if ctx.verbosity >= 1:
      ctx.dbg("Closing: " & reason)
    discard await ctx.sendDisconnect()
    ctx.s.close()
    ctx.state = Disconnected

proc send(ctx: MqttCtx, pkt: Pkt): Future[bool] {.async.} =
  ## Send the packet
  if ctx.state notin {Connecting, Connected, Disconnecting}:
    return false

  var buf: seq[uint8] = newSeqOfCap[uint8](1 + 4 + pkt.data.len)
  buf.add (pkt.typ.int shl 4).uint8 or pkt.flags

  var len = pkt.data.len
  while true:
    var b = len mod 128
    len = len div 128
    if len > 0:
      b = b or 128
    buf.add(b.uint8)
    if len == 0:
      break

  # TODO:
  # only when test
  ctx.dmp("tx> " & $pkt)
  if pkt.data.len > 0:
    let hdrlen = buf.len
    buf.setLen(hdrlen + pkt.data.len)
    copyMem(buf[hdrlen].addr, pkt.data[0].unsafeAddr, pkt.data.len)

  ctx.lastSend = getMonoTime()
  await ctx.s.send(buf[0].unsafeAddr, buf.len)

  return true

proc recv(ctx: MqttCtx): Future[Pkt] {.async.} =
  ## Receive and parse the packet
  if ctx.state notin {Connecting, Connected}:
    return

  var r: int
  var b: uint8

  # TODO:
  # When the broker is running in SSL-mode, we need the try/except, since
  # we will encounter a silent crash in recvInto, when the client is
  # not actually using SSL.
  try:
    r = await ctx.s.recvInto(b.addr, b.sizeof)
  except:
    return
  if r != 1:
    when not defined(broker):
      await ctx.close("remote closed connection")
    return

  let typ = (b shr 4).PktType
  let flags = (b and 0x0f)
  var pkt = newPkt(typ, flags)

  var len: int
  var mul = 1
  for i in 0..3:
    var b: uint8
    r = await ctx.s.recvInto(b.addr, b.sizeof)

    if r != 1:
      when not defined(broker):
        await ctx.close("remote closed connection")
      return

    assert r == 1
    inc len, (b and 127).int * mul
    mul *= 128
    if ((b.int) and 0x80) == 0:
      break

  if len > 0:
    pkt.data.setLen(len)

    var offset = 0
    while offset < len:
      try:
        r = await ctx.s.recvInto(addr pkt.data[offset], len - offset)
      except:
        break
      if r == 0:
        break
      offset += r

    if offset != len:
      when not defined(broker):
        await ctx.close("remote closed connection")
      return

  ctx.dmp("rx> " & $pkt)
  return pkt

proc sendConnect(ctx: MqttCtx): Future[bool] =
  var flags: uint8
  flags = flags or CleanSession.uint8
  if ctx.willFlag:
    flags = flags or WillFlag.uint8
    if ctx.willQoS == 1:
      flags = flags or WillQoS1.uint8
    elif ctx.willQoS == 2:
      flags = flags or WillQoS2.uint8
    if ctx.willRetain:
      flags = flags or WillRetain.uint8

  if ctx.username != "":
    flags = flags or UserNameFlag.uint8
  if ctx.password != "":
    flags = flags or PasswordFlag.uint8

  var pkt = newPkt(Connect)
  pkt.put("MQTT", true)
  pkt.put(4.uint8)
  pkt.put(flags)
  pkt.put(uint16(ctx.keepAlive.inSeconds))
  pkt.put(ctx.clientId, true)

  if ctx.willFlag:
    pkt.put((ctx.willTopic.len).uint16)
    pkt.put(ctx.willTopic, false)
    pkt.put((ctx.willMsg.len).uint16)
    pkt.put(ctx.willMsg, false)
  if ctx.username != "":
    pkt.put(ctx.username, true)
  if ctx.password != "":
    pkt.put(ctx.password, true)
  result = ctx.send(pkt)

proc sendDisconnect(ctx: MqttCtx): Future[bool] =
  let pkt = newPkt(Disconnect, 0)
  result = ctx.send(pkt)

proc sendSubscribe(ctx: MqttCtx, msgId: MsgId, topic: string, qos: QoS): Future[bool] =
  var pkt = newPkt(Subscribe, 0b0010)
  pkt.put(msgId.uint16)
  pkt.put(topic, true)
  pkt.put(qos.uint8)
  result = ctx.send(pkt)

proc sendUnsubscribe(ctx: MqttCtx, msgId: MsgId, topic: string): Future[bool] =
  var pkt = newPkt(Unsubscribe, 0b0010)
  pkt.put(msgId.uint16)
  pkt.put(topic, true)
  result = ctx.send(pkt)

proc sendPublish(ctx: MqttCtx, msgId: MsgId, topic: string, message: string, qos: QoS, retain: bool): Future[bool] =
  var flags = (qos.uint8 shl 1).uint8
  if retain:
    flags = flags or 1
  var pkt = newPkt(Publish, flags)
  pkt.put(topic, true)
  if qos > 0:
    pkt.put(msgId.uint16)
  pkt.put(message, false)
  result = ctx.send(pkt)

proc sendPubAck(ctx: MqttCtx, msgId: MsgId, qos: QoS): Future[bool] =
  when defined(broker):
    var pkt = newPkt(PubAck)
  else:
    var pkt = newPkt(PubAck, 2)
  pkt.put(msgId.uint16)
  result = ctx.send(pkt)

proc sendPubRec(ctx: MqttCtx, msgId: MsgId): Future[bool] =
  when defined(broker):
    var pkt = newPkt(PubRec)
  else:
    var pkt = newPkt(PubRec, 2)
  pkt.put(msgId.uint16)
  result = ctx.send(pkt)

proc sendPubRel(ctx: MqttCtx, msgId: MsgId): Future[bool] =
  var pkt = newPkt(PubRel, 0b0010)
  pkt.put(msgId.uint16)
  result = ctx.send(pkt)

proc sendPubComp(ctx: MqttCtx, msgId: MsgId): Future[bool] =
  when defined(broker):
    var pkt = newPkt(PubComp)
  else:
    var pkt = newPkt(PubComp, 2)
  pkt.put(msgId.uint16)
  result = ctx.send(pkt)

proc sendPingReq(ctx: MqttCtx): Future[bool] =
  var pkt = newPkt(Pingreq)
  result = ctx.send(pkt)

#when defined(broker):
proc sendConnAck(ctx: MqttCtx, flags: uint16): Future[bool] =
  var pkt = newPkt(ConnAck)
  pkt.put(flags.uint16)
  result = ctx.send(pkt)

#when defined(broker):
proc sendSubAck(ctx: MqttCtx, msgId: MsgId, message: string): Future[bool] =
  var pkt = newPkt(SubAck)
  pkt.put(msgId.uint16)
  pkt.put(message, false)
  result = ctx.send(pkt)

#when defined(broker):
proc sendUnsubAck(ctx: MqttCtx, msgId: MsgId): Future[bool] =
  var pkt = newPkt(Unsuback)
  pkt.put(msgId.uint16)
  result = ctx.send(pkt)

#when defined(broker):
proc sendPingResp(ctx: MqttCtx): Future[bool] =
  var pkt = newPkt(PingResp)
  result = ctx.send(pkt)

proc sendWork(ctx: MqttCtx, work: Work): Future[bool] =
  case work.typ
  of Publish:   # Publish
    result = ctx.sendPublish(work.msgId, work.topic, work.message, work.qos, work.retain)

  of PubRel:    # Publish qos=2 (activated from a PubRec)
    result = ctx.sendPubRel(work.msgId)

  of PubAck:    # Subscribe qos=1 (activated from a Publish)
    result = ctx.sendPubAck(work.msgId, work.qos)

  of PubRec:    # Subscribe qos=2 (1/2) (activated from a Publish)
    result = ctx.sendPubRec(work.msgId)

  of PubComp:   # Subscribe qos=2 (2/2) (activated from a PubRel)
    result = ctx.sendPubComp(work.msgId)

  of Subscribe:
    result = ctx.sendSubscribe(work.msgId, work.topic, work.qos)

  of Unsubscribe:
    result = ctx.sendUnsubscribe(work.msgId, work.topic)

  of ConnAck:
    #when defined(broker):
    result = ctx.sendConnAck(work.flags)

  of PingResp:
    #when defined(broker):
    result = ctx.sendPingResp()

  of SubAck:
    #when defined(broker):
    result = ctx.sendSubAck(work.msgId, work.message)

  of Unsuback:
    #when defined(broker):
    result = ctx.sendUnsubAck(work.msgId)

  else:
    ctx.wrn("Error sending unknown package: " & $work.typ)

proc work(ctx: MqttCtx) {.async.} =
  ctx.hasNewWorks = true
  if ctx.inWork:
    return
  ctx.inWork = true

  while ctx.state == Connected and ctx.hasNewWorks:
    ctx.hasNewWorks = false
    let workQueue = ctx.workQueue

    for msgId, work in workQueue:

      #when defined(broker):
      if work.typ in [ConnAck, SubAck, UnsubAck, PingResp]:
        if await ctx.sendWork(work):
          ctx.workQueue.del(msgId)
          continue

      if work.wk == PubWork and work.state == WorkNew:
        if work.typ == Publish:
          if work.qos == 0:
            if await ctx.sendWork(work):
              ctx.workQueue.del(msgId)

          elif hasInflightSlots(ctx):
            if await ctx.sendWork(work):
              work.state = WorkSent

        elif work.typ == PubAck and work.qos == 1:
          if await ctx.sendWork(work):
            ctx.workQueue.del(msgId)

        elif work.typ == PubComp and work.qos == 2:
          if await ctx.sendWork(work):
            ctx.workQueue.del(msgId)

        else:
          if await ctx.sendWork(work):
            work.state = WorkSent

      #when not defined(broker):
      elif work.wk == SubWork and work.state == WorkNew:
        if work.typ == Subscribe:
          if await ctx.sendWork(work): work.state = WorkSent

        elif work.typ == Unsubscribe:
          if await ctx.sendWork(work):
            work.state = WorkSent
            ctx.pubCallbacks.del(work.topic)

  ctx.inWork = false

when defined(broker):
  proc sendWill(ctx: MqttCtx) {.async.} =
    ## Send the will
    if not ctx.willFlag:
      return
    if ctx.willRetain:
      retainMsg(ctx.willTopic, ctx.willMsg, ctx.willQoS, ctx.clientid)

    if not mqttbroker.subscribers.hasKey(ctx.willTopic):
      return
    for c in mqttbroker.subscribers[ctx.willTopic].ctxs:
      let msgId = c.nextMsgId()
      let qos = qosAlign(ctx.willQoS, c.subscribed[ctx.willTopic])
      c.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: ctx.willTopic, qos: qos,
                                retain: ctx.willRetain, message: ctx.willMsg, typ: Publish)
      await c.work()

when defined(broker):
  proc publishToSubscribers(seqctx: seq[MqttCtx], pkt: Pkt, matchedTopic, topic, message: string, qos: QoS, retain: bool, senderId: string) {.async.} =
    ## Publish async to clients
    for c in seqctx:
      if c.state != Connected:
        asyncCheck removeSubscriber(c, topic)
        continue
      let
        msgId = c.nextMsgId()
        qosSub = qosAlign(qos, c.subscribed[matchedTopic])

      if mqttbroker.passClientId:
        c.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: topic, qos: qosSub, retain: retain, message: senderId & ":" & message, typ: Publish)
      else:
        c.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: topic, qos: qosSub, retain: retain, message: message, typ: Publish)
      await c.work()

when defined(broker):
  proc denyConnect(ctx: MqttCtx, connFlag: ConnAckFlag) {.async.} =
    ## Denying connections and closing socket
    discard await sendConnAck(ctx, connFlag.uint16)
    ctx.state = Disabled # [MQTT-3.2.2-5] Close the connection immediately
    ctx.s.close()

#when defined(broker):
proc onConnect(ctx: MqttCtx, pkt: Pkt) {.async.} =
  when not defined(broker):
    ctx.wrn("Packet type only supported for broker: " & $pkt.typ)
  else:
    var
      offset: int
      nextLen: uint16
      keepAlive: uint16

    # Main data
    (ctx.proto, offset)     = pkt.getstring(0, true)
    (ctx.version, offset)   = pkt.getu8(offset)
    (ctx.connFlags, offset) = getbin(pkt, offset)
    (keepAlive, offset)     = pkt.getu16(offset)
    ctx.keepAlive           = initDuration(seconds = int64(keepAlive))
    ctx.keepAliveTimeout    = initDuration(milliseconds = int64(float(ctx.keepAlive.inMilliseconds) * 1.5))
    (nextLen, offset)       = pkt.getu16(offset)
    (ctx.clientId, offset)  = pkt.getstring(offset, parseInt($nextLen))
    if not mqttbroker.spacesInClientId:
      ctx.clientid = ctx.clientid.replace(" ", "")

    # Will Topic
    ctx.willFlag = ctx.connFlags[5] == '1'
    if ctx.willFlag:
      ctx.willFlag = true
      (nextLen, offset)         = pkt.getu16(offset)
      (ctx.willTopic, offset)   = pkt.getstring(offset,  parseInt($nextLen))
      (nextLen, offset)         = pkt.getu16(offset)
      (ctx.willMsg, offset)     = pkt.getstring(offset,  parseInt($nextLen))

      # Will Retain
      if ctx.connFlags[2] == '1':
        ctx.willRetain = true

      # Will qos=2
      if ctx.connFlags[3] == '1':
        ctx.willQoS = 2
      # Will qos=1
      elif ctx.connFlags[4] == '1':
        ctx.willQoS = 1

    # Username
    if ctx.connFlags[0] == '1':
      (nextLen, offset)       = pkt.getu16(offset)
      (ctx.username, offset)  = pkt.getstring(offset,  parseInt($nextLen))

    # Password
    if ctx.connFlags[1] == '1':
      (nextLen, offset)      = pkt.getu16(offset)
      (ctx.password, offset) = pkt.getstring(offset,  parseInt($nextLen))

    # Check password and username
    if mqttbroker.passwords.len() > 0:
      let pass = mqttbroker.passwords.getOrDefault(ctx.username)
      when defined(Windows):
        ## TODO: Windows is using MD5 for storing the password, which is not
        ##       safe in any way.
        if pass == "" or pass[0..31] != makePassword(ctx.password, pass[32..pass.len-1], ""):
          await denyConnect(ctx, ConnRefBadUserPwd)
          return
      else:
        if pass == "" or pass[0..59] != makePassword(ctx.password, pass[60..pass.len-1], pass[0..59]):
          await denyConnect(ctx, ConnRefBadUserPwd)
          return

    # 3.1.2.2 Protocol Level
    if ctx.proto != "MQTT":
      await denyConnect(ctx, ConnRefProtocol)
      return

    # 3.1.2.2 Protocol Level
    if ctx.version != mqttbroker.version:
      await denyConnect(ctx, ConnRefProtocol)
      return

    # Check if clientid already exist as an connection.
    # Close connection if clientID is identical to existing clientID [MQTT-3.1.4-2]
    if mqttbroker.connections.hasKey(ctx.clientid):
      if mqttbroker.clientKickOld:
        mqttbroker.connections[ctx.clientid].state = Disabled
        mqttbroker.connections[ctx.clientid].s.close()
      else:
        await denyConnect(ctx, ConnRefRejected)
        return

    # Require a clientID [MQTT-3.1.3-5], [MQTT-3.1.3-8], [MQTT-3.1.3-9]
    if ctx.clientid.strip() == "":
      # [MQTT-3.1.3-6] Assign random unique key
      if mqttbroker.emptyClientId:
        ctx.clientid = $r.next()
      else:
        await denyConnect(ctx, ConnRefRejected)
        return

    # Check the lenght of the clientID
    if ctx.clientid.len() > mqttbroker.clientIdMaxLen:
      await denyConnect(ctx, ConnRefRejected)
      return

    # Check for max simultaneous connections
    if mqttbroker.maxConnections > 0 and mqttbroker.connections.len() == mqttbroker.maxConnections:
      await denyConnect(ctx, ConnRefUnavailable)
      return

    mqttbroker.connections[ctx.clientid] = ctx
    ctx.state = Connected
    ctx.beenConnected = true
    asyncCheck keepAliveMonitor(ctx)

    if mqttbroker.verbosity >= 1:
      verbose("Connections >> " & ctx.clientId & " has connected")
    if mqttbroker.verbosity >= 3:
      verbose(ctx)

    ctx.workQueue[0.uint16] = Work(wk: PubWork, flags: ConnAcc.uint16, state: WorkNew, qos: 0, typ: ConnAck)
    await ctx.work()

proc onConnAck(ctx: MqttCtx, pkt: Pkt): Future[void] =
  ctx.state = Connected
  let (code, _) = pkt.getu8(1)
  if code == 0:
    ctx.beenConnected = true
    if ctx.verbosity >= 1:
      ctx.dbg("Connection established")
  else:
    ctx.wrn("Connect failed, code: " & $code)
  result = ctx.work()

proc onPublish(ctx: MqttCtx, pkt: Pkt) {.async.} =
  let
    qos = QoS((pkt.flags shr 1) and 0x03)
    retain = if (pkt.flags and 0x01) == 1: true else: false
                                # When subscribing and first message is a
                                # retained message, this will be `1`
                                # otherwise `0`.
                                #
                                # TODO for nmqtt_sub:
                                # Allow for passing the retained flag, so
                                # the subscriber can decide, if it wants
                                # only reteained message with --retained-only flag.
  var
    offset: int
    msgid: MsgId
    topic, message: string
  (topic, offset) = pkt.getstring(0, true)
  if qos == 1 or qos == 2:
    (msgid, offset) = pkt.getu16(offset)
  (message, offset) = pkt.getstring(offset, false)

  when defined(broker):
    for matchedTopic, topicCtx in mqttbroker.subscribers:
      if matchTopic(topicCtx.templ, topic):
        await publishToSubscribers(topicCtx.ctxs, pkt, matchedTopic, topic, message, qos, retain, ctx.clientid)

    if mqttbroker.verbosity >= 1:
      verbose("Client      >> " & ctx.clientId & " has published a message")

    if retain:
      if message == "":
        mqttbroker.retained.del(topic)
      else:
        # Add or overwrite existing retained messages on this topic.
        retainMsg(topic, message, qos, ctx.clientid)

      if mqttbroker.verbosity >= 1:
        verbose("Retained   ", mqttbroker.retained)

  when not defined(broker):
    var callbacks: seq[PubCallback]
    for t in ctx.pubCallbacks.values():
      if matchTopic(t.templ, topic):
        callbacks.add(t.cb)

    for cb in callbacks:
      cb.cb(topic, message)

  if qos == 1:
    ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 1, typ: PubAck)
    await ctx.work()
  elif qos == 2:
    ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 2, typ: PubRec)
    await ctx.work()

proc onPubAck(ctx: MqttCtx, pkt: Pkt) {.async.} =
  let (msgId, _) = pkt.getu16(0)
  assert msgId in ctx.workQueue
  assert ctx.workQueue[msgId].wk == PubWork
  assert ctx.workQueue[msgId].state == WorkSent
  assert ctx.workQueue[msgId].qos == 1
  ctx.workQueue.del(msgId)
  await ctx.work()

proc onPubRec(ctx: MqttCtx, pkt: Pkt) {.async.} =
  let (msgId, _) = pkt.getu16(0)
  assert msgId in ctx.workQueue
  assert ctx.workQueue[msgId].wk == PubWork
  assert ctx.workQueue[msgId].state == WorkSent
  assert ctx.workQueue[msgId].qos == 2
  ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 2, typ: PubRel)
  await ctx.work()

proc onPubRel(ctx: MqttCtx, pkt: Pkt) {.async.} =
  let (msgId, _) = pkt.getu16(0)
  assert msgId in ctx.workQueue
  assert ctx.workQueue[msgId].wk == PubWork
  assert ctx.workQueue[msgId].state == WorkSent
  assert ctx.workQueue[msgId].qos == 2
  ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 2, typ: PubComp)
  await ctx.work()

proc onPubComp(ctx: MqttCtx, pkt: Pkt) {.async.} =
  let (msgId, _) = pkt.getu16(0)
  assert msgId in ctx.workQueue
  assert ctx.workQueue[msgId].wk == PubWork
  assert ctx.workQueue[msgId].state == WorkSent
  assert ctx.workQueue[msgId].qos == 2
  ctx.workQueue.del(msgId)
  await ctx.work()

#when defined(broker):
proc onSubscribe(ctx: MqttCtx, pkt: Pkt) {.async.} =
  when not defined(broker):
    ctx.wrn("Packet type only supported for broker: " & $pkt.typ)
  else:
    var
      offset: int
      msgId: MsgId
      topic: string
      qos: QoS
      nextLen: uint16

    (msgId, offset) = pkt.getu16(0)
    ctx.msgIdSeq    = msgId

    var
      newTopics: seq[string]
      reply: string
    while offset < pkt.data.len:
      (nextLen, offset) = pkt.getu16(offset)
      (topic, offset)   = pkt.getstring(offset, parseInt($nextLen))
      (qos, offset)     = pkt.getQoS(offset)

      ctx.subscribed[topic] = qos
      await addSubscriber(ctx, topic)
      newTopics.add(topic)
      reply = reply & char(qos)

      if mqttbroker.verbosity >= 1:
        verbose("Client      >> " & ctx.clientId & " has subscribed to a topic")
        verbose("Subscribers", mqttbroker.subscribers)

    ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 0, message: reply, typ: SubAck)
    await ctx.work()

    for topic in newTopics:
      if topic notin mqttbroker.subscribers:
        continue
      let templ = mqttbroker.subscribers[topic].templ
      for retTop, ret in mqttbroker.retained:
        if matchTopic(templ, retTop):
          let
            msgId = ctx.nextMsgId()
            qosRet = qosAlign(qos, ret.qos)
          ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: retTop, qos: qosRet, message: ret.msg, typ: Publish)

          await ctx.work()

proc onSubAck(ctx: MqttCtx, pkt: Pkt) {.async.} =
  let (msgId, _) = pkt.getu16(0)
  assert msgId in ctx.workQueue
  assert ctx.workQueue[msgId].wk == SubWork
  assert ctx.workQueue[msgId].state == WorkSent
  ctx.workQueue.del(msgId)

#when defined(broker):
proc onUnsubscribe(ctx: MqttCtx, pkt: Pkt) {.async.} =
  when not defined(broker):
    ctx.wrn("Packet type only supported for broker: " & $pkt.typ)
  else:
    var
      offset: int
      msgId: MsgId
      topic: string
      nextLen: uint16

    (msgId, offset) = pkt.getu16(0)
    ctx.msgIdSeq    = msgId

    while offset < pkt.data.len:
      (nextLen, offset) = pkt.getu16(offset)
      (topic, offset)   = pkt.getstring(offset, parseInt($nextLen))

      await removeSubscriber(ctx, topic)
      ctx.subscribed.del(topic)

    if mqttbroker.verbosity >= 1:
      verbose("Client      >> " & ctx.clientId & " has unsubscribed from a topic")
      verbose("Subscribers", mqttbroker.subscribers)

    ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 0, typ: UnsubAck)
    await ctx.work()

proc onUnsubAck(ctx: MqttCtx, pkt: Pkt) {.async.} =
  let (msgId, _) = pkt.getu16(0)
  assert msgId in ctx.workQueue
  assert ctx.workQueue[msgId].wk == SubWork
  assert ctx.workQueue[msgId].state == WorkSent
  ctx.workQueue.del(msgId)

#when defined(broker):
proc onDisconnect(ctx: MqttCtx, pkt: Pkt) {.async.} =
  when not defined(broker):
    ctx.wrn("Packet type only supported for broker: " & $pkt.typ)
  else:
    ctx.state = Disconnected
    if mqttbroker.verbosity >= 1:
      verbose("Connections >> " & ctx.clientid & " has disconnected")

#when defined(broker):
proc onPingReq(ctx: MqttCtx, pkt: Pkt) {.async.} =
  when not defined(broker):
    ctx.wrn("Packet type only supported for broker: " & $pkt.typ)
  else:
    var msgId = ctx.nextMsgId() + 1000
    while ctx.workQueue.hasKey(msgId):
      msgId += 1000
    ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, state: WorkNew, qos: 0, typ: PingResp)
    await ctx.work()

proc onPingResp(ctx: MqttCtx, pkt: Pkt) {.async.} =
  ctx.awaitingPingResp = false

proc handle(ctx: MqttCtx, pkt: Pkt) {.async.} =
  when defined(broker):
    ctx.lastAction = getMonoTime()

  case pkt.typ
    of ConnAck: await ctx.onConnAck(pkt)
    of Publish: await ctx.onPublish(pkt)
    of PubAck: await ctx.onPubAck(pkt)
    of PubRec: await ctx.onPubRec(pkt)
    of PubRel: await ctx.onPubRel(pkt)
    of PubComp: await ctx.onPubComp(pkt)
    of SubAck: await ctx.onSubAck(pkt)
    of UnsubAck: await ctx.onUnsubAck(pkt)
    of PingResp: await ctx.onPingResp(pkt)
    #when defined(broker):
    of Connect: await ctx.onConnect(pkt)
    of Subscribe: await ctx.onSubscribe(pkt)
    of Unsubscribe: await ctx.onUnsubscribe(pkt)
    of Disconnect: await ctx.onDisconnect(pkt)
    of PingReq: await ctx.onPingReq(pkt)
    else: ctx.wrn("Unknown pkt type " & $pkt.typ)

#
# Async work functions
#

proc runRx(ctx: MqttCtx) {.async.} =
  try:
    while true:
      var pkt = await ctx.recv()
      if pkt.typ == Notype:
        break
      await ctx.handle(pkt)
  except OsError:
    if ctx.verbosity >= 2:
      ctx.wrn("Boom, socket is closed")

proc runPing(ctx: MqttCtx) {.async.} =
  if ctx.pingRunned:
    return
  ctx.pingRunned = true
  while true:
    if ctx.state notin {Connecting, Connected, Disconnecting}:
      break
    if ctx.state != Connected:
      await sleepAsync(500)
      continue

    let now = getMonoTime()
    if ctx.awaitingPingResp:
      if now >= ctx.timeoutPingRespTime:
        await ctx.close("ping response timeout")
        break
      await sleepAsync(500)

    else:
      let ms = (ctx.keepAlive - (now - ctx.lastSend)).inMilliseconds
      if ms < 0:
        discard await ctx.sendPingReq()
        ctx.timeoutPingRespTime = now + ctx.pingRespTimeout
        ctx.awaitingPingResp = true

      elif ms < 1000:
        await sleepAsync(int(ms))

      else:
        await sleepAsync(1000)

  ctx.pingRunned = false

proc connectBroker(ctx: MqttCtx) {.async.} =
  ## Connect to the broker.
  if ctx.keepAlive == DurationZero:
    ctx.keepAlive = initDuration(seconds = 60)

  if ctx.verbosity >= 1:
    ctx.dbg("Connecting to " & ctx.host & ":" & $ctx.port)

  ctx.state = Connecting

  try:
    ctx.s = await asyncnet.dial(ctx.host, ctx.port)
    if ctx.sslOn:
      when defined(ssl):
        ctx.ssl = newContext(protSSLv23, CVerifyNone, ctx.sslCert, ctx.sslKey)
        wrapConnectedSocket(ctx.ssl, ctx.s, handshakeAsClient)
      else:
        ctx.wrn("Requested SSL session but ssl is not enabled")
        await ctx.close("SSL not enabled")
        raise newException(IOError, "SSL session requested but compiler flag -d:ssl is missing")
  except CatchableError as e:
    ctx.state = Error
    raise

  let ok = await ctx.sendConnect()
  if ok:
    asyncCheck ctx.runRx()
    asyncCheck ctx.runPing()


proc runConnect(ctx: MqttCtx) {.async.} =
  ## Auto-connect and reconnect to broker.

  while true:
    if ctx.state == Disabled:
      break
    elif ctx.state in [Disconnected, Error]:
      try:
        await ctx.connectBroker()
      except OSError as e:
        if ctx.verbosity >= 1 or not ctx.beenConnected:
          ctx.dbg("Error connecting to " & ctx.host)
        if ctx.verbosity >= 2:
          echo e.msg

      # If the client has been disconnect, it is necessary to tell the broker,
      # that we still want to be Subscribed. PubCallbacks still holds the
      # callbacks, but we need to re-Subscribe to the broker.
      #
      # If we Publish during the Disconnected, the msg will not be send, cause
      # work() checks that `state=Connected`. Therefor our re-Subscribe
      # will be inserted first in the queue.
      if ctx.workQueue.len() == 0:
        for topic, t in ctx.pubCallbacks:
          let msgId = ctx.nextMsgId()
          ctx.workQueue[msgId] = Work(wk: SubWork, msgId: msgId, topic: topic, qos: t.cb.qos, typ: Subscribe)

    await sleepAsync(1000)

#
# Public API
#

proc newMqttCtx*(clientId: string): MqttCtx =
  ## Initiate a new MQTT client.
  MqttCtx(clientId: clientId, state: Disconnected, maxInflightMessages: 20,
          keepAlive: initDuration(seconds = 60), pingRespTimeout: initDuration(seconds = 10))

proc setPingInterval*(ctx: MqttCtx, txInterval: int = 60) =
  ## Set the clients ping interval in seconds. Default is 60 seconds.
  if txInterval > 0 and txInterval < 65535:
    ctx.keepAlive = initDuration(seconds = int64(txInterval))

proc setHost*(ctx: MqttCtx, host: string, port: int = 1883, sslOn = false) =
  ## Set the MQTT host.
  ctx.host = host
  ctx.port = Port(port)
  ctx.sslOn = sslOn

proc setSSLCertificates*(ctx: MqttCtx, sslCert: string, sslKey: string) =
  # Sets the SSL Certificate and Key to use when connecting to the remote broker
  # for mutal TLS authentication
  ctx.sslCert = sslCert
  ctx.sslKey = sslKey

proc setAuth*(ctx: MqttCtx, username: string, password: string) =
  ## Set the authentication for the host.
  ctx.username = username
  ctx.password = password

proc setWill*(ctx: MqttCtx, topic, msg: string, qos: QoS = 0, retain = false) =
  ## Set the clients will.
  ctx.willFlag   = topic != ""
  ctx.willTopic  = topic
  ctx.willMsg    = msg
  ctx.willQoS    = qos
  ctx.willRetain = retain

proc setMaxInflightMessages*(ctx: MqttCtx, maxInflightMessages: int) =
  ## Sets the maximum number of unacknowledged MQTT messages (QoS 1 and QoS 2). Default = 20.
  ctx.maxInflightMessages = maxInflightMessages

proc setVerbosity*(ctx: MqttCtx, verbosity: int) =
  ## Set the verbosity.
  ctx.verbosity = verbosity

proc connect*(ctx: MqttCtx) {.async.} =
  ## Connect to the broker.
  await ctx.connectBroker()

proc start*(ctx: MqttCtx) {.async.} =
  ## Auto-connect and reconnect to the broker. The client will try to
  ## reconnect when the state is `Disconnected` or `Error`. The `Error`-state
  ## happens, when the broker is down, but the client will try to reconnect
  ## until the broker is up again.
  ctx.state = Disconnected
  asyncCheck ctx.runConnect()

proc disconnect*(ctx: MqttCtx) {.async.} =
  ## Disconnect from the broker.
  await ctx.close("disconnect")
  ctx.state = Disabled

proc publish*(ctx: MqttCtx, topic: string, message: string, qos: QoS = 0, retain = false) {.async.} =
  ## Publish a message.
  ##
  ## **Required:**
  ##  - topic: string
  ##  - message: string
  ##
  ## **Optional:**
  ##  - qos: int     = 0, 1 or 2
  ##  - retain: bool = true or false
  ##
  ##
  ## **Publish message:**
  ## .. code-block::nim
  ##    ctx.publish(topic = "nmqtt", message = "Hey there", qos = 0, retain = true)
  ##
  ##
  ## **Remove retained message on topic:**
  ##
  ## Set the `message` to _null_.
  ## .. code-block::nim
  ##    ctx.publish(topic = "nmqtt", message = "", qos = 0, retain = true)
  ##
  let msgId = ctx.nextMsgId()
  ctx.workQueue[msgId] = Work(wk: PubWork, msgId: msgId, topic: topic, qos: qos, message: message, retain: retain, typ: Publish)
  await ctx.work()

proc subscribe*(ctx: MqttCtx, topic: string, qos: QoS | int, callback: PubCallback.cb): Future[void] =
  ## Subscribe to a topic.
  ##
  ## Shared subscription topic filter must follow the format: $share/{ShareName}/{filter}
  ## 
  ## Access the callback with:
  ## .. code-block::nim
  ##    proc callbackName(topic: string, message: string) =
  ##      echo "Topic: ", topic, ": ", message
  let msgId = ctx.nextMsgId()
  ctx.workQueue[msgId] = Work(wk: SubWork, msgId: msgId, topic: topic, qos: QoS(qos), typ: Subscribe)
  let topicCb = if topic.startswith("$share"):
    let topicItems = topic.split('/', 3)
    if topicItems.len < 3:
      raise newException(ValueError, "Invalid shared topic. Expected '$share/{ShareName}/{filter}', but got fewer than 3 segments.")
    elif topicItems[1].len == 0 or topicItems[2].len == 0:
      raise newException(ValueError, "Invalid shared topic. 'ShareName' and 'filter' segments cannot be empty.")
    elif {'+', '#'} in topicItems[1]: 
      raise newException(ValueError, "Invalid shared topic. The shared subscription group name must not contain MQTT wildcards ('+' or '#').") 
    topicItems[2]
  else:
    topic

  ctx.pubCallbacks[topic] = (templ: splitTopic(topicCb), cb: PubCallback(cb: callback, qos: QoS(qos)))
  result = ctx.work()

proc unsubscribe*(ctx: MqttCtx, topic: string): Future[void] =
  ## Unsubscribe to a topic.
  let msgId = ctx.nextMsgId()
  ctx.workQueue[msgId] = Work(wk: SubWork, msgId: msgId, topic: topic, typ: Unsubscribe)
  result = ctx.work()

proc isConnected*(ctx: MqttCtx): bool {.inline.} =
  ## Returns true, if the client is connected to the broker.
  ctx.state == Connected

proc hasError*(ctx: MqttCtx): bool {.inline.} =
  ## Returns true, if the context is in an error state.
  ctx.state == Error

proc msgQueue*(ctx: MqttCtx): int {.inline.} =
  ## Returns the number of unfinished packages, which still are in the work queue.
  ## This includes all publish and subscribe packages, which has not been fully
  ## send, acknowledged or completed.
  ##
  ## You can use this to ensure, that all your of messages are sent, before
  ## exiting your program.
  ctx.workQueue.len()