Merge pull request #1343 from emqtt/emq24

Version 2.3.0

Makefile

@@ -1,6 +1,6 @@
 PROJECT = emqttd
 PROJECT_DESCRIPTION = Erlang MQTT Broker
-PROJECT_VERSION = 2.3
+PROJECT_VERSION = 2.3.0
 
 DEPS = goldrush gproc lager esockd ekka mochiweb pbkdf2 lager_syslog bcrypt clique jsx
 
@@ -25,8 +25,9 @@ NO_AUTOPATCH = cuttlefish
 BUILD_DEPS = cuttlefish
 dep_cuttlefish = git https://github.com/emqtt/cuttlefish
 
-TEST_DEPS = emqttc
+TEST_DEPS = emqttc emq_dashboard
 dep_emqttc = git https://github.com/emqtt/emqttc
+dep_emq_dashboard = git https://github.com/emqtt/emq_dashboard
 
 TEST_ERLC_OPTS += +debug_info
 TEST_ERLC_OPTS += +'{parse_transform, lager_transform}'
@@ -36,7 +37,7 @@ EUNIT_OPTS = verbose
 
 CT_SUITES = emqttd emqttd_access emqttd_lib emqttd_inflight emqttd_mod \
             emqttd_net emqttd_mqueue emqttd_protocol emqttd_topic \
-            emqttd_trie emqttd_vm emqttd_config
+            emqttd_router emqttd_trie emqttd_vm emqttd_config
 
 CT_OPTS = -cover test/ct.cover.spec -erl_args -name emqttd_ct@127.0.0.1
 

etc/certs/cacert.pem

@@ -1,17 +1,17 @@
 -----BEGIN CERTIFICATE-----
-MIICxjCCAa6gAwIBAgIJAPhU8tv3KMe/MA0GCSqGSIb3DQEBCwUAMBMxETAPBgNV
-BAMMCE15VGVzdENBMB4XDTE2MTAzMTA3MTU0NVoXDTE3MTAzMTA3MTU0NVowEzER
+MIICxjCCAa6gAwIBAgIJAJk1DbZBu8FDMA0GCSqGSIb3DQEBCwUAMBMxETAPBgNV
+BAMMCE15VGVzdENBMB4XDTE3MTEwMjEzNDI0N1oXDTE5MTEwMjEzNDI0N1owEzER
 MA8GA1UEAwwITXlUZXN0Q0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIB
-AQCtPcDnmjiVl7ScDhYvGaW+PUgfp7P5cM39mnrW6fkxhA0tgunWpWlYVKbcuh5y
-4bTNYrOQpcFO3Zg62tva4XEL8O1huqTlGsAeysZ3vWE4/8NGN/3wZy0TKDvwiwOB
-tbS3C5wcRQZohExL6yEL4XzDGk44x2mIs8/NzeG7Zycqybh9tsCJiHbLiTxnLa24
-v5USOtlvWye0hA0yUUqc2k7tKVmIMT4A4ulMb2sDVRrSLjyFDTI0c8grlPLfKbG8
-gpYLsHn9aAjqviyvmJdRLxwauqn+ghNWn1TyZwgAUxpoTtWeC0ilzEt18RP8vZjm
-eCbEP4qQDDvSCdLrie5CezyxAgMBAAGjHTAbMAwGA1UdEwQFMAMBAf8wCwYDVR0P
-BAQDAgEGMA0GCSqGSIb3DQEBCwUAA4IBAQBJ/I/QJjU+mgkIaaHImFcIYFrfBirC
-vDiWo2W+zRh7CbcSf+jsksI99d230ixSDY36CPLKZeZhELST7xWKEELKbPdNbtOO
-EM10+XteLSXKVNGXfrEbW973eum3FGLobMA9OcH6+qDaf08pibe7kuv10aAgSs/I
-0Qg5H/UTAKQJKO9hhOgERM/FettuF+WGJaaZZZb9Y2YYBNRf/GtM8KHCjpCX9+XD
-kdeQGO8Hn10H9tOmggyfdIpsunBcs2/6/exCp8RPBWurN2GSW2RcnS5xVL0r+SVW
-VOhSDy1JwnNPczpqkqE74qAbAah0dTJFcFWzeGLVk7Kp+2pissAiU3gg
+AQDshDho6ef1JClDJ24peSsXdFnFO3xIB7+BSp1YPcOvmRECKUG0mLORw3hNm15m
+8eGOn1iLGE/xKlaZ74/xjyq8f7qIGZCmvZj59m+eiJCAmy8SiUJZtSVoOlOzepJd
+PoDgcBvDKA4ogZ3iJHMUNI3EdlD6nrKEJF2qe2JUrL0gv65uo2/N7XVNvE87Dk3J
+83KyCAmeu+x+moS1ILnjs2DuPEGSxZqzf7IQMbXuNWJYAOZg9t4Fg0YjTiAaWw3G
+JKAoMY4tI3JCqlvwGR4lH7kfk3WsD4ofGlFhxU4nEG0xgnJl8BcoJWD1A2RjGe1f
+qCijqPSe93l2wt8OpbyHzwc7AgMBAAGjHTAbMAwGA1UdEwQFMAMBAf8wCwYDVR0P
+BAQDAgEGMA0GCSqGSIb3DQEBCwUAA4IBAQAi+t5jBrMxFzoF76kyRd3riNDlWp0w
+NCewkohBkwBHsQfHzSnc6c504jdyzkEiD42UcI8asPsJcsYrQ+Uo6OBn049u49Wn
+zcSERVSVec1/TAPS/egFTU9QMWtPSAm8AEaQ6YYAuiwOLCcC+Cm/a3e3dWSRWt8o
+LqKX6CWTlmKWe182MhFPpZYxZQLGapti4R4mb5QusUbc6tXbkcX82GjDPTOuAw7b
+mWpzVd5xnlp7Vz+50u+YaAYUmCobg0hR/AuTrA4GDMlgzTnuZQhF6o8iVkypXOtS
+Ufz6X3tVVErVVc7UUfzSnupHj1M2h4rzlQ3oqHoAEnXcJmV4f/Pf/6FW
 -----END CERTIFICATE-----

etc/certs/cert.pem

@@ -1,18 +1,18 @@
 -----BEGIN CERTIFICATE-----
-MIIC9jCCAd6gAwIBAgIBATANBgkqhkiG9w0BAQsFADATMREwDwYDVQQDDAhNeVRl
-c3RDQTAeFw0xNjEwMzEwNzE1NDVaFw0xNzEwMzEwNzE1NDVaMDkxJjAkBgNVBAMT
-HWRlbmdoYWlndWlkZU1hY0Jvb2stQWlyLmxvY2FsMQ8wDQYDVQQKEwZzZXJ2ZXIw
-ggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQC4Ena4vgWrzwUB0hGW1v0v
-K986FhU5ZdYz5H5MGonfWwv89nR2DlftSDXEvKFyc2MT81GGm16VJv3mVpQJLuKA
-xLBLY7a1zSrJdugXWy+mgJJTPW6KjTY4jPtfCl6x/yVr8YclVa8XO0JFzOme2LMV
-Ylc/ixVEa66UpxRNrg5yWHS26KcB1lE3GLERoRBKF7nsyGqGY4X9TypBwglCVoqK
-3dKVGwCvFur+oPnt/C5pwR6UmUV/Ppf1EaRD7Po+xcyJSeCvszG3FH4iHsDHnjLe
-DR6lxouvMCb+aKJi9d0xowOjhbKoFMF179t4SVnptQeq+U6ui3cPKUjia7Zh1tZT
-AgMBAAGjLzAtMAkGA1UdEwQCMAAwCwYDVR0PBAQDAgUgMBMGA1UdJQQMMAoGCCsG
-AQUFBwMBMA0GCSqGSIb3DQEBCwUAA4IBAQB2jlDPiZfP/whsvvFn43g37QMwX5ST
-Z5OpmEFnFjAH3ec0PPqPrKYEu00q5wEC+8L6uVH8FHOFf11JLH4wl11/C/mvE92D
-qZtGG8KCnG2+rk5OJPGX+28Z+OnCZlXOjQ8qd2x5KtIW50JuXJ3cbDRHtF/TVanm
-Exu+TCBeToNwbcU2sfQnbljkUTj4idUFz0pq3uvw3dA4R1J2foungPAYXSWcVhtb
-RYtG8epIvkAyyUE5nY3kC05AUml6gSZkrJiYM5I1IJTX1lQ7Pv2yxRBZUtTx33rP
-ccnsW6tbHTDBG8UDHx4LKHErdWFgCJWI81EUEcTip9g2zCOGTWKnpz+z
+MIIC6jCCAdKgAwIBAgIBATANBgkqhkiG9w0BAQsFADATMREwDwYDVQQDDAhNeVRl
+c3RDQTAeFw0xNzExMDIxMzQyNDhaFw0xOTExMDIxMzQyNDhaMC0xGjAYBgNVBAMT
+EU1hY0Jvb2stQWlyLmxvY2FsMQ8wDQYDVQQKEwZzZXJ2ZXIwggEiMA0GCSqGSIb3
+DQEBAQUAA4IBDwAwggEKAoIBAQDUO/kL3ar3WsopPF12qAf+cwDHklGJIxJsjdoZ
+XgI1lPEe1W1QXwb/G/tyf6Fj2J8CD5bfsRjDxAemFIBVrFwlunCk+Gs6xR7vzz4O
+Fonoj4pmleruLQrNY/bHa2WN97OdISyXzhOgDwSaqobnF0n/f0Mx+9sdHO3p8LNB
+3JXUyBpwDNr/TTfAb4pbQEu3LF4p7uyd1eLhKzUxSiWzKtjB1EYObA87fZu0tBJZ
+iGujuFiI7tf4qWKeuAoRa/cXkgVZhk0utYauDoa7qBZ5O6ZdEko9ov0+i5+1JGU/
+w5wrSPNAnM2lYVUn0kJmcV2gwa4RZFjdqp+/Fx+HnKbnhZEnAgMBAAGjLzAtMAkG
+A1UdEwQCMAAwCwYDVR0PBAQDAgUgMBMGA1UdJQQMMAoGCCsGAQUFBwMBMA0GCSqG
+SIb3DQEBCwUAA4IBAQByWhNxX/L5QYBiMY4JM1RRciV4uI3F2vsc0yMFDSrZza+5
+tNJQS86hjQsCRZh9VshezvT7k1yVsAC4pnu2pzob8H3KG4vYBafMdl2Ghgv3RMix
+J3NrBhcoYYhXEoZHost+htxEi7P3QBo/qDkk48/d30+aDPbms6kQd8Fj8+C5tD3b
+aznO5Qlni72uTaM7fNA8exoc/YZc83lsqv7v+UzNQR595jnYSIAZcgil1qqygOan
+Zx/RsMGUz6EYI9lPpoyyVtw13SoQshfgwvUlvBMiekSuI/pp6N7QPK6C8DLO0tVv
+gXJjDgioqHc3hcgG4cskLbfVnohiwdhQTFayrLEk
 -----END CERTIFICATE-----

etc/certs/client-cert.pem

@@ -1,18 +1,18 @@
 -----BEGIN CERTIFICATE-----
-MIIC9jCCAd6gAwIBAgIBAjANBgkqhkiG9w0BAQsFADATMREwDwYDVQQDDAhNeVRl
-c3RDQTAeFw0xNjEwMzEwNzE1NDZaFw0xNzEwMzEwNzE1NDZaMDkxJjAkBgNVBAMT
-HWRlbmdoYWlndWlkZU1hY0Jvb2stQWlyLmxvY2FsMQ8wDQYDVQQKEwZjbGllbnQw
-ggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCmPMkieMtJO4PGIQG30uxI
-SEoRJoF2w0ufFhZGYCEaqFlHaSoc6nTiCUmnxadDpjkNBs4R6RDfM9zPJ0QdgSFO
-OJsWgQEHym/EQTcEx11+/2NDZWMJyZdpWZlU57SwHfWDwYa2XFX1bV+pAvhB8cli
-wCkygTwp1cZcwQpb8TfZySy8r5mwrWq2nhCQPtYqMxjNjpR/UeeZzt+Uh3CEXQ8h
-omjGinDXnnGwrYwBEP9G6fzTvyCWTyrsWC1Q37oAMzbkwFRoIBSAQWXBv9hgI08s
-IBYvXnRGKWOJZGxAP4a4TvpFS+nqi+fFVn4ktUfcH3PoSMh7PKavrFT2hQaryLt1
-AgMBAAGjLzAtMAkGA1UdEwQCMAAwCwYDVR0PBAQDAgeAMBMGA1UdJQQMMAoGCCsG
-AQUFBwMCMA0GCSqGSIb3DQEBCwUAA4IBAQAeimI8AQBFWiE9/Nf/0radux355mod
-5vPLbKn6I6nzb/sS/Ug8SMoFnkhncwj+XOgTSliUyWcwOB11UDVJbUIkB/x+Qo3w
-hvrATTdby2WdFNQvH4X7PmP8asDDN7ZxoLyRmuhjL4avJ3giwRcuQK4cB35b+Lb2
-p1e7hW81RaV7OEc0o4/vJgPvv9N7wvUuipwJns6PrN7VDn99lT8zWrt2pQ06e2mk
-jDuXulVpiUtLHJhTnABkCaKiHWCYAFfMjFeRb3gUXKqShzOyDSGWY91YMID/HE4r
-sVLm2mD1zurue8EmYtQQ6uiJIW9SzvshMHG6EA5QWA1ytoalfePbvf+c
+MIIC6jCCAdKgAwIBAgIBAjANBgkqhkiG9w0BAQsFADATMREwDwYDVQQDDAhNeVRl
+c3RDQTAeFw0xNzExMDIxMzQyNDhaFw0xOTExMDIxMzQyNDhaMC0xGjAYBgNVBAMT
+EU1hY0Jvb2stQWlyLmxvY2FsMQ8wDQYDVQQKEwZjbGllbnQwggEiMA0GCSqGSIb3
+DQEBAQUAA4IBDwAwggEKAoIBAQC8GptpL25hv1Qa3jCn4VLvDRH/SrHg9wXvqRkz
+HuiKMxYT30m4+kcaXv350CJrkV+8lR24wdN7DBVewpCUnyUBbzkLccy1LUzunZ3z
+nm37j6cautD3rlC9gsC9d0uJ745FLx5t/6f1jMk9rWxn+4iSGAnkWC3mVaQxP1zQ
+q8GI97uob9HNb0OH6ygHJAcKOWB+85a29LIMa1uo/lT3hMr8sBg2vX+1F/gTusmW
+xVoQc9XJxBCs995qsH0UkZIuOY0XZp9/qFfcZv2QmslG8DojIIHKcujzu8bItE2M
+OyL5NlWLvN6qg59hHzF4+D+T+8GkhhKWSC+xdY14eQ5fB4S5AgMBAAGjLzAtMAkG
+A1UdEwQCMAAwCwYDVR0PBAQDAgeAMBMGA1UdJQQMMAoGCCsGAQUFBwMCMA0GCSqG
+SIb3DQEBCwUAA4IBAQBLV4ZfhiKiFVnL/xO0MRGSKr3xd0LK64SW8Iw5DYkc0jNX
+sDrRbj2I/KJ/Rc4AeKT751L+C+KBzYpFgiLrxDmt/5pmgiFH51hPQtL7kRC0z2NY
+EY/P+u4IFVSo+b1hHYU7y+OMj6/Vvd4x0ETS4rHWI4mPDfGfvClEVLOktgRKrMU5
+9aTltF4U0FBUlYZTQBNBUFwBzj1+0lxK4EdhRmmWJ+uW9rgkQxpnUdbCPGvUKFRp
+3AbdHBAU9H2zVd2VZoJu6r7LMp6agxu0rYLgmamRAt+8rnDXvy7H1ZNdjT6fTbUO
+omVBMyJAc1+10gjpHw/EUD58t5/I5tZrnrANPgIs
 -----END CERTIFICATE-----

etc/certs/client-key.pem

@@ -1,27 +1,27 @@
 -----BEGIN RSA PRIVATE KEY-----
-MIIEpQIBAAKCAQEApjzJInjLSTuDxiEBt9LsSEhKESaBdsNLnxYWRmAhGqhZR2kq
-HOp04glJp8WnQ6Y5DQbOEekQ3zPczydEHYEhTjibFoEBB8pvxEE3BMddfv9jQ2Vj
-CcmXaVmZVOe0sB31g8GGtlxV9W1fqQL4QfHJYsApMoE8KdXGXMEKW/E32cksvK+Z
-sK1qtp4QkD7WKjMYzY6Uf1Hnmc7flIdwhF0PIaJoxopw155xsK2MARD/Run8078g
-lk8q7FgtUN+6ADM25MBUaCAUgEFlwb/YYCNPLCAWL150RiljiWRsQD+GuE76RUvp
-6ovnxVZ+JLVH3B9z6EjIezymr6xU9oUGq8i7dQIDAQABAoIBAFkHEMjPXD96ChZf
-suXZpgUIAfKxZoBOEv+9+mvyK4h1RGsEHTOjNLmhM7sQFYYbTU52qIHbCdgflE+0
-vbv3XfjgQ96HdB/SAI1gR7DdfGr5JxX/BE1HkzkubPmVpaT0RnoreJPNW5O24ZZI
-KuBWNv4V33pWz/uvqy4djAi1ZK3TPDhn9cVCMwV/ISCPlofrNDB/4ZNOMeaQgiR+
-sGqv+Q0ok2ao7Y04QHPh5i+5o+5oBoiJAO/49q9uPdpO181/8H71jll0QL+h5Off
-nyWkAAOcgEeX9T4ZnfTUivGdSwB/Y+LS97Ozdr6kp5Fdk8WdDn0DL4fHRrnJ4IJD
-EIAn/sECgYEA2oOCRBMccr49wbu+cKlkICt/4ARzJWKysdLlK0tYQknkDK1bzoHO
-9JerRJL4E9bKp8zNlobfP1hWV0TFpwYsK3RvZoLvCwaSHeqUCZ4wQvKrWP1FieJ2
-5kjO5iMvXiy/kNHdTEXsj0x6RKuUSVgzNIuILvCCQ9Z7JVa/3NWS1SkCgYEAwsF0
-TWxCjryQv8y4mFSUlyF+y+ntnWAvpe/1Wv3+dNdhsccUfcq3zPMuLEj5DEoIvlTy
-jLkFLVJ468Ou7S1oSVetVT3wWoLP2eFDEU/sYjjPdf4IMSO1jWIPLC3WV7zsFb62
-jwG2en1qfz8AxrVl+zj4lWCbgA9Soi41NMiCUW0CgYEAokQEST8T4hVp0OL1Qb5Y
-bxc+Z4GGbF3Fqw2cRrE1wkwSwGNACLMWl0XF1i95b2oSpdcNWFmhkO2teDLGwAhy
-ZnaZfzt9/ecMPJEFC7tfxWdlXLj/mawFdW7dzcKVG08JlqZxuoE2cRduuG3duTV5
-GO0A3TKW2X99hTXNVlV3KzkCgYEAsaE8cHkzY3h9FVKlctqCBC3atiWQQZ+/Fbv8
-rpdHBE6Fnl4TRIAmj9mk3WNZM2o6+04DQ3JlVGcKPw7ldxGZMnuzbjHmDMeOyAx6
-3UlmMlfacKXX1unY5zDu4b6U5sU7FsIxQ9GuG55UCebu0E4Wy8G0iJnqeix/k8hN
-Yu0WXykCgYEAo0kIm7sh9j0+r419Lo2kT4zlzFlNdJEa4+lFVISRqouDuhUO8VFE
-/ZpGRcqIM7dH6iBM2Htasf7l/hyWKzDEvWCEpa4icicFYAJ92AgK7UBWbNbhueof
-PyVx5G2o7amvyZNtJYUo4TpJ9eH5YbsBRBqWCJcBUAfrItrprxB1LMs=
+MIIEowIBAAKCAQEAvBqbaS9uYb9UGt4wp+FS7w0R/0qx4PcF76kZMx7oijMWE99J
+uPpHGl79+dAia5FfvJUduMHTewwVXsKQlJ8lAW85C3HMtS1M7p2d855t+4+nGrrQ
+965QvYLAvXdLie+ORS8ebf+n9YzJPa1sZ/uIkhgJ5Fgt5lWkMT9c0KvBiPe7qG/R
+zW9Dh+soByQHCjlgfvOWtvSyDGtbqP5U94TK/LAYNr1/tRf4E7rJlsVaEHPVycQQ
+rPfearB9FJGSLjmNF2aff6hX3Gb9kJrJRvA6IyCBynLo87vGyLRNjDsi+TZVi7ze
+qoOfYR8xePg/k/vBpIYSlkgvsXWNeHkOXweEuQIDAQABAoIBAHnFV7peRDzvGUlT
+cXgcvA2ZDn+QIVsbTzJ466FWbv+YVsCCmj0veHwv5oakIMQ2Fh4FAnqqr3dGuUbg
++avc4p3tHKa2Aul+7ADE9I3TkCt8MZdyPPk6VXZ5gMCmy7X96MIM4Mwg5uBlRZmx
+/S3Lffvlp/G0y/ICmwpulG1Z4y4A5Vc0Qf7fBO03Ekl31oReARnB6ex7RnDHH1mW
+RyLWNqyu9BhUbFpIyFPWDSkBcajNIbQ6qVJfGLm5Y2xVhwdqbyvY8M06uuMKz/IR
+SYfdIpiC4PpQZQzzXMn/6LTKWcCe0T+dBcWTZHC3C2abrC7+5fwFobs2xoUaCwz0
+1CclogECgYEA6Jdv+2VSYIBLbS0VIe07JiZaQNd1QNg63MK/y7oqAKEzYvpWzJel
+owPdBU3GxZH6vUUF7sCABcjumEDazoqTtzHQBo0xYpJrjmAL0ANNGVvF09pJK2eq
+yotxJJAS5/lQNSgWOxGVc6qu6ZpgeIXVLIx816yq04h10yVgZ2Lm3+ECgYEAzwj5
+/UVpN/ak6PwZ+Tq/8qOYjY2ABylRmP+T0Rkqmwh2B5Sp9oXjkDQwWseY0Wybhd8F
+kO6BUCMUApnB3uU0baawVbDUSrt43SkkKV9m3pA35wA3pYw1a56QIEFr56npFYBS
+sn9yl/ZtNvnuwmrHWOq8HdwPJsWREyO61yknn9kCgYB1PdixpSo4AJOErePoHRfi
+rBR0eObez+Aj5Xsea3G+rYMkkkHskUhp+omPodvfPS1h+If8CEbAI7+5OX/R+uJo
+xpAwrT1Gjb3vn5R0vyU+8havKmoVmgTqYg2fO4x8KBz5HoLONZfbHR9cG3gjaHrD
+IPHRGXVmeXPDAiUtGBp+oQKBgQDDRIAkNPdMZUCczknhG1w3Cb20pKUAHCRt3YAZ
+U1cv6gcIl1rGvPko5VBGDsM/ouP8m6CwVYN5hdw1p7eG9z8/vFvMNn/EDJWuYkNN
+EkH/4J4ZLcdOSLOJ0X+2LH4Nfd/s+58D49i9IxtXItviWruyTZMnxooz01tFZgmv
+LY3F4QKBgDirafhlJqFK6sa8WesHpD5+lm3Opzi4Ua8fAGHy2oHN3WCEL74q691C
+fA0P2UrzYiF7dXf4fgK9eMMQsdWS4nKyCSqM6xE4EAhAHUTYzY3ApNjI3XFDIrKC
+oQefIOLum2UyWFuEoUtrEfc5fxktiQohCwuAvwC59EwhmsNlECA8
 -----END RSA PRIVATE KEY-----

etc/certs/key.pem

@@ -1,27 +1,27 @@
 -----BEGIN RSA PRIVATE KEY-----
-MIIEpAIBAAKCAQEAuBJ2uL4Fq88FAdIRltb9LyvfOhYVOWXWM+R+TBqJ31sL/PZ0
-dg5X7Ug1xLyhcnNjE/NRhptelSb95laUCS7igMSwS2O2tc0qyXboF1svpoCSUz1u
-io02OIz7Xwpesf8la/GHJVWvFztCRczpntizFWJXP4sVRGuulKcUTa4Oclh0tuin
-AdZRNxixEaEQShe57MhqhmOF/U8qQcIJQlaKit3SlRsArxbq/qD57fwuacEelJlF
-fz6X9RGkQ+z6PsXMiUngr7MxtxR+Ih7Ax54y3g0epcaLrzAm/miiYvXdMaMDo4Wy
-qBTBde/beElZ6bUHqvlOrot3DylI4mu2YdbWUwIDAQABAoIBADXYWNhT5c7LYTiW
-HcUVIL0CxWr1eMHwk0dcyME0Zi5rMMePxKOgMIJdxDTHxSZ4sHvuimOo4XMaE92k
-Z+uDxohKgROcmJ735FNIsD3c08SOCb/F0adABaNnQkUcAHVrIKRB4/m85doS4KEQ
-fyqTU1enC8Svx8nbAhfEBEFw8BLsZD9UnQAEAU5W9S5aKPHNrYRDz5UE0ZP28ixC
-4PtCew96uCqA0u+xZnWCGawF27FD9P88pcYSJqebF1iFYkXrAwdhAbqewHOqQJXf
-KJpbpjflBvZr/oTVZ3GAnnHnZDiusFmCKIHB9dKimHMdTFVIU2ikOeJZLtgXsBjb
-Wn3Fa8kCgYEA2fK0t9NPmELw43D7VoCNeUmu6KmLLd7CeRiQ/OkPLKTqrudnUZGi
-uMinPFijGTLX3SmByAVOkzMKBQOYF+eB1X24kbRLmL4JKzr04hSqOKqG5gJctC+x
-V5qQX7ZxrNxFRiSodILbnQN/z1gwZMfrAU0t0EKIKjZR3lpj8CELv1cCgYEA2DWn
-9V6PCZPcHzoFabhb8DJFglUTHk0zINVe97qldvMvn0MgsjgyS2j954nX8ef7uE1O
-Cf+9nN709Fu8kEC7/KzWXxP3/O58TfJ6NivCQSr5i0OJLumQMVNrS+u/VG1PaVbS
-2oCwP3QFayOxZSj9wq2MARd1JkqzHmi8skZLz2UCgYEAgtnv3En3CLBwFe14SPgH
-eGFfrPpVwGV0luXD7sQyQxiEehwecN+iNZTqqxWAXpmi9np8G83r3f6PrnD4+Kka
-z0Wa8Yewt3So5paP/chwZnMjaKbUZ64WqET5Fy3fU+wvfyx1IvaJydwW+TK2Y1uP
-4Yknz1iSjd1tC7VzOPFuLyMCgYBrTFWKQ98glayMIrNFACVAUvKD98yBITbaeImk
-z5AGNDHSC/JR/+mV2wkGuzXb65DUqiisdaqYC13tVwmBXV7tyqiojrRnZcNyu39D
-GvxQcw9cuat/CJJyqD97cgeF0qmyUVBa97qAAwgdX51N4sXss0vjzsxosHGsCbZ7
-kr9UsQKBgQCMTtdCeA+uK/OeJtzf4CYZKR9xllQ+P6gCtbQ7WHuLBX/x+ZhvTC0p
-qVLVWwFsJ6ivc1f74sy8hZPiePk9fqAqA1JIjDHrof0M3TxRVFvB7dej5XIYVirn
-521DyZGfE+N7HA7qW5cGKZT0+UYLVp4gnv88nNKDuS18lafy8JRrfQ==
+MIIEogIBAAKCAQEA1Dv5C92q91rKKTxddqgH/nMAx5JRiSMSbI3aGV4CNZTxHtVt
+UF8G/xv7cn+hY9ifAg+W37EYw8QHphSAVaxcJbpwpPhrOsUe788+DhaJ6I+KZpXq
+7i0KzWP2x2tljfeznSEsl84ToA8EmqqG5xdJ/39DMfvbHRzt6fCzQdyV1MgacAza
+/003wG+KW0BLtyxeKe7sndXi4Ss1MUolsyrYwdRGDmwPO32btLQSWYhro7hYiO7X
++KlinrgKEWv3F5IFWYZNLrWGrg6Gu6gWeTumXRJKPaL9PouftSRlP8OcK0jzQJzN
+pWFVJ9JCZnFdoMGuEWRY3aqfvxcfh5ym54WRJwIDAQABAoIBABNq2UJIqZev6scT
+CsoMXY7eHrgjnuoZF1pvMAEaJMGaOuVDSZkM2KsGeF7lZnKoIwQhQQB+R3HBwaFk
+RsmP125sPFobkFP0LPxrzZWkYkGwwEzacoAQBuj7uFxOayAuBXTe0CGjbRA7z4QH
+DgiejNqfXhp4nHdxaiL5Lq1b7SlmarGXup3kcVTWxIiah4MK0o4YGiyQC8Mr+a7w
+UGYqdKQQMLOtly/HTEcyd/DAruboV+5L+pYx/pcFFXJupK6yaxELLHKeHAKA9MmA
+cnMNVpCQ8VdOyR9qrfwtABqd8egKea2Z3P+dK6PlxUAQe2kYlXxS0N+i/eU6PKYM
+B76UhQECgYEA9GvBNG6ffQqkX6bNLQUsU+nvKAQeFq02ua9LFKYw2sVO6RfUjrNz
+u2cwAUXSp+tnPMesKEVOOUfRMN/QiI/JNw62uSWSKJ/64103vX+F5AjQmE2f7Zgt
+o3X23cV544HM8E5xCvIe7DFLK6cUdRQngu/uWi63cB4hMVpB9MfZJscCgYEA3kne
+2sE4b67JkjmHGKahBJM5/iAHBqSubQmufIlaiLkyrDYGN2D+mi0fAF+uQ9KmNOrv
+TsZ1bZu9f+VvaH7xNJzcUriXYs+HoN9/CWnAR0ktSm8RN7BznVd41NuLnsoWUt41
+jglpNYMwy7JPRLQNgYHErG7puksNawFvSKQEYqECgYA4N/iueKtSdXotTg5vRntV
+qb8KczgAe0LVHs6kJz2hdDScRJDtabU665cNE+RKH0kVn8+nS5mcbzpchX5PitL7
+SPUaTNv7YCCy3yQNACHpu2VPQruASLpmmKF5jQxmGdrrgv9ZRyt5pDToC3wXGdWk
+tk8aixhCP4ve8CWvibAWzQKBgExxxwwf6tKtn3CEDCu0EifKoeT9Cq2EMOAatkDp
+05K1bfG/Wn/tAWHwJnswbHOym6oTKV1D7tpU9uRm+NtM3JKlZzejd5xpllECy2Nn
+VNKvHb49WAR40CnKDSnWnrtq8CZreKtyHRZkGYHTvmL4MLTa9dH/Cq4gZWrpQWYP
+0dpBAoGAD4inpSm7SMN3/rgYXEU1CMRKXREbEWhondiTXZ8x8ugnnYtfhcBvCMif
+JQ8tso63hCHvKPDViTbLDyV7OuGBEPTQAyacX0FJmr7g5ERlvfmL4yjmvW7Bcclh
+yrgbJXl2pdzMt9GpogIYFW0YyOr6VPIrGf62kRNrv2E8wyXEFAI=
 -----END RSA PRIVATE KEY-----

etc/emq.conf

@@ -1,5 +1,5 @@
 ##===================================================================
-## EMQ Configuration R2.3
+## EMQ Configuration R2.3.0
 ##===================================================================
 
 ##--------------------------------------------------------------------
@@ -288,8 +288,6 @@ mqtt.broker.sys_interval = 60
 ## PubSub Pool Size. Default should be scheduler numbers.
 mqtt.pubsub.pool_size = 8
 
-mqtt.pubsub.by_clientid = true
-
 ## Subscribe Asynchronously
 mqtt.pubsub.async = true
 

include/emqttd_trie.hrl

@@ -17,10 +17,10 @@
 -type(trie_node_id() :: binary() | atom()).
 
 -record(trie_node,
-        { node_id         :: trie_node_id(),
-          edge_count = 0  :: non_neg_integer(),
-          topic           :: binary() | undefined,
-          flags           :: [retained | static]
+        { node_id        :: trie_node_id(),
+          edge_count = 0 :: non_neg_integer(),
+          topic          :: binary() | undefined,
+          flags          :: [retained | static]
         }).
 
 -record(trie_edge,

priv/emq.schema

@@ -715,11 +715,6 @@ end}.
   {datatype, integer}
 ]}.
 
-{mapping, "mqtt.pubsub.by_clientid", "emqttd.pubsub", [
-  {default, true},
-  {datatype, {enum, [true, false]}}
-]}.
-
 {mapping, "mqtt.pubsub.async", "emqttd.pubsub", [
   {default, true},
   {datatype, {enum, [true, false]}}
@@ -727,7 +722,6 @@ end}.
 
 {translation, "emqttd.pubsub", fun(Conf) ->
   [{pool_size, cuttlefish:conf_get("mqtt.pubsub.pool_size", Conf)},
-   {by_clientid, cuttlefish:conf_get("mqtt.pubsub.by_clientid", Conf)},
    {async, cuttlefish:conf_get("mqtt.pubsub.async", Conf)}]
 end}.
 

src/emqttd.app.src

@@ -1,6 +1,6 @@
 {application,emqttd,
              [{description,"Erlang MQTT Broker"},
-              {vsn,"2.3"},
+              {vsn,"2.3.0"},
               {modules,[]},
               {registered,[emqttd_sup]},
               {applications,[kernel,stdlib,gproc,lager,esockd,mochiweb,

src/emqttd.erl

@@ -31,8 +31,7 @@
          unsubscribe/1, unsubscribe/2]).
 
 %% PubSub Management API
--export([setqos/3, topics/0, subscriptions/1, subscribers/1,
-         is_subscribed/2, subscriber_down/1]).
+-export([setqos/3, topics/0, subscriptions/1, subscribers/1, subscribed/2]).
 
 %% Hooks API
 -export([hook/4, hook/3, unhook/2, run_hooks/2, run_hooks/3]).
@@ -43,14 +42,13 @@
 %% Shutdown and reboot
 -export([shutdown/0, shutdown/1, reboot/0]).
 
--type(subscriber() :: pid() | binary()).
+-type(subid() :: binary()).
 
--type(suboption() :: local | {qos, non_neg_integer()} | {share, {'$queue' | binary()}}).
+-type(subscriber() :: pid() | subid() | {subid(), pid()}).
 
--type(pubsub_error() :: {error, {already_subscribed, binary()}
-                              | {subscription_not_found, binary()}}).
+-type(suboption() :: local | {qos, non_neg_integer()} | {share, {'$queue' | binary()}}).
 
--export_type([subscriber/0, suboption/0, pubsub_error/0]).
+-export_type([subscriber/0, suboption/0]).
 
 -define(APP, ?MODULE).
 
@@ -59,19 +57,19 @@
 %%--------------------------------------------------------------------
 
 %% @doc Start emqttd application.
--spec(start() -> ok | {error, any()}).
+-spec(start() -> ok | {error, term()}).
 start() -> application:start(?APP).
 
 %% @doc Stop emqttd application.
--spec(stop() -> ok | {error, any()}).
+-spec(stop() -> ok | {error, term()}).
 stop() -> application:stop(?APP).
 
 %% @doc Environment
--spec(env(Key:: atom()) -> {ok, any()} | undefined).
+-spec(env(Key :: atom()) -> {ok, any()} | undefined).
 env(Key) -> application:get_env(?APP, Key).
 
 %% @doc Get environment
--spec(env(Key:: atom(), Default:: any()) -> undefined | any()).
+-spec(env(Key :: atom(), Default :: any()) -> undefined | any()).
 env(Key, Default) -> application:get_env(?APP, Key, Default).
 
 %% @doc Is running?
@@ -88,15 +86,15 @@ is_running(Node) ->
 %%--------------------------------------------------------------------
 
 %% @doc Subscribe
--spec(subscribe(iodata()) -> ok | {error, any()}).
+-spec(subscribe(iodata()) -> ok | {error, term()}).
 subscribe(Topic) ->
-    subscribe(Topic, self()).
+    emqttd_server:subscribe(iolist_to_binary(Topic)).
 
--spec(subscribe(iodata(), subscriber()) -> ok | {error, any()}).
+-spec(subscribe(iodata(), subscriber()) -> ok | {error, term()}).
 subscribe(Topic, Subscriber) ->
-    subscribe(Topic, Subscriber, []).
+    emqttd_server:subscribe(iolist_to_binary(Topic), Subscriber).
 
--spec(subscribe(iodata(), subscriber(), [suboption()]) -> ok | pubsub_error()).
+-spec(subscribe(iodata(), subscriber(), [suboption()]) -> ok | {error, term()}).
 subscribe(Topic, Subscriber, Options) ->
     emqttd_server:subscribe(iolist_to_binary(Topic), Subscriber, Options).
 
@@ -106,11 +104,11 @@ publish(Msg) ->
     emqttd_server:publish(Msg).
 
 %% @doc Unsubscribe
--spec(unsubscribe(iodata()) -> ok | pubsub_error()).
+-spec(unsubscribe(iodata()) -> ok | {error, term()}).
 unsubscribe(Topic) ->
-    unsubscribe(Topic, self()).
+    emqttd_server:unsubscribe(iolist_to_binary(Topic)).
 
--spec(unsubscribe(iodata(), subscriber()) -> ok | pubsub_error()).
+-spec(unsubscribe(iodata(), subscriber()) -> ok | {error, term()}).
 unsubscribe(Topic, Subscriber) ->
     emqttd_server:unsubscribe(iolist_to_binary(Topic), Subscriber).
 
@@ -125,34 +123,30 @@ topics() -> emqttd_router:topics().
 subscribers(Topic) ->
     emqttd_server:subscribers(iolist_to_binary(Topic)).
 
--spec(subscriptions(subscriber()) -> [{binary(), binary(), list(suboption())}]).
+-spec(subscriptions(subscriber()) -> [{emqttd:subscriber(), binary(), list(emqttd:suboption())}]).
 subscriptions(Subscriber) ->
     emqttd_server:subscriptions(Subscriber).
 
--spec(is_subscribed(iodata(), subscriber()) -> boolean()).
-is_subscribed(Topic, Subscriber) ->
-    emqttd_server:is_subscribed(iolist_to_binary(Topic), Subscriber).
-
--spec(subscriber_down(subscriber()) -> ok).
-subscriber_down(Subscriber) ->
-    emqttd_server:subscriber_down(Subscriber).
+-spec(subscribed(iodata(), subscriber()) -> boolean()).
+subscribed(Topic, Subscriber) ->
+    emqttd_server:subscribed(iolist_to_binary(Topic), Subscriber).
 
 %%--------------------------------------------------------------------
 %% Hooks API
 %%--------------------------------------------------------------------
 
 -spec(hook(atom(), function() | {emqttd_hooks:hooktag(), function()}, list(any()))
-      -> ok | {error, any()}).
+      -> ok | {error, term()}).
 hook(Hook, TagFunction, InitArgs) ->
     emqttd_hooks:add(Hook, TagFunction, InitArgs).
 
 -spec(hook(atom(), function() | {emqttd_hooks:hooktag(), function()}, list(any()), integer())
-      -> ok | {error, any()}).
+      -> ok | {error, term()}).
 hook(Hook, TagFunction, InitArgs, Priority) ->
     emqttd_hooks:add(Hook, TagFunction, InitArgs, Priority).
 
 -spec(unhook(atom(), function() | {emqttd_hooks:hooktag(), function()})
-      -> ok | {error, any()}).
+      -> ok | {error, term()}).
 unhook(Hook, TagFunction) ->
     emqttd_hooks:delete(Hook, TagFunction).
 

src/emqttd_access_control.erl

@@ -43,12 +43,12 @@
 %%--------------------------------------------------------------------
 
 %% @doc Start access control server.
--spec(start_link() -> {ok, pid()} | ignore | {error, any()}).
+-spec(start_link() -> {ok, pid()} | ignore | {error, term()}).
 start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
 %% @doc Authenticate MQTT Client.
--spec(auth(Client :: mqtt_client(), Password :: password()) -> ok | {ok, boolean()} | {error, any()}).
+-spec(auth(Client :: mqtt_client(), Password :: password()) -> ok | {ok, boolean()} | {error, term()}).
 auth(Client, Password) when is_record(Client, mqtt_client) ->
     auth(Client, Password, lookup_mods(auth)).
 auth(_Client, _Password, []) ->
@@ -88,16 +88,16 @@ reload_acl() ->
     [Mod:reload_acl(State) || {Mod, State, _Seq} <- lookup_mods(acl)].
 
 %% @doc Register Authentication or ACL module.
--spec(register_mod(auth | acl, atom(), list()) -> ok | {error, any()}).
+-spec(register_mod(auth | acl, atom(), list()) -> ok | {error, term()}).
 register_mod(Type, Mod, Opts) when Type =:= auth; Type =:= acl->
     register_mod(Type, Mod, Opts, 0).
 
--spec(register_mod(auth | acl, atom(), list(), non_neg_integer()) -> ok | {error, any()}).
+-spec(register_mod(auth | acl, atom(), list(), non_neg_integer()) -> ok | {error, term()}).
 register_mod(Type, Mod, Opts, Seq) when Type =:= auth; Type =:= acl->
     gen_server:call(?SERVER, {register_mod, Type, Mod, Opts, Seq}).
 
 %% @doc Unregister authentication or ACL module
--spec(unregister_mod(Type :: auth | acl, Mod :: atom()) -> ok | {error, any()}).
+-spec(unregister_mod(Type :: auth | acl, Mod :: atom()) -> ok | {error, not_found | term()}).
 unregister_mod(Type, Mod) when Type =:= auth; Type =:= acl ->
     gen_server:call(?SERVER, {unregister_mod, Type, Mod}).
 

src/emqttd_acl_mod.erl

@@ -32,7 +32,7 @@
                      PubSub :: pubsub(),
                      Topic  :: binary()}, State :: any()) -> allow | deny | ignore).
 
--callback(reload_acl(State :: any()) -> ok | {error, any()}).
+-callback(reload_acl(State :: any()) -> ok | {error, term()}).
 
 -callback(description() -> string()).
 

src/emqttd_bridge_sup.erl

@@ -23,7 +23,7 @@
 %%--------------------------------------------------------------------
 
 %% @doc Start bridge pool supervisor
--spec(start_link(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, any()}).
+-spec(start_link(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, term()}).
 start_link(Node, Topic, Options) ->
     MFA = {emqttd_bridge, start_link, [Node, Topic, Options]},
     emqttd_pool_sup:start_link({bridge, Node, Topic}, random, MFA).

src/emqttd_bridge_sup_sup.erl

@@ -40,11 +40,11 @@ bridges() ->
                              <- supervisor:which_children(?MODULE)].
 
 %% @doc Start a bridge
--spec(start_bridge(atom(), binary()) -> {ok, pid()} | {error, any()}).
+-spec(start_bridge(atom(), binary()) -> {ok, pid()} | {error, term()}).
 start_bridge(Node, Topic) when is_atom(Node) andalso is_binary(Topic) ->
     start_bridge(Node, Topic, []).
 
--spec(start_bridge(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, any()}).
+-spec(start_bridge(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, term()}).
 start_bridge(Node, _Topic, _Options) when Node =:= node() ->
     {error, bridge_to_self};
 start_bridge(Node, Topic, Options) when is_atom(Node) andalso is_binary(Topic) ->

src/emqttd_broker.erl

@@ -61,7 +61,7 @@
 %%--------------------------------------------------------------------
 
 %% @doc Start emqttd broker
--spec(start_link() -> {ok, pid()} | ignore | {error, any()}).
+-spec(start_link() -> {ok, pid()} | ignore | {error, term()}).
 start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 

src/emqttd_cli.erl

@@ -218,7 +218,9 @@ routes(["list"]) ->
     foreach(fun print/1, Routes);
 
 routes(["show", Topic]) ->
-    print(mnesia:dirty_read(mqtt_route, bin(Topic)));
+    Routes = lists:append(ets:lookup(mqtt_route, bin(Topic)),
+                          ets:lookup(mqtt_local_route, bin(Topic))),
+    foreach(fun print/1, Routes);
 
 routes(_) ->
     ?USAGE([{"routes list",         "List all routes"},
@@ -243,39 +245,38 @@ subscriptions(["list"]) ->
                   end, ets:tab2list(mqtt_subscription));
 
 subscriptions(["show", ClientId]) ->
-    case ets:lookup(mqtt_subscription, bin(ClientId)) of
-        []      -> ?PRINT_MSG("Not Found.~n");
-        Records -> [print(subscription, Subscription) || Subscription <- Records]
+    case emqttd:subscriptions(bin(ClientId)) of
+        [] ->
+            ?PRINT_MSG("Not Found.~n");
+        Subscriptions ->
+            [print(subscription, Sub) || Sub <- Subscriptions]
     end;
 
-
 subscriptions(["add", ClientId, Topic, QoS]) ->
-   Add = fun(IntQos) ->
-           case emqttd:subscribe(bin(Topic), bin(ClientId), [{qos, IntQos}]) of
-               ok ->
-                   ?PRINT_MSG("ok~n");
-               {error, Reason} ->
-                   ?PRINT("Error: ~p~n", [Reason])
-           end
-         end,
-   if_valid_qos(QoS, Add);
-
-
-
-subscriptions(["del", ClientId]) ->
-   Ok = emqttd:subscriber_down(bin(ClientId)),
-   ?PRINT("~p~n", [Ok]);
+   if_valid_qos(QoS, fun(IntQos) ->
+                        case emqttd_sm:lookup_session(bin(ClientId)) of
+                            undefined ->
+                                ?PRINT_MSG("Error: Session not found!");
+                            #mqtt_session{sess_pid = SessPid} ->
+                                {Topic1, Options} = emqttd_topic:parse(bin(Topic)),
+                                emqttd_session:subscribe(SessPid, [{Topic1, [{qos, IntQos}|Options]}]),
+                                ?PRINT_MSG("ok~n")
+                        end
+                     end);
 
 subscriptions(["del", ClientId, Topic]) ->
-   Ok = emqttd:unsubscribe(bin(Topic), bin(ClientId)),
-   ?PRINT("~p~n", [Ok]);
-
+    case emqttd_sm:lookup_session(bin(ClientId)) of
+        undefined ->
+            ?PRINT_MSG("Error: Session not found!");
+        #mqtt_session{sess_pid = SessPid} ->
+            emqttd_session:unsubscribe(SessPid, [emqttd_topic:parse(bin(Topic))]),
+            ?PRINT_MSG("ok~n")
+    end;
 
 subscriptions(_) ->
     ?USAGE([{"subscriptions list",                         "List all subscriptions"},
             {"subscriptions show <ClientId>",              "Show subscriptions of a client"},
             {"subscriptions add <ClientId> <Topic> <QoS>", "Add a static subscription manually"},
-            {"subscriptions del <ClientId>",               "Delete static subscriptions manually"},
             {"subscriptions del <ClientId> <Topic>",       "Delete a static subscription manually"}]).
 
 % if_could_print(Tab, Fun) ->
@@ -579,14 +580,25 @@ print({ClientId, _ClientPid, _Persistent, SessInfo}) ->
            "deliver_msg=~w, enqueue_msg=~w, created_at=~w)~n",
             [ClientId | [format(Key, get_value(Key, Data)) || Key <- InfoKeys]]).
 
-print(subscription, {Sub, {_Share, Topic}}) when is_pid(Sub) ->
+print(subscription, {Sub, {share, _Share, Topic}}) when is_pid(Sub) ->
     ?PRINT("~p -> ~s~n", [Sub, Topic]);
 print(subscription, {Sub, Topic}) when is_pid(Sub) ->
     ?PRINT("~p -> ~s~n", [Sub, Topic]);
-print(subscription, {Sub, {_Share, Topic}}) ->
-    ?PRINT("~s -> ~s~n", [Sub, Topic]);
-print(subscription, {Sub, Topic}) ->
-    ?PRINT("~s -> ~s~n", [Sub, Topic]).
+print(subscription, {{SubId, SubPid}, {share, _Share, Topic}})
+    when is_binary(SubId), is_pid(SubPid) ->
+    ?PRINT("~s~p -> ~s~n", [SubId, SubPid, Topic]);
+print(subscription, {{SubId, SubPid}, Topic})
+    when is_binary(SubId), is_pid(SubPid) ->
+    ?PRINT("~s~p -> ~s~n", [SubId, SubPid, Topic]);
+print(subscription, {Sub, Topic, Props}) ->
+    print(subscription, {Sub, Topic}),
+    lists:foreach(fun({K, V}) when is_binary(V) ->
+                      ?PRINT("  ~-8s: ~s~n", [K, V]);
+                     ({K, V}) ->
+                      ?PRINT("  ~-8s: ~w~n", [K, V]);
+                     (K) ->
+                      ?PRINT("  ~-8s: true~n", [K])
+                  end, Props).
 
 format(created_at, Val) ->
     emqttd_time:now_secs(Val);

src/emqttd_cm.erl

@@ -47,7 +47,7 @@
 %%--------------------------------------------------------------------
 
 %% @doc Start Client Manager
--spec(start_link(atom(), pos_integer(), fun()) -> {ok, pid()} | ignore | {error, any()}).
+-spec(start_link(atom(), pos_integer(), fun()) -> {ok, pid()} | ignore | {error, term()}).
 start_link(Pool, Id, StatsFun) ->
     gen_server2:start_link(?MODULE, [Pool, Id, StatsFun], []).
 

src/emqttd_gen_mod.erl

@@ -24,9 +24,9 @@
 
 -ifdef(use_specs).
 
--callback(load(Opts :: any()) -> ok | {error, any()}).
+-callback(load(Opts :: any()) -> ok | {error, term()}).
 
--callback(unload(State :: any()) -> any()).
+-callback(unload(State :: term()) -> term()).
 
 -else.
 

src/emqttd_keepalive.erl

@@ -29,7 +29,7 @@
 -export_type([keepalive/0]).
 
 %% @doc Start a keepalive
--spec(start(fun(), integer(), any()) -> {ok, keepalive()} | {error, any()}).
+-spec(start(fun(), integer(), any()) -> {ok, keepalive()} | {error, term()}).
 start(_, 0, _) ->
     {ok, #keepalive{}};
 start(StatFun, TimeoutSec, TimeoutMsg) ->
@@ -43,7 +43,7 @@ start(StatFun, TimeoutSec, TimeoutMsg) ->
     end.
 
 %% @doc Check keepalive, called when timeout.
--spec(check(keepalive()) -> {ok, keepalive()} | {error, any()}).
+-spec(check(keepalive()) -> {ok, keepalive()} | {error, term()}).
 check(KeepAlive = #keepalive{statfun = StatFun, statval = LastVal, repeat = Repeat}) ->
     case StatFun() of
         {ok, NewVal} ->

src/emqttd_metrics.erl

@@ -96,8 +96,8 @@
 %% API
 %%--------------------------------------------------------------------
 
-%% @doc Start metrics server
--spec(start_link() -> {ok, pid()} | ignore | {error, any()}).
+%% @doc Start the metrics server
+-spec(start_link() -> {ok, pid()} | ignore | {error, term()}).
 start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 

src/emqttd_mgmt.erl

@@ -455,29 +455,36 @@ subscription_list(Key, PageNo, PageSize) when ?EMPTY_KEY(Key) ->
     query_table(Qh, PageNo, PageSize, TotalNum);
 
 subscription_list(Key, PageNo, PageSize) ->
-    Keys = ets:lookup(mqtt_subscription, Key),
-    Fun = case length(Keys) == 0 of
-        true ->
-            MP = {{Key, '_'}, '_'},
-            fun() -> ets:match_object(mqtt_subproperty, MP) end;
-        false ->
-            fun() ->
-                lists:map(fun({S, T}) ->[R] = ets:lookup(mqtt_subproperty, {T, S}), R end, Keys)
-            end
-    end,
+    Fun = fun() -> ets:match_object(mqtt_subproperty, {{'_', {Key, '_'}}, '_'}) end,
     lookup_table(Fun, PageNo, PageSize).
 
 route_list(Topic, PageNo, PageSize) when ?EMPTY_KEY(Topic) ->
-    TotalNum = lists:sum([ets:info(Tab, size) || Tab <- tables()]),
-    Qh = qlc:append([qlc:q([E || E <- ets:table(Tab)]) || Tab <- tables()]),
-    query_table(Qh, PageNo, PageSize, TotalNum);
+    Tables = [mqtt_route],
+    TotalNum = lists:sum([ets:info(Tab, size) || Tab <- [mqtt_route, mqtt_local_route]]),
+    Qh = qlc:append([qlc:q([E || E <- ets:table(Tab)]) || Tab <- Tables]),
+    Data = query_table(Qh, PageNo, PageSize, TotalNum),
+    Route = get_value(result, Data),
+    LocalRoute = local_route_list(Topic, PageNo, PageSize),
+    lists:keyreplace(result, 1, Data, {result, lists:append(Route, LocalRoute)});
 
 route_list(Topic, PageNo, PageSize) ->
-    Fun = fun() -> lists:append([ets:lookup(Tab, Topic) || Tab <- tables()]) end,
-    lookup_table(Fun, PageNo, PageSize).
+    Tables = [mqtt_route],
+    Fun = fun() -> lists:append([ets:lookup(Tab, Topic) || Tab <- Tables]) end,
+    Route = lookup_table(Fun, PageNo, PageSize),
+    LocalRoute = local_route_list(Topic, PageNo, PageSize),
+    lists:append(Route, LocalRoute).
+
+local_route_list(Topic, PageNo, PageSize) when ?EMPTY_KEY(Topic) ->
+    TotalNum = lists:sum([ets:info(Tab, size) || Tab <- [mqtt_local_route]]),
+    Qh = qlc:append([qlc:q([E || E <- ets:table(Tab)]) || Tab <- [mqtt_local_route]]),
+    Data = query_table(Qh, PageNo, PageSize, TotalNum),
+    lists:map(fun({Topic1, Node}) -> {<<"$local/", Topic1/binary>>, Node} end, get_value(result, Data));
+
+local_route_list(Topic, PageNo, PageSize) ->
+    Fun = fun() -> lists:append([ets:lookup(Tab, Topic) || Tab <- [mqtt_local_route]]) end,
+    Data = lookup_table(Fun, PageNo, PageSize),
+    lists:map(fun({Topic1, Node}) -> {<<"$local/", Topic1/binary>>, Node} end, Data).
 
-tables() ->
-    [mqtt_route, mqtt_local_route].
 
 format_error(Val, Msg) ->
     re:replace(Msg, <<"\\$\\{[^}]+\\}">>, Val, [global, {return, binary}]).

src/emqttd_mod_sup.erl

@@ -42,7 +42,7 @@ start_child(ChildSpec) when is_tuple(ChildSpec) ->
 start_child(Mod, Type) when is_atom(Mod) andalso is_atom(Type) ->
     supervisor:start_child(?MODULE, ?CHILD(Mod, Type)).
 
--spec(stop_child(any()) -> ok | {error, any()}).
+-spec(stop_child(any()) -> ok | {error, term()}).
 stop_child(ChildId) ->
     case supervisor:terminate_child(?MODULE, ChildId) of
         ok    -> supervisor:delete_child(?MODULE, ChildId);

src/emqttd_parser.erl

@@ -39,7 +39,7 @@ initial_state(MaxSize) ->
 
 %% @doc Parse MQTT Packet
 -spec(parse(binary(), {none, pos_integer()} | fun())
-            -> {ok, mqtt_packet()} | {error, any()} | {more, fun()}).
+            -> {ok, mqtt_packet()} | {error, term()} | {more, fun()}).
 parse(<<>>, {none, MaxLen}) ->
     {more, fun(Bin) -> parse(Bin, {none, MaxLen}) end};
 parse(<<Type:4, Dup:1, QoS:2, Retain:1, Rest/binary>>, {none, Limit}) ->
@@ -124,7 +124,7 @@ parse_frame(Bin, #mqtt_packet_header{type = Type, qos  = Qos} = Header, Length)
                                       _ -> <<Id:16/big, R/binary>> = Rest1,
                                           {Id, R}
                                   end,
-            wrap(Header, #mqtt_packet_publish{topic_name = TopicName,
+            wrap(fixdup(Header), #mqtt_packet_publish{topic_name = TopicName,
                                               packet_id = PacketId},
                  Payload, Rest);
         {?PUBACK, <<FrameBin:Length/binary, Rest/binary>>} ->
@@ -222,3 +222,9 @@ fixqos(?SUBSCRIBE, 0)   -> 1;
 fixqos(?UNSUBSCRIBE, 0) -> 1;
 fixqos(_Type, QoS)      -> QoS.
 
+%% Fix Issue#1319
+fixdup(Header = #mqtt_packet_header{qos = ?QOS0, dup = true}) ->
+    Header#mqtt_packet_header{dup = false};
+fixdup(Header = #mqtt_packet_header{qos = ?QOS2, dup = true}) ->
+    Header#mqtt_packet_header{dup = false};
+fixdup(Header) -> Header.

src/emqttd_plugins.erl

@@ -47,7 +47,7 @@ init_config(CfgFile) ->
                   end, AppsEnv).
 
 %% @doc Load all plugins when the broker started.
--spec(load() -> list() | {error, any()}).
+-spec(load() -> list() | {error, term()}).
 load() ->
     case emqttd:env(plugins_loaded_file) of
         {ok, File} ->
@@ -80,7 +80,7 @@ load_plugins(Names, Persistent) ->
     [load_plugin(find_plugin(Name, Plugins), Persistent) || Name <- NeedToLoad].
 
 %% @doc Unload all plugins before broker stopped.
--spec(unload() -> list() | {error, any()}).
+-spec(unload() -> list() | {error, term()}).
 unload() ->
     case emqttd:env(plugins_loaded_file) of
         {ok, File} ->
@@ -119,7 +119,7 @@ plugin(CfgFile) ->
     #mqtt_plugin{name = AppName, version = Ver, descr = Descr}.
 
 %% @doc Load a Plugin
--spec(load(atom()) -> ok | {error, any()}).
+-spec(load(atom()) -> ok | {error, term()}).
 load(PluginName) when is_atom(PluginName) ->
     case lists:member(PluginName, names(started_app)) of
         true ->
@@ -172,7 +172,7 @@ find_plugin(Name, Plugins) ->
     lists:keyfind(Name, 2, Plugins). 
 
 %% @doc UnLoad a Plugin
--spec(unload(atom()) -> ok | {error, any()}).
+-spec(unload(atom()) -> ok | {error, term()}).
 unload(PluginName) when is_atom(PluginName) ->
     case {lists:member(PluginName, names(started_app)), lists:member(PluginName, names(plugin))} of
         {true, true} ->

src/emqttd_pool_sup.erl

@@ -36,12 +36,12 @@ spec(ChildId, Args) ->
     {ChildId, {?MODULE, start_link, Args},
         transient, infinity, supervisor, [?MODULE]}.
 
--spec(start_link(atom() | tuple(), atom(), mfa()) -> {ok, pid()} | {error, any()}).
+-spec(start_link(atom() | tuple(), atom(), mfa()) -> {ok, pid()} | {error, term()}).
 start_link(Pool, Type, MFA) ->
     Schedulers = erlang:system_info(schedulers),
     start_link(Pool, Type, Schedulers, MFA).
 
--spec(start_link(atom(), atom(), pos_integer(), mfa()) -> {ok, pid()} | {error, any()}).
+-spec(start_link(atom(), atom(), pos_integer(), mfa()) -> {ok, pid()} | {error, term()}).
 start_link(Pool, Type, Size, MFA) ->
     supervisor:start_link(?MODULE, [Pool, Type, Size, MFA]).
 

src/emqttd_pooler.erl

@@ -40,7 +40,7 @@ start_link() ->
 %% API
 %%--------------------------------------------------------------------
 
--spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, any()}).
+-spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, term()}).
 start_link(Pool, Id) ->
     gen_server:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id], []).
 

src/emqttd_protocol.erl

@@ -124,7 +124,7 @@ session(#proto_state{session = Session}) ->
 %% CONNECT – Client requests a connection to a Server
 
 %% A Client can only send the CONNECT Packet once over a Network Connection. 
--spec(received(mqtt_packet(), proto_state()) -> {ok, proto_state()} | {error, any()}).
+-spec(received(mqtt_packet(), proto_state()) -> {ok, proto_state()} | {error, term()}).
 received(Packet = ?PACKET(?CONNECT),
          State = #proto_state{connected = false, stats_data = Stats}) ->
     trace(recv, Packet, State), Stats1 = inc_stats(recv, ?CONNECT, Stats),

src/emqttd_pubsub.erl

@@ -46,7 +46,7 @@
 %% Start PubSub
 %%--------------------------------------------------------------------
 
--spec(start_link(atom(), pos_integer(), list()) -> {ok, pid()} | ignore | {error, any()}).
+-spec(start_link(atom(), pos_integer(), list()) -> {ok, pid()} | ignore | {error, term()}).
 start_link(Pool, Id, Env) ->
     gen_server2:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id, Env], []).
 
@@ -54,7 +54,7 @@ start_link(Pool, Id, Env) ->
 %% PubSub API
 %%--------------------------------------------------------------------
 
-%% @doc Subscribe a Topic
+%% @doc Subscribe to a Topic
 -spec(subscribe(binary(), emqttd:subscriber(), [emqttd:suboption()]) -> ok).
 subscribe(Topic, Subscriber, Options) ->
     call(pick(Topic), {subscribe, Topic, Subscriber, Options}).
@@ -63,8 +63,8 @@ subscribe(Topic, Subscriber, Options) ->
 async_subscribe(Topic, Subscriber, Options) ->
     cast(pick(Topic), {subscribe, Topic, Subscriber, Options}).
 
-%% @doc Publish MQTT Message to Topic
--spec(publish(binary(), any()) -> {ok, mqtt_delivery()} | ignore).
+%% @doc Publish MQTT Message to Topic.
+-spec(publish(binary(), mqtt_message()) -> {ok, mqtt_delivery()} | ignore).
 publish(Topic, Msg) ->
     route(lists:append(emqttd_router:match(Topic),
                        emqttd_router:match_local(Topic)), delivery(Msg)).
@@ -72,7 +72,7 @@ publish(Topic, Msg) ->
 route([], #mqtt_delivery{message = #mqtt_message{topic = Topic}}) ->
     dropped(Topic), ignore;
 
-%% Dispatch on the local node
+%% Dispatch on the local node.
 route([#mqtt_route{topic = To, node = Node}],
       Delivery = #mqtt_delivery{flows = Flows}) when Node =:= node() ->
     dispatch(To, Delivery#mqtt_delivery{flows = [{route, Node, To} | Flows]});
@@ -82,8 +82,8 @@ route([#mqtt_route{topic = To, node = Node}], Delivery = #mqtt_delivery{flows =
     forward(Node, To, Delivery#mqtt_delivery{flows = [{route, Node, To}|Flows]});
 
 route(Routes, Delivery) ->
-    {ok, lists:foldl(fun(Route, DelAcc) ->
-                    {ok, DelAcc1} = route([Route], DelAcc), DelAcc1
+    {ok, lists:foldl(fun(Route, Acc) ->
+                    {ok, Acc1} = route([Route], Acc), Acc1
             end, Delivery, Routes)}.
 
 delivery(Msg) -> #mqtt_delivery{sender = self(), message = Msg, flows = []}.
@@ -92,7 +92,7 @@ delivery(Msg) -> #mqtt_delivery{sender = self(), message = Msg, flows = []}.
 forward(Node, To, Delivery) ->
     rpc:cast(Node, ?PUBSUB, dispatch, [To, Delivery]), {ok, Delivery}.
 
-%% @doc Dispatch Message to Subscribers
+%% @doc Dispatch Message to Subscribers.
 -spec(dispatch(binary(), mqtt_delivery()) -> mqtt_delivery()).
 dispatch(Topic, Delivery = #mqtt_delivery{message = Msg, flows = Flows}) ->
     case subscribers(Topic) of
@@ -107,16 +107,16 @@ dispatch(Topic, Delivery = #mqtt_delivery{message = Msg, flows = Flows}) ->
             {ok, Delivery#mqtt_delivery{flows = Flows1}}
     end.
 
-dispatch(Pid, Topic, Msg) when is_pid(Pid) ->
-    Pid ! {dispatch, Topic, Msg};
-dispatch(SubId, Topic, Msg) when is_binary(SubId) ->
-    emqttd_sm:dispatch(SubId, Topic, Msg);
-dispatch({_Share, [Sub]}, Topic, Msg) ->
+%%TODO: Is SubPid aliving???
+dispatch(SubPid, Topic, Msg) when is_pid(SubPid) ->
+    SubPid ! {dispatch, Topic, Msg};
+dispatch({SubId, SubPid}, Topic, Msg) when is_binary(SubId), is_pid(SubPid) ->
+    SubPid ! {dispatch, Topic, Msg};
+dispatch({{share, _Share}, [Sub]}, Topic, Msg) ->
     dispatch(Sub, Topic, Msg);
-dispatch({_Share, []}, _Topic, _Msg) ->
+dispatch({{share, _Share}, []}, _Topic, _Msg) ->
     ok;
-%%TODO: round-robbin
-dispatch({_Share, Subs}, Topic, Msg) ->
+dispatch({{share, _Share}, Subs}, Topic, Msg) -> %% round-robbin?
     dispatch(lists:nth(rand:uniform(length(Subs)), Subs), Topic, Msg).
 
 subscribers(Topic) ->
@@ -126,8 +126,8 @@ group_by_share([]) -> [];
 
 group_by_share(Subscribers) ->
     {Subs1, Shares1} =
-    lists:foldl(fun({Share, Sub}, {Subs, Shares}) ->
-                    {Subs, dict:append(Share, Sub, Shares)};
+    lists:foldl(fun({share, Share, Sub}, {Subs, Shares}) ->
+                    {Subs, dict:append({share, Share}, Sub, Shares)};
                    (Sub, {Subs, Shares}) ->
                     {[Sub|Subs], Shares}
                 end, {[], dict:new()}, Subscribers),
@@ -155,8 +155,8 @@ call(PubSub, Req) when is_pid(PubSub) ->
 cast(PubSub, Msg) when is_pid(PubSub) ->
     gen_server2:cast(PubSub, Msg).
 
-pick(Subscriber) ->
-    gproc_pool:pick_worker(pubsub, Subscriber).
+pick(Topic) ->
+    gproc_pool:pick_worker(pubsub, Topic).
 
 %%--------------------------------------------------------------------
 %% gen_server Callbacks
@@ -169,22 +169,22 @@ init([Pool, Id, Env]) ->
 
 handle_call({subscribe, Topic, Subscriber, Options}, _From, State) ->
     add_subscriber(Topic, Subscriber, Options),
-    {reply, ok, setstats(State), hibernate};
+    reply(ok, setstats(State));
 
 handle_call({unsubscribe, Topic, Subscriber, Options}, _From, State) ->
     del_subscriber(Topic, Subscriber, Options),
-    {reply, ok, setstats(State), hibernate};
+    reply(ok, setstats(State));
 
 handle_call(Req, _From, State) ->
     ?UNEXPECTED_REQ(Req, State).
 
 handle_cast({subscribe, Topic, Subscriber, Options}, State) ->
     add_subscriber(Topic, Subscriber, Options),
-    {noreply, setstats(State), hibernate};
+    noreply(setstats(State));
 
 handle_cast({unsubscribe, Topic, Subscriber, Options}, State) ->
     del_subscriber(Topic, Subscriber, Options),
-    {noreply, setstats(State), hibernate};
+    noreply(setstats(State));
 
 handle_cast(Msg, State) ->
     ?UNEXPECTED_MSG(Msg, State).
@@ -205,39 +205,48 @@ code_change(_OldVsn, State, _Extra) ->
 add_subscriber(Topic, Subscriber, Options) ->
     Share = proplists:get_value(share, Options),
     case ?is_local(Options) of
-        false -> add_subscriber_(Share, Topic, Subscriber);
-        true  -> add_local_subscriber_(Share, Topic, Subscriber)
+        false -> add_global_subscriber(Share, Topic, Subscriber);
+        true  -> add_local_subscriber(Share, Topic, Subscriber)
     end.
 
-add_subscriber_(Share, Topic, Subscriber) ->
-    (not ets:member(mqtt_subscriber, Topic)) andalso emqttd_router:add_route(Topic),
+add_global_subscriber(Share, Topic, Subscriber) ->
+    case ets:member(mqtt_subscriber, Topic) and emqttd_router:has_route(Topic) of
+        true  -> ok;
+        false -> emqttd_router:add_route(Topic)
+    end,
     ets:insert(mqtt_subscriber, {Topic, shared(Share, Subscriber)}).
 
-add_local_subscriber_(Share, Topic, Subscriber) ->
+add_local_subscriber(Share, Topic, Subscriber) ->
     (not ets:member(mqtt_subscriber, {local, Topic})) andalso emqttd_router:add_local_route(Topic),
     ets:insert(mqtt_subscriber, {{local, Topic}, shared(Share, Subscriber)}).
 
 del_subscriber(Topic, Subscriber, Options) ->
     Share = proplists:get_value(share, Options),
     case ?is_local(Options) of
-        false -> del_subscriber_(Share, Topic, Subscriber);
-        true  -> del_local_subscriber_(Share, Topic, Subscriber)
+        false -> del_global_subscriber(Share, Topic, Subscriber);
+        true  -> del_local_subscriber(Share, Topic, Subscriber)
     end.
 
-del_subscriber_(Share, Topic, Subscriber) ->
+del_global_subscriber(Share, Topic, Subscriber) ->
     ets:delete_object(mqtt_subscriber, {Topic, shared(Share, Subscriber)}),
     (not ets:member(mqtt_subscriber, Topic)) andalso emqttd_router:del_route(Topic).
 
-del_local_subscriber_(Share, Topic, Subscriber) ->
+del_local_subscriber(Share, Topic, Subscriber) ->
     ets:delete_object(mqtt_subscriber, {{local, Topic}, shared(Share, Subscriber)}),
     (not ets:member(mqtt_subscriber, {local, Topic})) andalso emqttd_router:del_local_route(Topic).
 
 shared(undefined, Subscriber) ->
     Subscriber;
 shared(Share, Subscriber) ->
-    {Share, Subscriber}.
+    {share, Share, Subscriber}.
 
 setstats(State) ->
     emqttd_stats:setstats('subscribers/count', 'subscribers/max', ets:info(mqtt_subscriber, size)),
     State.
 
+reply(Reply, State) ->
+    {reply, Reply, State, hibernate}.
+
+noreply(State) ->
+    {noreply, State, hibernate}.
+

src/emqttd_rest_api.erl

@@ -241,10 +241,14 @@ subscription_list('GET', Params, Node, Key) ->
     Data = emqttd_mgmt:subscription_list(l2a(Node), l2b(Key), PageNo, PageSize),
     {ok, [{objects, [subscription_row(Row) || Row <- Data]}]}.
 
-subscription_row({{Topic, ClientId}, Option}) when is_pid(ClientId) ->
-    subscription_row({{Topic, l2b(pid_to_list(ClientId))}, Option});    
-subscription_row({{Topic, ClientId}, Option}) ->
-    Qos = get_value(qos, Option),
+subscription_row({{Topic, SubPid}, Options}) when is_pid(SubPid) ->
+    subscription_row({{Topic, {undefined, SubPid}}, Options});
+subscription_row({{Topic, {SubId, SubPid}}, Options}) ->
+    Qos = proplists:get_value(qos, Options),
+    ClientId = case SubId of
+                   undefined -> list_to_binary(pid_to_list(SubPid));
+                   SubId     -> SubId
+               end,
     [{client_id, ClientId}, {topic, Topic}, {qos, Qos}].
 
 %%--------------------------------------------------------------------------

src/emqttd_router.erl

@@ -28,15 +28,17 @@
 -boot_mnesia({mnesia, [boot]}).
 -copy_mnesia({mnesia, [copy]}).
 
-%% Start/Stop
--export([start_link/0, topics/0, local_topics/0, stop/0]).
+-export([start_link/0, topics/0, local_topics/0]).
+
+%% For eunit tests
+-export([start/0, stop/0]).
 
 %% Route APIs
--export([add_route/1, add_route/2, add_routes/1, match/1, print/1,
-         del_route/1, del_route/2, del_routes/1, has_route/1]).
+-export([add_route/1, del_route/1, match/1, print/1, has_route/1]).
 
 %% Local Route API
--export([add_local_route/1, del_local_route/1, match_local/1]).
+-export([get_local_routes/0, add_local_route/1, match_local/1,
+         del_local_route/1, clean_local_routes/0]).
 
 %% gen_server Function Exports
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
@@ -55,10 +57,6 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(mqtt_topic, [
-                {ram_copies, [node()]},
-                {record_name, mqtt_topic},
-                {attributes, record_info(fields, mqtt_topic)}]),
     ok = ekka_mnesia:create_table(mqtt_route, [
                 {type, bag},
                 {ram_copies, [node()]},
@@ -66,7 +64,6 @@ mnesia(boot) ->
                 {attributes, record_info(fields, mqtt_route)}]);
 
 mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(mqtt_topic),
     ok = ekka_mnesia:copy_table(mqtt_route, ram_copies).
 
 %%--------------------------------------------------------------------
@@ -77,19 +74,26 @@ start_link() ->
     gen_server:start_link({local, ?ROUTER}, ?MODULE, [], []).
 
 %%--------------------------------------------------------------------
-%% API
+%% Topics
 %%--------------------------------------------------------------------
 
+-spec(topics() -> list(binary())).
 topics() ->
     mnesia:dirty_all_keys(mqtt_route).
 
+-spec(local_topics() -> list(binary())).
 local_topics() ->
     ets:select(mqtt_local_route, [{{'$1', '_'}, [], ['$1']}]).
 
+%%--------------------------------------------------------------------
+%% Match API
+%%--------------------------------------------------------------------
+
 %% @doc Match Routes.
 -spec(match(Topic:: binary()) -> [mqtt_route()]).
 match(Topic) when is_binary(Topic) ->
-    Matched = mnesia:async_dirty(fun emqttd_trie:match/1, [Topic]),
+    %% Optimize: ets???
+    Matched = mnesia:ets(fun emqttd_trie:match/1, [Topic]),
     %% Optimize: route table will be replicated to all nodes.
     lists:append([ets:lookup(mqtt_route, To) || To <- [Topic | Matched]]).
 
@@ -99,93 +103,68 @@ print(Topic) ->
     [io:format("~s -> ~s~n", [To, Node]) ||
         #mqtt_route{topic = To, node = Node} <- match(Topic)].
 
-%% @doc Add Route
--spec(add_route(binary() | mqtt_route()) -> ok | {error, Reason :: any()}).
+%%--------------------------------------------------------------------
+%% Route Management API
+%%--------------------------------------------------------------------
+
+%% @doc Add Route.
+-spec(add_route(binary() | mqtt_route()) -> ok | {error, Reason :: term()}).
 add_route(Topic) when is_binary(Topic) ->
     add_route(#mqtt_route{topic = Topic, node = node()});
-add_route(Route) when is_record(Route, mqtt_route) ->
-    add_routes([Route]).
-
--spec(add_route(Topic :: binary(), Node :: node()) -> ok | {error, Reason :: any()}).
-add_route(Topic, Node) when is_binary(Topic), is_atom(Node) ->
-    add_route(#mqtt_route{topic = Topic, node = Node}).
-
-%% @doc Add Routes
--spec(add_routes([mqtt_route()]) -> ok | {error, Reason :: any()}).
-add_routes(Routes) ->
-    AddFun = fun() -> [add_route_(Route) || Route <- Routes] end,
-    case mnesia:is_transaction() of
-        true  -> AddFun();
-        false -> trans(AddFun)
+add_route(Route = #mqtt_route{topic = Topic}) ->
+    case emqttd_topic:wildcard(Topic) of
+        true  -> case mnesia:is_transaction() of
+                     true  -> add_trie_route(Route);
+                     false -> trans(fun add_trie_route/1, [Route])
+                 end;
+        false -> add_direct_route(Route)
     end.
 
-%% @private
-add_route_(Route = #mqtt_route{topic = Topic}) ->
+add_direct_route(Route) ->
+    mnesia:async_dirty(fun mnesia:write/1, [Route]).
+
+add_trie_route(Route = #mqtt_route{topic = Topic}) ->
     case mnesia:wread({mqtt_route, Topic}) of
-        [] ->
-            case emqttd_topic:wildcard(Topic) of
-                true  -> emqttd_trie:insert(Topic);
-                false -> ok
-            end,
-            mnesia:write(Route),
-            mnesia:write(#mqtt_topic{topic = Topic});
-        Records ->
-            case lists:member(Route, Records) of
-                true  -> ok;
-                false -> mnesia:write(Route)
-            end
-    end.
+        [] -> emqttd_trie:insert(Topic);
+        _  -> ok
+    end,
+    mnesia:write(Route).
 
 %% @doc Delete Route
--spec(del_route(binary() | mqtt_route()) -> ok | {error, Reason :: any()}).
+-spec(del_route(binary() | mqtt_route()) -> ok | {error, Reason :: term()}).
 del_route(Topic) when is_binary(Topic) ->
     del_route(#mqtt_route{topic = Topic, node = node()});
-del_route(Route) when is_record(Route, mqtt_route) ->
-    del_routes([Route]).
-
--spec(del_route(Topic :: binary(), Node :: node()) -> ok | {error, Reason :: any()}).
-del_route(Topic, Node) when is_binary(Topic), is_atom(Node) ->
-    del_route(#mqtt_route{topic = Topic, node = Node}).
-
-%% @doc Delete Routes
--spec(del_routes([mqtt_route()]) -> ok | {error, any()}).
-del_routes(Routes) ->
-    DelFun = fun() -> [del_route_(Route) || Route <- Routes] end,
-    case mnesia:is_transaction() of
-        true  -> DelFun();
-        false -> trans(DelFun)
+del_route(Route = #mqtt_route{topic = Topic}) ->
+    case emqttd_topic:wildcard(Topic) of
+        true  -> case mnesia:is_transaction() of
+                     true  -> del_trie_route(Route);
+                     false -> trans(fun del_trie_route/1, [Route])
+                 end;
+        false -> del_direct_route(Route)
     end.
 
-del_route_(Route = #mqtt_route{topic = Topic}) ->
+del_direct_route(Route) ->
+    mnesia:async_dirty(fun mnesia:delete_object/1, [Route]).
+
+del_trie_route(Route = #mqtt_route{topic = Topic}) ->
     case mnesia:wread({mqtt_route, Topic}) of
-        [] ->
-            ok;
-        [Route] ->
-            %% Remove route and trie
-            mnesia:delete_object(Route),
-            case emqttd_topic:wildcard(Topic) of
-                true  -> emqttd_trie:delete(Topic);
-                false -> ok
-            end,
-            mnesia:delete({mqtt_topic, Topic});
-        _More ->
-            %% Remove route only
-            mnesia:delete_object(Route)
+        [Route] -> %% Remove route and trie
+                   mnesia:delete_object(Route),
+                   emqttd_trie:delete(Topic);
+        [_|_]   -> %% Remove route only
+                   mnesia:delete_object(Route);
+        []      -> ok
     end.
 
-%% @doc Has Route?
+%% @doc Has route?
 -spec(has_route(binary()) -> boolean()).
-has_route(Topic) ->
-    Routes = case mnesia:is_transaction() of
-                 true  -> mnesia:read(mqtt_route, Topic);
-                 false -> mnesia:dirty_read(mqtt_route, Topic)
-             end,
-    length(Routes) > 0.
+has_route(Topic) when is_binary(Topic) ->
+    ets:member(mqtt_route, Topic).
 
 %% @private
--spec(trans(function()) -> ok | {error, any()}).
-trans(Fun) ->
-    case mnesia:transaction(Fun) of
+-spec(trans(function(), list(any())) -> ok | {error, term()}).
+trans(Fun, Args) ->
+    case mnesia:transaction(Fun, Args) of
         {atomic, _}      -> ok;
         {aborted, Error} -> {error, Error}
     end.
@@ -194,24 +173,44 @@ trans(Fun) ->
 %% Local Route API
 %%--------------------------------------------------------------------
 
+-spec(get_local_routes() -> list({binary(), node()})).
+get_local_routes() ->
+    ets:tab2list(mqtt_local_route).
+
 -spec(add_local_route(binary()) -> ok).
 add_local_route(Topic) ->
-    gen_server:cast(?ROUTER, {add_local_route, Topic}).
+    gen_server:call(?ROUTER, {add_local_route, Topic}).
     
 -spec(del_local_route(binary()) -> ok).
 del_local_route(Topic) ->
-    gen_server:cast(?ROUTER, {del_local_route, Topic}).
+    gen_server:call(?ROUTER, {del_local_route, Topic}).
     
 -spec(match_local(binary()) -> [mqtt_route()]).
 match_local(Name) ->
-    [#mqtt_route{topic = {local, Filter}, node = Node}
-        || {Filter, Node} <- ets:tab2list(mqtt_local_route),
-           emqttd_topic:match(Name, Filter)].
+    case ets:info(mqtt_local_route, size) of
+        0 -> [];
+        _ -> ets:foldl(
+               fun({Filter, Node}, Matched) ->
+                   case emqttd_topic:match(Name, Filter) of
+                       true  -> [#mqtt_route{topic = {local, Filter}, node = Node} | Matched];
+                       false -> Matched
+                   end
+               end, [], mqtt_local_route)
+    end.
+
+-spec(clean_local_routes() -> ok).
+clean_local_routes() ->
+    gen_server:call(?ROUTER, clean_local_routes).
 
 dump() ->
     [{route, ets:tab2list(mqtt_route)}, {local_route, ets:tab2list(mqtt_local_route)}].
 
-stop() -> gen_server:call(?ROUTER, stop).
+%% For unit test.
+start() ->
+    gen_server:start({local, ?ROUTER}, ?MODULE, [], []).
+
+stop() ->
+    gen_server:call(?ROUTER, stop).
 
 %%--------------------------------------------------------------------
 %% gen_server Callbacks
@@ -223,21 +222,25 @@ init([]) ->
     {ok, TRef} = timer:send_interval(timer:seconds(1), stats),
     {ok, #state{stats_timer = TRef}}.
 
+handle_call({add_local_route, Topic}, _From, State) ->
+    %% why node()...?
+    ets:insert(mqtt_local_route, {Topic, node()}),
+    {reply, ok, State};
+
+handle_call({del_local_route, Topic}, _From, State) ->
+    ets:delete(mqtt_local_route, Topic),
+    {reply, ok, State};
+
+handle_call(clean_local_routes, _From, State) ->
+    ets:delete_all_objects(mqtt_local_route),
+    {reply, ok, State};
+
 handle_call(stop, _From, State) ->
     {stop, normal, ok, State};
 
 handle_call(_Req, _From, State) ->
     {reply, ignore, State}.
 
-handle_cast({add_local_route, Topic}, State) ->
-    %% why node()...?
-    ets:insert(mqtt_local_route, {Topic, node()}),
-    {noreply, State};
-
-handle_cast({del_local_route, Topic}, State) ->
-    ets:delete(mqtt_local_route, Topic),
-    {noreply, State};
-
 handle_cast(_Msg, State) ->
     {noreply, State}.
 

src/emqttd_server.erl

@@ -37,8 +37,7 @@
          async_unsubscribe/1, async_unsubscribe/2]).
 
 %% Management API.
--export([setqos/3, subscriptions/1, subscribers/1, is_subscribed/2,
-         subscriber_down/1]).
+-export([setqos/3, subscriptions/1, subscribers/1, subscribed/2]).
 
 %% Debug API
 -export([dump/0]).
@@ -47,10 +46,10 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--record(state, {pool, id, env, submon :: emqttd_pmon:pmon()}).
+-record(state, {pool, id, env, subids :: map(), submon :: emqttd_pmon:pmon()}).
 
-%% @doc Start server
--spec(start_link(atom(), pos_integer(), list()) -> {ok, pid()} | ignore | {error, any()}).
+%% @doc Start the server
+-spec(start_link(atom(), pos_integer(), list()) -> {ok, pid()} | ignore | {error, term()}).
 start_link(Pool, Id, Env) ->
     gen_server2:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id, Env], []).
 
@@ -58,21 +57,21 @@ start_link(Pool, Id, Env) ->
 %% PubSub API
 %%--------------------------------------------------------------------
 
-%% @doc Subscribe a Topic
--spec(subscribe(binary()) -> ok | emqttd:pubsub_error()).
+%% @doc Subscribe to a Topic.
+-spec(subscribe(binary()) -> ok | {error, term()}).
 subscribe(Topic) when is_binary(Topic) ->
     subscribe(Topic, self()).
 
--spec(subscribe(binary(), emqttd:subscriber()) -> ok | emqttd:pubsub_error()).
+-spec(subscribe(binary(), emqttd:subscriber()) -> ok | {error, term()}).
 subscribe(Topic, Subscriber) when is_binary(Topic) ->
     subscribe(Topic, Subscriber, []).
 
 -spec(subscribe(binary(), emqttd:subscriber(), [emqttd:suboption()]) ->
-      ok | emqttd:pubsub_error()).
+      ok | {error, term()}).
 subscribe(Topic, Subscriber, Options) when is_binary(Topic) ->
-    call(pick(Subscriber), {subscribe, Topic, Subscriber, Options}).
+    call(pick(Subscriber), {subscribe, Topic, with_subpid(Subscriber), Options}).
 
-%% @doc Subscribe a Topic Asynchronously
+%% @doc Subscribe to a Topic asynchronously.
 -spec(async_subscribe(binary()) -> ok).
 async_subscribe(Topic) when is_binary(Topic) ->
     async_subscribe(Topic, self()).
@@ -83,7 +82,7 @@ async_subscribe(Topic, Subscriber) when is_binary(Topic) ->
 
 -spec(async_subscribe(binary(), emqttd:subscriber(), [emqttd:suboption()]) -> ok).
 async_subscribe(Topic, Subscriber, Options) when is_binary(Topic) ->
-    cast(pick(Subscriber), {subscribe, Topic, Subscriber, Options}).
+    cast(pick(Subscriber), {subscribe, Topic, with_subpid(Subscriber), Options}).
 
 %% @doc Publish message to Topic.
 -spec(publish(mqtt_message()) -> {ok, mqtt_delivery()} | ignore).
@@ -109,14 +108,14 @@ trace(publish, From, #mqtt_message{topic = Topic, payload = Payload}) ->
                 "~s PUBLISH to ~s: ~p", [From, Topic, Payload]).
 
 %% @doc Unsubscribe
--spec(unsubscribe(binary()) -> ok | emqttd:pubsub_error()).
+-spec(unsubscribe(binary()) -> ok | {error, term()}).
 unsubscribe(Topic) when is_binary(Topic) ->
     unsubscribe(Topic, self()).
 
 %% @doc Unsubscribe
--spec(unsubscribe(binary(), emqttd:subscriber()) -> ok | emqttd:pubsub_error()).
+-spec(unsubscribe(binary(), emqttd:subscriber()) -> ok | {error, term()}).
 unsubscribe(Topic, Subscriber) when is_binary(Topic) ->
-    call(pick(Subscriber), {unsubscribe, Topic, Subscriber}).
+    call(pick(Subscriber), {unsubscribe, Topic, with_subpid(Subscriber)}).
 
 %% @doc Async Unsubscribe
 -spec(async_unsubscribe(binary()) -> ok).
@@ -125,32 +124,47 @@ async_unsubscribe(Topic) when is_binary(Topic) ->
 
 -spec(async_unsubscribe(binary(), emqttd:subscriber()) -> ok).
 async_unsubscribe(Topic, Subscriber) when is_binary(Topic) ->
-    cast(pick(Subscriber), {unsubscribe, Topic, Subscriber}).
+    cast(pick(Subscriber), {unsubscribe, Topic, with_subpid(Subscriber)}).
 
+-spec(setqos(binary(), emqttd:subscriber(), mqtt_qos()) -> ok).
 setqos(Topic, Subscriber, Qos) when is_binary(Topic) ->
-    call(pick(Subscriber), {setqos, Topic, Subscriber, Qos}).
-
--spec(subscriptions(emqttd:subscriber()) -> [{binary(), binary(), list(emqttd:suboption())}]).
-subscriptions(Subscriber) ->
-    lists:map(fun({_, {_Share, Topic}}) ->
-                subscription(Topic, Subscriber);
-                 ({_, Topic}) ->
-                subscription(Topic, Subscriber)
-        end, ets:lookup(mqtt_subscription, Subscriber)).
-
-subscription(Topic, Subscriber) ->
-    {Topic, Subscriber, ets:lookup_element(mqtt_subproperty, {Topic, Subscriber}, 2)}.
-
-subscribers(Topic) ->
+    call(pick(Subscriber), {setqos, Topic, with_subpid(Subscriber), Qos}).
+
+with_subpid(SubPid) when is_pid(SubPid) ->
+    SubPid;
+with_subpid(SubId) when is_binary(SubId) ->
+    {SubId, self()};
+with_subpid({SubId, SubPid}) when is_binary(SubId), is_pid(SubPid) ->
+    {SubId, SubPid}.
+
+-spec(subscriptions(emqttd:subscriber()) -> [{emqttd:subscriber(), binary(), list(emqttd:suboption())}]).
+subscriptions(SubPid) when is_pid(SubPid) ->
+    with_subproperty(ets:lookup(mqtt_subscription, SubPid));
+
+subscriptions(SubId) when is_binary(SubId) ->
+    with_subproperty(ets:match_object(mqtt_subscription, {{SubId, '_'}, '_'}));
+
+subscriptions({SubId, SubPid}) when is_binary(SubId), is_pid(SubPid) ->
+    with_subproperty(ets:lookup(mqtt_subscription, {SubId, SubPid})).
+
+with_subproperty({Subscriber, {share, _Share, Topic}}) ->
+    with_subproperty({Subscriber, Topic});
+with_subproperty({Subscriber, Topic}) ->
+    {Subscriber, Topic, ets:lookup_element(mqtt_subproperty, {Topic, Subscriber}, 2)};
+with_subproperty(Subscriptions) when is_list(Subscriptions) ->
+    [with_subproperty(Subscription) || Subscription <- Subscriptions].
+
+-spec(subscribers(binary()) -> list(emqttd:subscriber())).
+subscribers(Topic) when is_binary(Topic) ->
     emqttd_pubsub:subscribers(Topic).
 
--spec(is_subscribed(binary(), emqttd:subscriber()) -> boolean()).
-is_subscribed(Topic, Subscriber) when is_binary(Topic) ->
-    ets:member(mqtt_subproperty, {Topic, Subscriber}).
-
--spec(subscriber_down(emqttd:subscriber()) -> ok).
-subscriber_down(Subscriber) ->
-    cast(pick(Subscriber), {subscriber_down, Subscriber}).
+-spec(subscribed(binary(), emqttd:subscriber()) -> boolean()).
+subscribed(Topic, SubPid) when is_binary(Topic), is_pid(SubPid) ->
+    ets:member(mqtt_subproperty, {Topic, SubPid});
+subscribed(Topic, SubId) when is_binary(Topic), is_binary(SubId) ->
+    length(ets:match_object(mqtt_subproperty, {{Topic, {SubId, '_'}}, '_'}, 1)) == 1;
+subscribed(Topic, {SubId, SubPid}) when is_binary(Topic), is_binary(SubId), is_pid(SubPid) ->
+    ets:member(mqtt_subproperty, {Topic, {SubId, SubPid}}).
 
 call(Server, Req) ->
     gen_server2:call(Server, Req, infinity).
@@ -158,8 +172,12 @@ call(Server, Req) ->
 cast(Server, Msg) when is_pid(Server) ->
     gen_server2:cast(Server, Msg).
 
-pick(Subscriber) ->
-    gproc_pool:pick_worker(server, Subscriber).
+pick(SubPid) when is_pid(SubPid) ->
+    gproc_pool:pick_worker(server, SubPid);
+pick(SubId) when is_binary(SubId) ->
+    gproc_pool:pick_worker(server, SubId);
+pick({SubId, SubPid}) when is_binary(SubId), is_pid(SubPid) ->
+    pick(SubId).
 
 dump() ->
     [{Tab, ets:tab2list(Tab)} || Tab <- [mqtt_subproperty, mqtt_subscription, mqtt_subscriber]].
@@ -170,18 +188,20 @@ dump() ->
 
 init([Pool, Id, Env]) ->
     ?GPROC_POOL(join, Pool, Id),
-    {ok, #state{pool = Pool, id = Id, env = Env, submon = emqttd_pmon:new()}}.
+    State = #state{pool = Pool, id = Id, env = Env,
+                   subids = #{}, submon = emqttd_pmon:new()},
+    {ok, State, hibernate, {backoff, 2000, 2000, 20000}}.
 
 handle_call({subscribe, Topic, Subscriber, Options}, _From, State) ->
-    case do_subscribe_(Topic, Subscriber, Options, State) of
-        {ok, NewState} -> {reply, ok, setstats(NewState)};
-        {error, Error} -> {reply, {error, Error}, State}
+    case do_subscribe(Topic, Subscriber, Options, State) of
+        {ok, NewState} -> reply(ok, setstats(NewState));
+        {error, Error} -> reply({error, Error}, State)
     end;
 
 handle_call({unsubscribe, Topic, Subscriber}, _From, State) ->
-    case do_unsubscribe_(Topic, Subscriber, State) of
-        {ok, NewState} -> {reply, ok, setstats(NewState), hibernate};
-        {error, Error} -> {reply, {error, Error}, State}
+    case do_unsubscribe(Topic, Subscriber, State) of
+        {ok, NewState} -> reply(ok, setstats(NewState));
+        {error, Error} -> reply({error, Error}, State)
     end;
 
 handle_call({setqos, Topic, Subscriber, Qos}, _From, State) ->
@@ -190,36 +210,37 @@ handle_call({setqos, Topic, Subscriber, Qos}, _From, State) ->
         [{_, Opts}] ->
             Opts1 = lists:ukeymerge(1, [{qos, Qos}], Opts),
             ets:insert(mqtt_subproperty, {Key, Opts1}),
-            {reply, ok, State};
+            reply(ok, State);
         [] ->
-            {reply, {error, {subscription_not_found, Topic}}, State}
+            reply({error, {subscription_not_found, Topic}}, State)
     end;
 
 handle_call(Req, _From, State) ->
     ?UNEXPECTED_REQ(Req, State).
 
 handle_cast({subscribe, Topic, Subscriber, Options}, State) ->
-    case do_subscribe_(Topic, Subscriber, Options, State) of
-        {ok, NewState}  -> {noreply, setstats(NewState)};
-        {error, _Error} -> {noreply, State}
+    case do_subscribe(Topic, Subscriber, Options, State) of
+        {ok, NewState}  -> noreply(setstats(NewState));
+        {error, _Error} -> noreply(State)
     end;
 
 handle_cast({unsubscribe, Topic, Subscriber}, State) ->
-    case do_unsubscribe_(Topic, Subscriber, State) of
-        {ok, NewState}  -> {noreply, setstats(NewState), hibernate};
-        {error, _Error} -> {noreply, State}
+    case do_unsubscribe(Topic, Subscriber, State) of
+        {ok, NewState}  -> noreply(setstats(NewState));
+        {error, _Error} -> noreply(State)
     end;
 
-handle_cast({subscriber_down, Subscriber}, State) ->
-    subscriber_down_(Subscriber),
-    {noreply, setstats(State)};
-
 handle_cast(Msg, State) ->
     ?UNEXPECTED_MSG(Msg, State).
 
-handle_info({'DOWN', _MRef, process, DownPid, _Reason}, State = #state{submon = PM}) ->
-    subscriber_down_(DownPid),
-    {noreply, setstats(State#state{submon = PM:erase(DownPid)}), hibernate};
+handle_info({'DOWN', _MRef, process, DownPid, _Reason}, State = #state{subids = SubIds}) ->
+    case maps:find(DownPid, SubIds) of
+        {ok, SubId} ->
+            clean_subscriber({SubId, DownPid});
+        error ->
+            clean_subscriber(DownPid)
+    end,
+    noreply(setstats(demonitor_subscriber(DownPid, State)));
 
 handle_info(Info, State) ->
     ?UNEXPECTED_INFO(Info, State).
@@ -234,62 +255,54 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal Functions
 %%--------------------------------------------------------------------
 
-do_subscribe_(Topic, Subscriber, Options, State) ->
+do_subscribe(Topic, Subscriber, Options, State) ->
     case ets:lookup(mqtt_subproperty, {Topic, Subscriber}) of
         [] ->
             emqttd_pubsub:async_subscribe(Topic, Subscriber, Options),
             Share = proplists:get_value(share, Options),
-            add_subscription_(Share, Subscriber, Topic),
+            add_subscription(Share, Subscriber, Topic),
             ets:insert(mqtt_subproperty, {{Topic, Subscriber}, Options}),
-            {ok, monitor_subpid(Subscriber, State)};
+            {ok, monitor_subscriber(Subscriber, State)};
         [_] ->
             {error, {already_subscribed, Topic}}
     end.
 
-add_subscription_(undefined, Subscriber, Topic) ->
+add_subscription(undefined, Subscriber, Topic) ->
     ets:insert(mqtt_subscription, {Subscriber, Topic});
-add_subscription_(Share, Subscriber, Topic) ->
-    ets:insert(mqtt_subscription, {Subscriber, {Share, Topic}}).
+add_subscription(Share, Subscriber, Topic) ->
+    ets:insert(mqtt_subscription, {Subscriber, {share, Share, Topic}}).
 
-monitor_subpid(SubPid, State = #state{submon = PMon}) when is_pid(SubPid) ->
-    State#state{submon = PMon:monitor(SubPid)};
-monitor_subpid(_SubPid, State) ->
-    State.
+monitor_subscriber(SubPid, State = #state{submon = SubMon}) when is_pid(SubPid) ->
+    State#state{submon = SubMon:monitor(SubPid)};
+monitor_subscriber({SubId, SubPid}, State = #state{subids = SubIds, submon = SubMon}) ->
+    State#state{subids = maps:put(SubPid, SubId, SubIds), submon = SubMon:monitor(SubPid)}.
 
-do_unsubscribe_(Topic, Subscriber, State) ->
+do_unsubscribe(Topic, Subscriber, State) ->
     case ets:lookup(mqtt_subproperty, {Topic, Subscriber}) of
         [{_, Options}] ->
             emqttd_pubsub:async_unsubscribe(Topic, Subscriber, Options),
             Share = proplists:get_value(share, Options),
-            del_subscription_(Share, Subscriber, Topic),
+            del_subscription(Share, Subscriber, Topic),
             ets:delete(mqtt_subproperty, {Topic, Subscriber}),
-            {ok, case ets:member(mqtt_subscription, Subscriber) of
-                true  -> State;
-                false -> demonitor_subpid(Subscriber, State)
-            end};
+            {ok, State};
         [] ->
             {error, {subscription_not_found, Topic}}
     end.
 
-del_subscription_(undefined, Subscriber, Topic) ->
+del_subscription(undefined, Subscriber, Topic) ->
     ets:delete_object(mqtt_subscription, {Subscriber, Topic});
-del_subscription_(Share, Subscriber, Topic) ->
-    ets:delete_object(mqtt_subscription, {Subscriber, {Share, Topic}}).
+del_subscription(Share, Subscriber, Topic) ->
+    ets:delete_object(mqtt_subscription, {Subscriber, {share, Share, Topic}}).
 
-demonitor_subpid(SubPid, State = #state{submon = PMon}) when is_pid(SubPid) ->
-    State#state{submon = PMon:demonitor(SubPid)};
-demonitor_subpid(_SubPid, State) ->
-    State.
-
-subscriber_down_(Subscriber) ->
-    lists:foreach(fun({_, {Share, Topic}}) ->
-                        subscriber_down_(Share, Subscriber, Topic);
+clean_subscriber(Subscriber) ->
+    lists:foreach(fun({_, {share, Share, Topic}}) ->
+                      clean_subscriber(Share, Subscriber, Topic);
                      ({_, Topic}) ->
-                        subscriber_down_(undefined, Subscriber, Topic)
+                      clean_subscriber(undefined, Subscriber, Topic)
         end, ets:lookup(mqtt_subscription, Subscriber)),
     ets:delete(mqtt_subscription, Subscriber).
 
-subscriber_down_(Share, Subscriber, Topic) ->
+clean_subscriber(Share, Subscriber, Topic) ->
     case ets:lookup(mqtt_subproperty, {Topic, Subscriber}) of
         [] ->
             %% TODO:....???
@@ -300,7 +313,16 @@ subscriber_down_(Share, Subscriber, Topic) ->
             ets:delete(mqtt_subproperty, {Topic, Subscriber})
     end.
 
+demonitor_subscriber(SubPid, State = #state{subids = SubIds, submon = SubMon}) ->
+    State#state{subids = maps:remove(SubPid, SubIds), submon = SubMon:demonitor(SubPid)}.
+
 setstats(State) ->
     emqttd_stats:setstats('subscriptions/count', 'subscriptions/max',
                           ets:info(mqtt_subscription, size)), State.
 
+reply(Reply, State) ->
+    {reply, Reply, State, hibernate}.
+
+noreply(State) ->
+    {noreply, State, hibernate}.
+

src/emqttd_session.erl

@@ -172,7 +172,7 @@
                         "Session(~s): " ++ Format, [State#state.client_id | Args])).
 
 %% @doc Start a Session
--spec(start_link(boolean(), {mqtt_client_id(), mqtt_username()}, pid()) -> {ok, pid()} | {error, any()}).
+-spec(start_link(boolean(), {mqtt_client_id(), mqtt_username()}, pid()) -> {ok, pid()} | {error, term()}).
 start_link(CleanSess, {ClientId, Username}, ClientPid) ->
     gen_server2:start_link(?MODULE, [CleanSess, {ClientId, Username}, ClientPid], []).
 
@@ -192,7 +192,7 @@ subscribe(Session, PacketId, TopicTable) -> %%TODO: the ack function??...
     gen_server2:cast(Session, {subscribe, From, TopicTable, AckFun}).
 
 %% @doc Publish Message
--spec(publish(pid(), mqtt_message()) -> ok | {error, any()}).
+-spec(publish(pid(), mqtt_message()) -> ok | {error, term()}).
 publish(_Session, Msg = #mqtt_message{qos = ?QOS_0}) ->
     %% Publish QoS0 Directly
     emqttd_server:publish(Msg), ok;
@@ -389,6 +389,7 @@ handle_cast({subscribe, _From, TopicTable, AckFun},
                         SubMap;
                     {ok, OldQos} ->
                         emqttd:setqos(Topic, ClientId, NewQos),
+                        emqttd_hooks:run('session.subscribed', [ClientId, Username], {Topic, Opts}),
                         ?LOG(warning, "Duplicated subscribe ~s, old_qos=~w, new_qos=~w",
                             [Topic, OldQos, NewQos], State),
                         maps:put(Topic, NewQos, SubMap);
@@ -534,11 +535,11 @@ handle_info({dispatch, Topic, Msg = #mqtt_message{from = {ClientId, _}}},
                             ignore_loop_deliver = IgnoreLoopDeliver}) when is_record(Msg, mqtt_message) ->
     case IgnoreLoopDeliver of
         true  -> {noreply, State, hibernate};
-        false -> {noreply, gc(dispatch(tune_qos(Topic, Msg, State), State)), hibernate}
+        false -> {noreply, handle_dispatch(Topic, Msg, State), hibernate}
     end;
 %% Dispatch Message
 handle_info({dispatch, Topic, Msg}, State) when is_record(Msg, mqtt_message) ->
-    {noreply, gc(dispatch(tune_qos(Topic, Msg, State), State)), hibernate};
+    {noreply, handle_dispatch(Topic, Msg, State), hibernate};
 
 %% Do nothing if the client has been disconnected.
 handle_info({timeout, _Timer, retry_delivery}, State = #state{client_pid = undefined}) ->
@@ -581,9 +582,9 @@ handle_info(Info, Session) ->
     ?UNEXPECTED_INFO(Info, Session).
 
 terminate(Reason, #state{client_id = ClientId, username = Username}) ->
-    emqttd_stats:del_session_stats(ClientId),
+    %% Move to emqttd_sm to avoid race condition
+    %%emqttd_stats:del_session_stats(ClientId),
     emqttd_hooks:run('session.terminated', [ClientId, Username, Reason]),
-    emqttd_server:subscriber_down(ClientId),
     emqttd_sm:unregister_session(ClientId).
 
 code_change(_OldVsn, Session, _Extra) ->
@@ -686,6 +687,9 @@ is_awaiting_full(#state{awaiting_rel = AwaitingRel, max_awaiting_rel = MaxLen})
 %% Dispatch Messages
 %%--------------------------------------------------------------------
 
+handle_dispatch(Topic, Msg, State) ->
+    gc(dispatch(tune_qos(Topic, reset_dup(Msg), State), State)).
+
 %% Enqueue message if the client has been disconnected
 dispatch(Msg, State = #state{client_pid = undefined}) ->
     enqueue_msg(Msg, State);
@@ -800,6 +804,14 @@ tune_qos(Topic, Msg = #mqtt_message{qos = PubQoS},
             Msg
     end.
 
+%%--------------------------------------------------------------------
+%% Reset Dup
+%%--------------------------------------------------------------------
+
+reset_dup(Msg = #mqtt_message{dup = true}) ->
+    Msg#mqtt_message{dup = false};
+reset_dup(Msg) -> Msg.
+
 %%--------------------------------------------------------------------
 %% Next Msg Id
 %%--------------------------------------------------------------------

src/emqttd_sm.erl

@@ -76,12 +76,12 @@ mnesia(copy) ->
 %%--------------------------------------------------------------------
 
 %% @doc Start a session manager
--spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, any()}).
+-spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, term()}).
 start_link(Pool, Id) ->
     gen_server2:start_link({local, ?PROC_NAME(?MODULE, Id)}, ?MODULE, [Pool, Id], []).
 
 %% @doc Start a session
--spec(start_session(boolean(), {binary(), binary() | undefined}) -> {ok, pid(), boolean()} | {error, any()}).
+-spec(start_session(boolean(), {binary(), binary() | undefined}) -> {ok, pid(), boolean()} | {error, term()}).
 start_session(CleanSess, {ClientId, Username}) ->
     SM = gproc_pool:pick_worker(?POOL, ClientId),
     call(SM, {start_session, CleanSess, {ClientId, Username}, self()}).
@@ -107,6 +107,7 @@ unregister_session(ClientId) ->
 unregister_session(ClientId, Pid) ->
     case ets:lookup(mqtt_local_session, ClientId) of
         [LocalSess = {_, Pid, _, _}] ->
+            emqttd_stats:del_session_stats(ClientId),
             ets:delete_object(mqtt_local_session, LocalSess);
         _ ->
             false
@@ -187,7 +188,6 @@ handle_info({'DOWN', MRef, process, DownPid, _Reason}, State) ->
                     [] ->
                         ok;
                     [Sess = #mqtt_session{sess_pid = DownPid}] ->
-                        emqttd_stats:del_session_stats(ClientId),
                         mnesia:delete_object(mqtt_session, Sess, write);
                     [_Sess] ->
                         ok

src/emqttd_sm_helper.erl

@@ -39,7 +39,7 @@
 -define(LOCK, {?MODULE, clean_sessions}).
 
 %% @doc Start a session helper
--spec(start_link(fun()) -> {ok, pid()} | ignore | {error, any()}).
+-spec(start_link(fun()) -> {ok, pid()} | ignore | {error, term()}).
 start_link(StatsFun) ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, [StatsFun], []).
 

src/emqttd_trace.erl

@@ -47,7 +47,7 @@ start_link() ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
 %% @doc Start to trace client or topic.
--spec(start_trace(trace_who(), string()) -> ok | {error, any()}).
+-spec(start_trace(trace_who(), string()) -> ok | {error, term()}).
 start_trace({client, ClientId}, LogFile) ->
     start_trace({start_trace, {client, ClientId}, LogFile});
 
@@ -57,7 +57,7 @@ start_trace({topic, Topic}, LogFile) ->
 start_trace(Req) -> gen_server:call(?MODULE, Req, infinity).
 
 %% @doc Stop tracing client or topic.
--spec(stop_trace(trace_who()) -> ok | {error, any()}).
+-spec(stop_trace(trace_who()) -> ok | {error, term()}).
 stop_trace({client, ClientId}) ->
     gen_server:call(?MODULE, {stop_trace, {client, ClientId}});
 stop_trace({topic, Topic}) ->

src/emqttd_trie.erl

@@ -31,7 +31,7 @@
 -copy_mnesia({mnesia, [copy]}).
 
 %% Trie API
--export([insert/1, match/1, delete/1, lookup/1]).
+-export([insert/1, match/1, lookup/1, delete/1]).
 
 %%--------------------------------------------------------------------
 %% Mnesia Callbacks
@@ -65,22 +65,22 @@ mnesia(copy) ->
 -spec(insert(Topic :: binary()) -> ok).
 insert(Topic) when is_binary(Topic) ->
     case mnesia:read(mqtt_trie_node, Topic) of
-    [#trie_node{topic=Topic}] ->
-        ok;
-    [TrieNode=#trie_node{topic=undefined}] ->
-        write_trie_node(TrieNode#trie_node{topic=Topic});
-    [] ->
-        % Add trie path
-        lists:foreach(fun add_path/1, emqttd_topic:triples(Topic)),
-        % Add last node
-        write_trie_node(#trie_node{node_id=Topic, topic=Topic})
+        [#trie_node{topic = Topic}] ->
+            ok;
+        [TrieNode = #trie_node{topic = undefined}] ->
+            write_trie_node(TrieNode#trie_node{topic = Topic});
+        [] ->
+            % Add trie path
+            lists:foreach(fun add_path/1, emqttd_topic:triples(Topic)),
+            % Add last node
+            write_trie_node(#trie_node{node_id = Topic, topic = Topic})
     end.
 
 %% @doc Find trie nodes that match topic
 -spec(match(Topic :: binary()) -> list(MatchedTopic :: binary())).
 match(Topic) when is_binary(Topic) ->
     TrieNodes = match_node(root, emqttd_topic:words(Topic)),
-    [Name || #trie_node{topic=Name} <- TrieNodes, Name =/= undefined].
+    [Name || #trie_node{topic = Name} <- TrieNodes, Name =/= undefined].
 
 %% @doc Lookup a Trie Node
 -spec(lookup(NodeId :: binary()) -> [#trie_node{}]).
@@ -91,13 +91,13 @@ lookup(NodeId) ->
 -spec(delete(Topic :: binary()) -> ok).
 delete(Topic) when is_binary(Topic) ->
     case mnesia:read(mqtt_trie_node, Topic) of
-    [#trie_node{edge_count=0}] ->
-        mnesia:delete({mqtt_trie_node, Topic}),
-        delete_path(lists:reverse(emqttd_topic:triples(Topic)));
-    [TrieNode] ->
-        write_trie_node(TrieNode#trie_node{topic = undefined});
-    [] ->
-        ok
+        [#trie_node{edge_count = 0}] ->
+            mnesia:delete({mqtt_trie_node, Topic}),
+            delete_path(lists:reverse(emqttd_topic:triples(Topic)));
+        [TrieNode] ->
+            write_trie_node(TrieNode#trie_node{topic = undefined});
+        [] ->
+            ok
     end.
 
 %%--------------------------------------------------------------------
@@ -107,19 +107,19 @@ delete(Topic) when is_binary(Topic) ->
 %% @private
 %% @doc Add path to trie tree.
 add_path({Node, Word, Child}) ->
-    Edge = #trie_edge{node_id=Node, word=Word},
+    Edge = #trie_edge{node_id = Node, word = Word},
     case mnesia:read(mqtt_trie_node, Node) of
-    [TrieNode = #trie_node{edge_count=Count}] ->
-        case mnesia:wread({mqtt_trie, Edge}) of
-        [] -> 
-            write_trie_node(TrieNode#trie_node{edge_count=Count+1}),
-            write_trie(#trie{edge=Edge, node_id=Child});
-        [_] -> 
-            ok
-        end;
-    [] ->
-        write_trie_node(#trie_node{node_id=Node, edge_count=1}),
-        write_trie(#trie{edge=Edge, node_id=Child})
+        [TrieNode = #trie_node{edge_count = Count}] ->
+            case mnesia:wread({mqtt_trie, Edge}) of
+                [] ->
+                    write_trie_node(TrieNode#trie_node{edge_count = Count+1}),
+                    write_trie(#trie{edge = Edge, node_id = Child});
+                [_] ->
+                    ok
+            end;
+        [] ->
+            write_trie_node(#trie_node{node_id = Node, edge_count = 1}),
+            write_trie(#trie{edge = Edge, node_id = Child})
     end.
 
 %% @private
@@ -135,20 +135,20 @@ match_node(NodeId, [], ResAcc) ->
 
 match_node(NodeId, [W|Words], ResAcc) ->
     lists:foldl(fun(WArg, Acc) ->
-        case mnesia:read(mqtt_trie, #trie_edge{node_id=NodeId, word=WArg}) of
-        [#trie{node_id=ChildId}] -> match_node(ChildId, Words, Acc);
-        [] -> Acc
+        case mnesia:read(mqtt_trie, #trie_edge{node_id = NodeId, word = WArg}) of
+            [#trie{node_id = ChildId}] -> match_node(ChildId, Words, Acc);
+            [] -> Acc
         end
     end, 'match_#'(NodeId, ResAcc), [W, '+']).
 
 %% @private
 %% @doc Match node with '#'.
 'match_#'(NodeId, ResAcc) ->
-    case mnesia:read(mqtt_trie, #trie_edge{node_id=NodeId, word = '#'}) of
-    [#trie{node_id=ChildId}] ->
-        mnesia:read(mqtt_trie_node, ChildId) ++ ResAcc;
-    [] ->
-        ResAcc
+    case mnesia:read(mqtt_trie, #trie_edge{node_id = NodeId, word = '#'}) of
+        [#trie{node_id = ChildId}] ->
+            mnesia:read(mqtt_trie_node, ChildId) ++ ResAcc;
+        [] ->
+            ResAcc
     end.
 
 %% @private
@@ -156,17 +156,17 @@ match_node(NodeId, [W|Words], ResAcc) ->
 delete_path([]) ->
     ok;
 delete_path([{NodeId, Word, _} | RestPath]) ->
-    mnesia:delete({mqtt_trie, #trie_edge{node_id=NodeId, word=Word}}),
+    mnesia:delete({mqtt_trie, #trie_edge{node_id = NodeId, word = Word}}),
     case mnesia:read(mqtt_trie_node, NodeId) of
-    [#trie_node{edge_count=1, topic=undefined}] -> 
-        mnesia:delete({mqtt_trie_node, NodeId}),
-        delete_path(RestPath);
-    [TrieNode=#trie_node{edge_count=1, topic=_}] -> 
-        write_trie_node(TrieNode#trie_node{edge_count=0});
-    [TrieNode=#trie_node{edge_count=C}] ->
-        write_trie_node(TrieNode#trie_node{edge_count=C-1});
-    [] ->
-        throw({notfound, NodeId}) 
+        [#trie_node{edge_count = 1, topic = undefined}] ->
+            mnesia:delete({mqtt_trie_node, NodeId}),
+            delete_path(RestPath);
+        [TrieNode = #trie_node{edge_count = 1, topic = _}] ->
+            write_trie_node(TrieNode#trie_node{edge_count = 0});
+        [TrieNode = #trie_node{edge_count = C}] ->
+            write_trie_node(TrieNode#trie_node{edge_count = C-1});
+        [] ->
+            mnesia:abort({node_not_found, NodeId})
     end.
 
 %% @private

test/emqttd_SUITE.erl

@@ -67,7 +67,6 @@ all() ->
      {group, http},
      {group, alarms},
      {group, cli},
-     {group, get_api},
      {group, cleanSession}].
 
 groups() ->
@@ -99,7 +98,8 @@ groups() ->
        run_hooks]},
     {http, [sequence], 
      [request_status,
-      request_publish
+      request_publish,
+      get_api_lists
      % websocket_test
      ]},
      {alarms, [sequence], 
@@ -122,8 +122,7 @@ groups() ->
          ]},
        cli_vm]},
     {cleanSession, [sequence],
-      [cleanSession_validate]},
-    {get_api, [sequence], [get_api_lists]}].
+      [cleanSession_validate]}].
 
 init_per_suite(Config) ->
     NewConfig = generate_config(),
@@ -235,13 +234,15 @@ pubsub(_) ->
     emqttd:unsubscribe(<<"a/b/c">>).
 
 t_local_subscribe(_) ->
-    emqttd:subscribe("$local/topic0"),
-    emqttd:subscribe("$local/topic1", <<"x">>),
-    emqttd:subscribe("$local/topic2", <<"x">>, [{qos, 2}]),
+    ok = emqttd:subscribe("$local/topic0"),
+    ok = emqttd:subscribe("$local/topic1", <<"x">>),
+    ok = emqttd:subscribe("$local/topic2", <<"x">>, [{qos, 2}]),
     timer:sleep(10),
     ?assertEqual([self()], emqttd:subscribers("$local/topic0")),
-    ?assertEqual([<<"x">>], emqttd:subscribers("$local/topic1")),
-    ?assertEqual([{<<"$local/topic1">>,<<"x">>,[]},{<<"$local/topic2">>,<<"x">>,[{qos,2}]}], emqttd:subscriptions(<<"x">>)),
+    ?assertEqual([{<<"x">>, self()}], emqttd:subscribers("$local/topic1")),
+    ?assertEqual([{{<<"x">>, self()}, <<"$local/topic1">>, []},
+                  {{<<"x">>, self()}, <<"$local/topic2">>, [{qos,2}]}],
+                 emqttd:subscriptions(<<"x">>)),
     
     ?assertEqual(ok, emqttd:unsubscribe("$local/topic0")),
     ?assertMatch({error, {subscription_not_found, _}}, emqttd:unsubscribe("$local/topic0")),
@@ -256,9 +257,9 @@ t_shared_subscribe(_) ->
     emqttd:subscribe("$queue/topic3"),
     timer:sleep(10),
     ?assertEqual([self()], emqttd:subscribers(<<"$local/$share/group1/topic1">>)),
-    ?assertEqual([{<<"$local/$share/group1/topic1">>, self(), []},
-                  {<<"$queue/topic3">>, self(), []},
-                  {<<"$share/group2/topic2">>, self(), []}],
+    ?assertEqual([{self(), <<"$local/$share/group1/topic1">>, []},
+                  {self(), <<"$queue/topic3">>, []},
+                  {self(), <<"$share/group2/topic2">>, []}],
                  lists:sort(emqttd:subscriptions(self()))),
     emqttd:unsubscribe("$local/$share/group1/topic1"),
     emqttd:unsubscribe("$share/group2/topic2"),
@@ -298,7 +299,7 @@ router_add_del(_) ->
     %% Add
     emqttd_router:add_route(<<"#">>),
     emqttd_router:add_route(<<"a/b/c">>),
-    emqttd_router:add_route(<<"+/#">>, node()),
+    emqttd_router:add_route(<<"+/#">>),
     Routes = [R1, R2 | _] = [
             #mqtt_route{topic = <<"#">>,     node = node()},
             #mqtt_route{topic = <<"+/#">>,   node = node()},
@@ -306,7 +307,7 @@ router_add_del(_) ->
     Routes = lists:sort(emqttd_router:match(<<"a/b/c">>)),
 
     %% Batch Add
-    emqttd_router:add_routes(Routes),
+    lists:foreach(fun(R) -> emqttd_router:add_route(R) end, Routes),
     Routes = lists:sort(emqttd_router:match(<<"a/b/c">>)),
 
     %% Del
@@ -317,7 +318,8 @@ router_add_del(_) ->
     %% Batch Del
     R3 = #mqtt_route{topic = <<"#">>, node = 'a@127.0.0.1'},
     emqttd_router:add_route(R3),
-    emqttd_router:del_routes([R1, R2]),
+    emqttd_router:del_route(R1),
+    emqttd_router:del_route(R2),
     emqttd_router:del_route(R3),
     [] = lists:sort(emqttd_router:match(<<"a/b/c">>)).
 
@@ -325,7 +327,7 @@ router_print(_) ->
     Routes = [#mqtt_route{topic = <<"a/b/c">>, node = node()},
               #mqtt_route{topic = <<"#">>,     node = node()},
               #mqtt_route{topic = <<"+/#">>,   node = node()}],
-    emqttd_router:add_routes(Routes),
+    lists:foreach(fun(R) -> emqttd_router:add_route(R) end, Routes),
     emqttd_router:print(<<"a/b/c">>).
 
 router_unused(_) ->
@@ -448,19 +450,20 @@ request_status(_) ->
     ?assertEqual(binary_to_list(Status), Return).
 
 request_publish(_) ->
+    application:start(emq_dashboard),
     emqttc:start_link([{host, "localhost"},
                        {port, 1883},
                        {client_id, <<"random">>},
                        {clean_sess, false}]),
     SubParams = "{\"qos\":1, \"topic\" : \"a\/b\/c\", \"client_id\" :\"random\"}",
-    ?assert(connect_emqttd_pubsub_(post, "api/v2/mqtt/subscribe", SubParams, auth_header_("", ""))),
+    ?assert(connect_emqttd_pubsub_(post, "api/v2/mqtt/subscribe", SubParams, auth_header_("admin", "public"))),
     ok = emqttd:subscribe(<<"a/b/c">>, self(), [{qos, 1}]),
     Params = "{\"qos\":1, \"retain\":false, \"topic\" : \"a\/b\/c\", \"messages\" :\"hello\"}",
-    ?assert(connect_emqttd_pubsub_(post, "api/v2/mqtt/publish", Params, auth_header_("", ""))),
+    ?assert(connect_emqttd_pubsub_(post, "api/v2/mqtt/publish", Params, auth_header_("admin", "public"))),
     ?assert(receive {dispatch, <<"a/b/c">>, _} -> true after 2 -> false end),
 
     UnSubParams = "{\"topic\" : \"a\/b\/c\", \"client_id\" :\"random\"}",
-    ?assert(connect_emqttd_pubsub_(post, "api/v2/mqtt/unsubscribe", UnSubParams, auth_header_("", ""))).
+    ?assert(connect_emqttd_pubsub_(post, "api/v2/mqtt/unsubscribe", UnSubParams, auth_header_("admin", "public"))).
 
 connect_emqttd_pubsub_(Method, Api, Params, Auth) ->
     Url = "http://127.0.0.1:8080/" ++ Api,
@@ -479,6 +482,9 @@ auth_header_(User, Pass) ->
     Encoded = base64:encode_to_string(lists:append([User,":",Pass])),
     {"Authorization","Basic " ++ Encoded}.
 
+get_api_lists(_Config) ->
+    lists:foreach(fun request/1, ?GET_API).
+
 websocket_test(_) ->
     Conn = esockd_connection:new(esockd_transport, nil, []),
     Req = mochiweb_request:new(Conn, 'GET', "/mqtt", {1, 1},
@@ -589,9 +595,9 @@ conflict_listeners(_) ->
                {current_clients, esockd:get_current_clients(Pid)},
                {shutdown_count, esockd:get_shutdown_count(Pid)}]}
               end, esockd:listeners()),
-    L =proplists:get_value("mqtt:tcp:0.0.0.0:1883", Listeners),
+    L = proplists:get_value("mqtt:tcp:0.0.0.0:1883", Listeners),
     ?assertEqual(1, proplists:get_value(current_clients, L)),
-    ?assertEqual(1, proplists:get_value(conflict, L)),
+    ?assertEqual(1, proplists:get_value(conflict, proplists:get_value(shutdown_count, L))),
     emqttc:disconnect(C2).
 
 cli_vm(_) ->
@@ -623,9 +629,6 @@ cleanSession_validate(_) ->
     emqttc:disconnect(Pub),
     emqttc:disconnect(C11).
 
-get_api_lists(_Config) ->
-    lists:foreach(fun request/1, ?GET_API).
-
 change_opts(SslType) ->
     {ok, Listeners} = application:get_env(?APP, listeners),
     NewListeners =
@@ -694,7 +697,7 @@ http_post(Method, Path, Params) ->
 
 req(Method, Path, Body) ->
    Url = ?URL ++ Path,
-   Headers = auth_header_("", ""),
+   Headers = auth_header_("admin", "public"),
    case httpc:request(Method, {Url, [Headers]}, [], []) of
    {error, socket_closed_remotely} ->
        false;

test/emqttd_cli_SUITE.erl

@@ -0,0 +1,52 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2017 EMQ Enterprise, Inc. (http://emqtt.io)
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqttd_cli_SUITE).
+
+-compile(export_all).
+
+-include("emqttd.hrl").
+
+-include_lib("eunit/include/eunit.hrl").
+
+all() ->
+    [{group, subscriptions}].
+
+groups() ->
+    [{subscriptions, [sequence],
+      [t_subsciptions_list,
+       t_subsciptions_show,
+       t_subsciptions_add,
+       t_subsciptions_del]}].
+
+init_per_suite(Config) ->
+    Config.
+
+end_per_suite(_Config) ->
+    todo.
+
+t_subsciptions_list(_) ->
+    todo.
+
+t_subsciptions_show(_) ->
+    todo.
+
+t_subsciptions_add(_) ->
+    todo.
+
+t_subsciptions_del(_) ->
+    todo.
+

test/emqttd_router_SUITE.erl

@@ -0,0 +1,132 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2013-2017 EMQ Enterprise, Inc. (http://emqtt.io)
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqttd_router_SUITE).
+
+-compile(export_all).
+
+-include("emqttd.hrl").
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(R, emqttd_router).
+
+all() ->
+    [{group, route},
+     {group, local_route}].
+
+groups() ->
+    [{route, [sequence],
+      [t_get_topics,
+       t_add_del_route,
+       t_match_route,
+       t_print,
+       t_has_route]},
+     {local_route, [sequence],
+      [t_get_local_topics,
+       t_add_del_local_route,
+       t_match_local_route]}].
+
+init_per_suite(Config) ->
+    ekka:start(),
+    ekka_mnesia:ensure_started(),
+    {ok, _R} = emqttd_router:start(),
+    Config.
+
+end_per_suite(_Config) ->
+    emqttd_router:stop(),
+    ekka:stop(),
+    ekka_mnesia:ensure_stopped(),
+    ekka_mnesia:delete_schema().
+
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(_TestCase, _Config) ->
+    clear_tables().
+
+t_get_topics(_) ->
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/+/b">>),
+    ?assertEqual([<<"a/+/b">>, <<"a/b/c">>], lists:sort(?R:topics())),
+    ?R:del_route(<<"a/b/c">>),
+    ?R:del_route(<<"a/+/b">>),
+    ?assertEqual([], lists:sort(?R:topics())).
+
+t_add_del_route(_) ->
+    %%Node = node(),
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/+/b">>),
+    ?R:del_route(<<"a/b/c">>),
+    ?R:del_route(<<"a/+/b">>).
+
+t_match_route(_) ->
+    Node = node(),
+    ?R:add_route(<<"a/b/c">>),
+    ?R:add_route(<<"a/+/c">>),
+    ?R:add_route(<<"a/b/#">>),
+    ?R:add_route(<<"#">>),
+    ?assertEqual([#mqtt_route{topic = <<"#">>, node = Node},
+                  #mqtt_route{topic = <<"a/+/c">>, node = Node},
+                  #mqtt_route{topic = <<"a/b/#">>, node = Node},
+                  #mqtt_route{topic = <<"a/b/c">>, node = Node}],
+                 lists:sort(?R:match(<<"a/b/c">>))).
+
+t_print(_) ->
+    ?R:add_route(<<"topic">>),
+    ?R:add_route(<<"topic/#">>),
+    ?R:print(<<"topic">>).
+
+t_has_route(_) ->
+    ?R:add_route(<<"devices/+/messages">>),
+    ?assert(?R:has_route(<<"devices/+/messages">>)).
+
+t_get_local_topics(_) ->
+    ?R:add_local_route(<<"a/b/c">>),
+    ?R:add_local_route(<<"x/+/y">>),
+    ?R:add_local_route(<<"z/#">>),
+    ?assertEqual([<<"z/#">>, <<"x/+/y">>, <<"a/b/c">>], ?R:local_topics()),
+    ?R:del_local_route(<<"x/+/y">>),
+    ?R:del_local_route(<<"z/#">>),
+    ?assertEqual([<<"a/b/c">>], ?R:local_topics()).
+
+t_add_del_local_route(_) ->
+    Node = node(),
+    ?R:add_local_route(<<"a/b/c">>),
+    ?R:add_local_route(<<"x/+/y">>),
+    ?R:add_local_route(<<"z/#">>),
+    ?assertEqual([{<<"a/b/c">>, Node},
+                  {<<"x/+/y">>, Node},
+                  {<<"z/#">>, Node}],
+                 lists:sort(?R:get_local_routes())),
+    ?R:del_local_route(<<"x/+/y">>),
+    ?R:del_local_route(<<"z/#">>),
+    ?assertEqual([{<<"a/b/c">>, Node}], lists:sort(?R:get_local_routes())).
+
+t_match_local_route(_) ->
+    ?R:add_local_route(<<"$SYS/#">>),
+    ?R:add_local_route(<<"a/b/c">>),
+    ?R:add_local_route(<<"a/+/c">>),
+    ?R:add_local_route(<<"a/b/#">>),
+    ?R:add_local_route(<<"#">>),
+    Matched = [Topic || #mqtt_route{topic = {local, Topic}} <- ?R:match_local(<<"a/b/c">>)],
+    ?assertEqual([<<"#">>, <<"a/+/c">>, <<"a/b/#">>, <<"a/b/c">>], lists:sort(Matched)).
+
+clear_tables() ->
+    ?R:clean_local_routes(),
+    lists:foreach(fun mnesia:clear_table/1, [mqtt_route, mqtt_trie, mqtt_trie_node]).
+