diff --git a/Makefile b/Makefile index 2826d547d..930ad76a0 100644 --- a/Makefile +++ b/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 diff --git a/etc/certs/cacert.pem b/etc/certs/cacert.pem index ca4948ed9..595baf450 100644 --- a/etc/certs/cacert.pem +++ b/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----- diff --git a/etc/certs/cert.pem b/etc/certs/cert.pem index 58aa2c4ef..fb76ecde6 100644 --- a/etc/certs/cert.pem +++ b/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----- diff --git a/etc/certs/client-cert.pem b/etc/certs/client-cert.pem index e1690d9aa..1c44b3716 100644 --- a/etc/certs/client-cert.pem +++ b/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----- diff --git a/etc/certs/client-key.pem b/etc/certs/client-key.pem index 74cf487cb..008f78481 100644 --- a/etc/certs/client-key.pem +++ b/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----- diff --git a/etc/certs/key.pem b/etc/certs/key.pem index 7001093ef..739f39880 100644 --- a/etc/certs/key.pem +++ b/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----- diff --git a/etc/emq.conf b/etc/emq.conf index 227a22457..511292d30 100644 --- a/etc/emq.conf +++ b/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 diff --git a/include/emqttd_trie.hrl b/include/emqttd_trie.hrl index c5b9d03c7..eb4e1390d 100644 --- a/include/emqttd_trie.hrl +++ b/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, diff --git a/priv/emq.schema b/priv/emq.schema index 6b3f0001f..d05cc79cf 100644 --- a/priv/emq.schema +++ b/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}. diff --git a/src/emqttd.app.src b/src/emqttd.app.src index 0b85fe0f0..269601bb8 100644 --- a/src/emqttd.app.src +++ b/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, diff --git a/src/emqttd.erl b/src/emqttd.erl index d4cdd8437..65739952f 100644 --- a/src/emqttd.erl +++ b/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(subscriber() :: pid() | subid() | {subid(), pid()}). -type(suboption() :: local | {qos, non_neg_integer()} | {share, {'$queue' | binary()}}). --type(pubsub_error() :: {error, {already_subscribed, binary()} - | {subscription_not_found, 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). diff --git a/src/emqttd_access_control.erl b/src/emqttd_access_control.erl index 6cfcc03cf..601fd263f 100644 --- a/src/emqttd_access_control.erl +++ b/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}). diff --git a/src/emqttd_acl_mod.erl b/src/emqttd_acl_mod.erl index 12e949afe..4ed07b369 100644 --- a/src/emqttd_acl_mod.erl +++ b/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()). diff --git a/src/emqttd_bridge_sup.erl b/src/emqttd_bridge_sup.erl index d28b2274c..75138332f 100644 --- a/src/emqttd_bridge_sup.erl +++ b/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). diff --git a/src/emqttd_bridge_sup_sup.erl b/src/emqttd_bridge_sup_sup.erl index 1d47bd003..11679aba8 100644 --- a/src/emqttd_bridge_sup_sup.erl +++ b/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) -> diff --git a/src/emqttd_broker.erl b/src/emqttd_broker.erl index 9f939a45e..0161720f2 100644 --- a/src/emqttd_broker.erl +++ b/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, [], []). diff --git a/src/emqttd_cli.erl b/src/emqttd_cli.erl index a4029d46f..280c050ee 100644 --- a/src/emqttd_cli.erl +++ b/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 ", "Show subscriptions of a client"}, {"subscriptions add ", "Add a static subscription manually"}, - {"subscriptions del ", "Delete static subscriptions manually"}, {"subscriptions del ", "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); diff --git a/src/emqttd_cm.erl b/src/emqttd_cm.erl index 2e57ebe5b..c5f0ff145 100644 --- a/src/emqttd_cm.erl +++ b/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], []). diff --git a/src/emqttd_gen_mod.erl b/src/emqttd_gen_mod.erl index 824b65f4c..f8d690024 100644 --- a/src/emqttd_gen_mod.erl +++ b/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. diff --git a/src/emqttd_keepalive.erl b/src/emqttd_keepalive.erl index 5d26ea8a6..a0458038a 100644 --- a/src/emqttd_keepalive.erl +++ b/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} -> diff --git a/src/emqttd_metrics.erl b/src/emqttd_metrics.erl index c21d274a0..17e6e96d4 100644 --- a/src/emqttd_metrics.erl +++ b/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, [], []). diff --git a/src/emqttd_mgmt.erl b/src/emqttd_mgmt.erl index 2c671f3bd..01dd50b1e 100644 --- a/src/emqttd_mgmt.erl +++ b/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}]). diff --git a/src/emqttd_mod_sup.erl b/src/emqttd_mod_sup.erl index 586d5af1a..749b84a42 100644 --- a/src/emqttd_mod_sup.erl +++ b/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); diff --git a/src/emqttd_parser.erl b/src/emqttd_parser.erl index dde9ae4dc..4699b3f77 100644 --- a/src/emqttd_parser.erl +++ b/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(<>, {none, Limit}) -> @@ -124,7 +124,7 @@ parse_frame(Bin, #mqtt_packet_header{type = Type, qos = Qos} = Header, Length) _ -> <> = 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, <>} -> @@ -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. diff --git a/src/emqttd_plugins.erl b/src/emqttd_plugins.erl index 73f184a3a..81ff61a4d 100644 --- a/src/emqttd_plugins.erl +++ b/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} -> diff --git a/src/emqttd_pool_sup.erl b/src/emqttd_pool_sup.erl index fa1d6aace..87654bcff 100644 --- a/src/emqttd_pool_sup.erl +++ b/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]). diff --git a/src/emqttd_pooler.erl b/src/emqttd_pooler.erl index 1b73d138c..a74e01fec 100644 --- a/src/emqttd_pooler.erl +++ b/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], []). diff --git a/src/emqttd_protocol.erl b/src/emqttd_protocol.erl index 31354dd84..35c914b14 100644 --- a/src/emqttd_protocol.erl +++ b/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), diff --git a/src/emqttd_pubsub.erl b/src/emqttd_pubsub.erl index d976618cd..994ef6230 100644 --- a/src/emqttd_pubsub.erl +++ b/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}. + diff --git a/src/emqttd_rest_api.erl b/src/emqttd_rest_api.erl index 85981ea45..0eb6adc11 100644 --- a/src/emqttd_rest_api.erl +++ b/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}]. %%-------------------------------------------------------------------------- diff --git a/src/emqttd_router.erl b/src/emqttd_router.erl index b3dd8b4ad..4d4e22160 100644 --- a/src/emqttd_router.erl +++ b/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}. diff --git a/src/emqttd_server.erl b/src/emqttd_server.erl index 69d18e1e4..4e05c00aa 100644 --- a/src/emqttd_server.erl +++ b/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}). + call(pick(Subscriber), {setqos, Topic, with_subpid(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)). +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}. -subscription(Topic, Subscriber) -> - {Topic, Subscriber, ets:lookup_element(mqtt_subproperty, {Topic, Subscriber}, 2)}. +-spec(subscriptions(emqttd:subscriber()) -> [{emqttd:subscriber(), binary(), list(emqttd:suboption())}]). +subscriptions(SubPid) when is_pid(SubPid) -> + with_subproperty(ets:lookup(mqtt_subscription, SubPid)); -subscribers(Topic) -> +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}. + diff --git a/src/emqttd_session.erl b/src/emqttd_session.erl index e8e694530..aa08d746c 100644 --- a/src/emqttd_session.erl +++ b/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 %%-------------------------------------------------------------------- diff --git a/src/emqttd_sm.erl b/src/emqttd_sm.erl index dacb0b6a1..8d4c220b2 100644 --- a/src/emqttd_sm.erl +++ b/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 diff --git a/src/emqttd_sm_helper.erl b/src/emqttd_sm_helper.erl index 2e8e9a749..0721339fd 100644 --- a/src/emqttd_sm_helper.erl +++ b/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], []). diff --git a/src/emqttd_trace.erl b/src/emqttd_trace.erl index eda84bcb1..05734c2d2 100644 --- a/src/emqttd_trace.erl +++ b/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}) -> diff --git a/src/emqttd_trie.erl b/src/emqttd_trie.erl index 5b36e6e04..0bb6ec63e 100644 --- a/src/emqttd_trie.erl +++ b/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 diff --git a/test/emqttd_SUITE.erl b/test/emqttd_SUITE.erl index 195820248..fe48fb402 100644 --- a/test/emqttd_SUITE.erl +++ b/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; diff --git a/test/emqttd_cli_SUITE.erl b/test/emqttd_cli_SUITE.erl new file mode 100644 index 000000000..273518b7f --- /dev/null +++ b/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. + diff --git a/test/emqttd_router_SUITE.erl b/test/emqttd_router_SUITE.erl new file mode 100644 index 000000000..addd36288 --- /dev/null +++ b/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]). +