diff --git a/.gitignore b/.gitignore index d3fc9149b..1259471d2 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,4 @@ _build .rebar3 rebar3.crashdump .DS_Store +rebar.config diff --git a/.travis.yml b/.travis.yml index 6c41760ee..c625470d0 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,7 +4,7 @@ otp_release: - 20.0 - 20.1 -script: +script: - make sudo: false diff --git a/Makefile b/Makefile index 21eacc064..d9d04c517 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ PROJECT = emqx PROJECT_DESCRIPTION = EMQ X Broker -PROJECT_VERSION = 2.4 +PROJECT_VERSION = 2.3.0 NO_AUTOPATCH = gen_rpc cuttlefish @@ -8,11 +8,11 @@ DEPS = goldrush gproc gen_rpc lager esockd ekka mochiweb pbkdf2 lager_syslog bcr dep_goldrush = git https://github.com/basho/goldrush 0.1.9 dep_gproc = git https://github.com/uwiger/gproc -dep_gen_rpc = git https://github.com/priestjim/gen_rpc +dep_gen_rpc = git https://github.com/priestjim/gen_rpc dep_getopt = git https://github.com/jcomellas/getopt v0.8.2 dep_lager = git https://github.com/basho/lager master dep_lager_syslog = git https://github.com/basho/lager_syslog -dep_jsx = git https://github.com/talentdeficit/jsx +dep_jsx = git https://github.com/talentdeficit/jsx dep_esockd = git https://github.com/emqtt/esockd master dep_ekka = git https://github.com/emqtt/ekka master dep_mochiweb = git https://github.com/emqtt/mochiweb master @@ -26,9 +26,9 @@ ERLC_OPTS += +'{parse_transform, lager_transform}' BUILD_DEPS = cuttlefish dep_cuttlefish = git https://github.com/emqtt/cuttlefish -TEST_DEPS = websocket_client emqttc +TEST_DEPS = emqttc emq_dashboard dep_emqttc = git https://github.com/emqtt/emqttc -dep_websocket_client = git https://github.com/jeremyong/websocket_client +dep_emq_dashboard = git https://github.com/emqtt/emq_dashboard TEST_ERLC_OPTS += +debug_info TEST_ERLC_OPTS += +'{parse_transform, lager_transform}' @@ -36,8 +36,8 @@ TEST_ERLC_OPTS += +'{parse_transform, lager_transform}' EUNIT_OPTS = verbose # EUNIT_ERL_OPTS = -CT_SUITES = emqx emqx_lib emqx_topic emqx_trie emqx_mqueue emqx_inflight \ -emqx_vm emqx_net emqx_protocol emqx_access emqx_config +CT_SUITES = emqx emqx_mod emqx_lib emqx_topic emqx_trie emqx_mqueue emqx_inflight \ + emqx_vm emqx_net emqx_protocol emqx_access emqx_config emqx_router CT_OPTS = -cover test/ct.cover.spec -erl_args -name emqxct@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/emqx.conf b/etc/emqx.conf index e24f3fab5..3cdf21103 100644 --- a/etc/emqx.conf +++ b/etc/emqx.conf @@ -1,5 +1,5 @@ ##=================================================================== -## EMQ X Configuration R2.4 +## EMQ X Configuration R2.3.0 ##=================================================================== ##-------------------------------------------------------------------- @@ -161,26 +161,44 @@ log.console = console ## Console log level. Enum: debug, info, notice, warning, error, critical, alert, emergency log.console.level = error -## Syslog. Enum: on, off -log.syslog = on - -## syslog level. Enum: debug, info, notice, warning, error, critical, alert, emergency -log.syslog.level = error - ## Console log file ## log.console.file = {{ platform_log_dir }}/console.log +## Console log file size +## log.console.size = 10485760 + +## Console log count size +## log.console.count = 5 + ## Info log file ## log.info.file = {{ platform_log_dir }}/info.log +## Info log file size +## log.info.size = 10485760 + +## Info log file count +## log.info.count = 5 + ## Error log file log.error.file = {{ platform_log_dir }}/error.log +## Error log file size +log.error.size = 10485760 + +## Error log file count +log.error.count = 5 + ## Enable the crash log. Enum: on, off log.crash = on log.crash.file = {{ platform_log_dir }}/crash.log +## Syslog. Enum: on, off +log.syslog = on + +## syslog level. Enum: debug, info, notice, warning, error, critical, alert, emergency +log.syslog.level = error + ##-------------------------------------------------------------------- ## Allow Anonymous and Default ACL ##-------------------------------------------------------------------- @@ -303,8 +321,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/emqx_trie.hrl b/include/emqx_trie.hrl index c5b9d03c7..eb4e1390d 100644 --- a/include/emqx_trie.hrl +++ b/include/emqx_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/emqx.schema b/priv/emqx.schema index 82f940a36..508b98121 100644 --- a/priv/emqx.schema +++ b/priv/emqx.schema @@ -384,13 +384,43 @@ end}. {datatype, file} ]}. +{mapping, "log.console.size", "lager.handlers", [ + {default, 10485760}, + {datatype, integer} +]}. + +{mapping, "log.console.count", "lager.handlers", [ + {default, 5}, + {datatype, integer} +]}. + +{mapping, "log.info.file", "lager.handlers", [ + {datatype, file} +]}. + +{mapping, "log.info.size", "lager.handlers", [ + {default, 10485760}, + {datatype, integer} +]}. + +{mapping, "log.info.count", "lager.handlers", [ + {default, 5}, + {datatype, integer} +]}. + {mapping, "log.error.file", "lager.handlers", [ {default, "log/error.log"}, {datatype, file} ]}. -{mapping, "log.info.file", "lager.handlers", [ - {datatype, file} +{mapping, "log.error.size", "lager.handlers", [ + {default, 10485760}, + {datatype, integer} +]}. + +{mapping, "log.error.count", "lager.handlers", [ + {default, 5}, + {datatype, integer} ]}. {mapping, "log.syslog", "lager.handlers", [ @@ -432,9 +462,18 @@ end}. undefined -> []; ErrorFilename -> [{lager_file_backend, [{file, ErrorFilename}, {level, error}, - {size, 10485760}, + {size, cuttlefish:conf_get("log.error.size", Conf)}, {date, "$D0"}, - {count, 5}]}] + {count, cuttlefish:conf_get("log.error.count", Conf)}]}] + end, + + InfoHandler = case cuttlefish:conf_get("log.info.file", Conf, undefined) of + undefined -> []; + InfoFilename -> [{lager_file_backend, [{file, InfoFilename}, + {level, info}, + {size, cuttlefish:conf_get("log.info.size", Conf)}, + {date, "$D0"}, + {count, cuttlefish:conf_get("log.info.count", Conf)}]}] end, InfoHandler = case cuttlefish:conf_get("log.info.file", Conf, undefined) of undefined -> []; @@ -451,9 +490,9 @@ end}. ConsoleHandler = {lager_console_backend, [{level, ConsoleLogLevel}]}, ConsoleFileHandler = {lager_file_backend, [{file, ConsoleLogFile}, {level, ConsoleLogLevel}, - {size, 10485760}, + {size, cuttlefish:conf_get("log.console.size", Conf)}, {date, "$D0"}, - {count, 5}]}, + {count, cuttlefish:conf_get("log.console.count", Conf)}]}, ConsoleHandlers = case cuttlefish:conf_get("log.console", Conf) of off -> []; @@ -469,8 +508,7 @@ end}. cuttlefish:conf_get("log.syslog.facility", Conf), cuttlefish:conf_get("log.syslog.level", Conf)]}] end, - - ConsoleHandlers ++ ErrorHandler ++ InfoHandler ++SyslogHandler + ConsoleHandlers ++ ErrorHandler ++ InfoHandler ++ SyslogHandler end }. @@ -545,7 +583,7 @@ end}. ]}. %% @doc Keepalive backoff -{mapping, "mqtt.keepalive_backoff", "emqttd.protocol", [ +{mapping, "mqtt.keepalive_backoff", "emqx.protocol", [ {default, 1.25}, {datatype, float} ]}. @@ -742,11 +780,6 @@ end}. {datatype, integer} ]}. -{mapping, "mqtt.pubsub.by_clientid", "emqx.pubsub", [ - {default, true}, - {datatype, {enum, [true, false]}} -]}. - {mapping, "mqtt.pubsub.async", "emqx.pubsub", [ {default, true}, {datatype, {enum, [true, false]}} @@ -754,7 +787,6 @@ end}. {translation, "emqx.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.erl b/src/emqttd.erl new file mode 100644 index 000000000..65739952f --- /dev/null +++ b/src/emqttd.erl @@ -0,0 +1,181 @@ +%%-------------------------------------------------------------------- +%% 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. +%%-------------------------------------------------------------------- + +%% @doc EMQ Main Module. + +-module(emqttd). + +-author("Feng Lee "). + +-include("emqttd.hrl"). + +-include("emqttd_protocol.hrl"). + +-export([start/0, env/1, env/2, is_running/1, stop/0]). + +%% PubSub API +-export([subscribe/1, subscribe/2, subscribe/3, publish/1, + unsubscribe/1, unsubscribe/2]). + +%% PubSub Management API +-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]). + +%% Debug API +-export([dump/0]). + +%% Shutdown and reboot +-export([shutdown/0, shutdown/1, reboot/0]). + +-type(subid() :: binary()). + +-type(subscriber() :: pid() | subid() | {subid(), pid()}). + +-type(suboption() :: local | {qos, non_neg_integer()} | {share, {'$queue' | binary()}}). + +-export_type([subscriber/0, suboption/0]). + +-define(APP, ?MODULE). + +%%-------------------------------------------------------------------- +%% Bootstrap, environment, configuration, is_running... +%%-------------------------------------------------------------------- + +%% @doc Start emqttd application. +-spec(start() -> ok | {error, term()}). +start() -> application:start(?APP). + +%% @doc Stop emqttd application. +-spec(stop() -> ok | {error, term()}). +stop() -> application:stop(?APP). + +%% @doc Environment +-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()). +env(Key, Default) -> application:get_env(?APP, Key, Default). + +%% @doc Is running? +-spec(is_running(node()) -> boolean()). +is_running(Node) -> + case rpc:call(Node, erlang, whereis, [?APP]) of + {badrpc, _} -> false; + undefined -> false; + Pid when is_pid(Pid) -> true + end. + +%%-------------------------------------------------------------------- +%% PubSub APIs +%%-------------------------------------------------------------------- + +%% @doc Subscribe +-spec(subscribe(iodata()) -> ok | {error, term()}). +subscribe(Topic) -> + emqttd_server:subscribe(iolist_to_binary(Topic)). + +-spec(subscribe(iodata(), subscriber()) -> ok | {error, term()}). +subscribe(Topic, Subscriber) -> + emqttd_server:subscribe(iolist_to_binary(Topic), Subscriber). + +-spec(subscribe(iodata(), subscriber(), [suboption()]) -> ok | {error, term()}). +subscribe(Topic, Subscriber, Options) -> + emqttd_server:subscribe(iolist_to_binary(Topic), Subscriber, Options). + +%% @doc Publish MQTT Message +-spec(publish(mqtt_message()) -> {ok, mqtt_delivery()} | ignore). +publish(Msg) -> + emqttd_server:publish(Msg). + +%% @doc Unsubscribe +-spec(unsubscribe(iodata()) -> ok | {error, term()}). +unsubscribe(Topic) -> + emqttd_server:unsubscribe(iolist_to_binary(Topic)). + +-spec(unsubscribe(iodata(), subscriber()) -> ok | {error, term()}). +unsubscribe(Topic, Subscriber) -> + emqttd_server:unsubscribe(iolist_to_binary(Topic), Subscriber). + +-spec(setqos(binary(), subscriber(), mqtt_qos()) -> ok). +setqos(Topic, Subscriber, Qos) -> + emqttd_server:setqos(iolist_to_binary(Topic), Subscriber, Qos). + +-spec(topics() -> [binary()]). +topics() -> emqttd_router:topics(). + +-spec(subscribers(iodata()) -> list(subscriber())). +subscribers(Topic) -> + emqttd_server:subscribers(iolist_to_binary(Topic)). + +-spec(subscriptions(subscriber()) -> [{emqttd:subscriber(), binary(), list(emqttd:suboption())}]). +subscriptions(Subscriber) -> + emqttd_server:subscriptions(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, term()}). +hook(Hook, TagFunction, InitArgs) -> + emqttd_hooks:add(Hook, TagFunction, InitArgs). + +-spec(hook(atom(), function() | {emqttd_hooks:hooktag(), function()}, list(any()), integer()) + -> 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, term()}). +unhook(Hook, TagFunction) -> + emqttd_hooks:delete(Hook, TagFunction). + +-spec(run_hooks(atom(), list(any())) -> ok | stop). +run_hooks(Hook, Args) -> + emqttd_hooks:run(Hook, Args). + +-spec(run_hooks(atom(), list(any()), any()) -> {ok | stop, any()}). +run_hooks(Hook, Args, Acc) -> + emqttd_hooks:run(Hook, Args, Acc). + +%%-------------------------------------------------------------------- +%% Shutdown and reboot +%%-------------------------------------------------------------------- + +shutdown() -> + shutdown(normal). + +shutdown(Reason) -> + lager:error("EMQ shutdown for ~s", [Reason]), + emqttd_plugins:unload(), + lists:foreach(fun application:stop/1, [emqttd, ekka, mochiweb, esockd, gproc]). + +reboot() -> + lists:foreach(fun application:start/1, [gproc, esockd, mochiweb, ekka, emqttd]). + +%%-------------------------------------------------------------------- +%% Debug +%%-------------------------------------------------------------------- + +dump() -> lists:append([emqttd_server:dump(), emqttd_router:dump()]). + diff --git a/src/emqx.erl b/src/emqx.erl index 66c368cf3..bdb88ed9a 100644 --- a/src/emqx.erl +++ b/src/emqx.erl @@ -37,8 +37,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]). @@ -51,7 +50,9 @@ -type(listener() :: {atom(), esockd:listen_on(), [esockd:option()]}). --type(subscriber() :: pid() | binary()). +-type(subid() :: binary()). + +-type(subscriber() :: pid() | subid() | {subid(), pid()}). -type(suboption() :: local | {qos, non_neg_integer()} | {share, {'$queue' | binary()}}). @@ -64,19 +65,19 @@ %%-------------------------------------------------------------------- %% @doc Start emqx application. --spec(start() -> ok | {error, any()}). +-spec(start() -> ok | {error, term()}). start() -> application:start(?APP). %% @doc Stop emqx application. --spec(stop() -> ok | {error, any()}). +-spec(stop() -> ok | {error, term()}). stop() -> application:stop(?APP). %% @doc Get 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 with default --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? @@ -176,13 +177,13 @@ merge_sockopts(Options) -> %%-------------------------------------------------------------------- %% @doc Subscribe --spec(subscribe(iodata()) -> ok | {error, any()}). +-spec(subscribe(iodata()) -> ok | {error, term()}). subscribe(Topic) -> - subscribe(Topic, self()). + emqx_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, []). + emqx_server:subscribe(iolist_to_binary(Topic), Subscriber). -spec(subscribe(iodata(), subscriber(), [suboption()]) -> ok | {error, term()}). subscribe(Topic, Subscriber, Options) -> @@ -196,7 +197,7 @@ publish(Msg) -> %% @doc Unsubscribe -spec(unsubscribe(iodata()) -> ok | {error, term()}). unsubscribe(Topic) -> - unsubscribe(Topic, self()). + emqx_server:unsubscribe(iolist_to_binary(Topic)). -spec(unsubscribe(iodata(), subscriber()) -> ok | {error, term()}). unsubscribe(Topic, Subscriber) -> @@ -217,34 +218,30 @@ topics() -> emqx_router:topics(). subscribers(Topic) -> emqx_server:subscribers(iolist_to_binary(Topic)). --spec(subscriptions(subscriber()) -> [{binary(), binary(), list(suboption())}]). +-spec(subscriptions(subscriber()) -> [{subscriber(), binary(), list(suboption())}]). subscriptions(Subscriber) -> emqx_server:subscriptions(Subscriber). --spec(is_subscribed(iodata(), subscriber()) -> boolean()). -is_subscribed(Topic, Subscriber) -> - emqx_server:is_subscribed(iolist_to_binary(Topic), Subscriber). - --spec(subscriber_down(subscriber()) -> ok). -subscriber_down(Subscriber) -> - emqx_server:subscriber_down(Subscriber). +-spec(subscribed(iodata(), subscriber()) -> boolean()). +subscribed(Topic, Subscriber) -> + emqx_server:subscribed(iolist_to_binary(Topic), Subscriber). %%-------------------------------------------------------------------- %% Hooks API %%-------------------------------------------------------------------- -spec(hook(atom(), function() | {emqx_hooks:hooktag(), function()}, list(any())) - -> ok | {error, any()}). + -> ok | {error, term()}). hook(Hook, TagFunction, InitArgs) -> emqx_hooks:add(Hook, TagFunction, InitArgs). -spec(hook(atom(), function() | {emqx_hooks:hooktag(), function()}, list(any()), integer()) - -> ok | {error, any()}). + -> ok | {error, term()}). hook(Hook, TagFunction, InitArgs, Priority) -> emqx_hooks:add(Hook, TagFunction, InitArgs, Priority). -spec(unhook(atom(), function() | {emqx_hooks:hooktag(), function()}) - -> ok | {error, any()}). + -> ok | {error, term()}). unhook(Hook, TagFunction) -> emqx_hooks:delete(Hook, TagFunction). diff --git a/src/emqx_access_control.erl b/src/emqx_access_control.erl index 2515689fa..bdaa20626 100644 --- a/src/emqx_access_control.erl +++ b/src/emqx_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/emqx_acl_mod.erl b/src/emqx_acl_mod.erl index 6ce338209..be167f5c1 100644 --- a/src/emqx_acl_mod.erl +++ b/src/emqx_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/emqx_bridge_sup.erl b/src/emqx_bridge_sup.erl index dcfb09a6c..ee4101579 100644 --- a/src/emqx_bridge_sup.erl +++ b/src/emqx_bridge_sup.erl @@ -25,8 +25,7 @@ %%-------------------------------------------------------------------- %% @doc Start bridge pool supervisor --spec(start_link(atom(), binary(), [emqx_bridge:option()]) -> - {ok, pid()} | {error, any()}). +-spec(start_link(atom(), binary(), [emqx_bridge:option()]) -> {ok, pid()} | {error, term()}). start_link(Node, Topic, Options) -> MFA = {emqx_bridge, start_link, [Node, Topic, Options]}, emqx_pool_sup:start_link({bridge, Node, Topic}, random, MFA). diff --git a/src/emqx_bridge_sup_sup.erl b/src/emqx_bridge_sup_sup.erl index d3f5d6bc5..4da7eded5 100644 --- a/src/emqx_bridge_sup_sup.erl +++ b/src/emqx_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(), [emqx_bridge:option()]) -> {ok, pid()} | {error, any()}). +-spec(start_bridge(atom(), binary(), [emqx_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/emqx_broker.erl b/src/emqx_broker.erl index dd59c2f11..416cadb20 100644 --- a/src/emqx_broker.erl +++ b/src/emqx_broker.erl @@ -61,7 +61,7 @@ %%-------------------------------------------------------------------- %% @doc Start the 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/emqx_cli.erl b/src/emqx_cli.erl index 37f9a904a..14c1498b4 100644 --- a/src/emqx_cli.erl +++ b/src/emqx_cli.erl @@ -219,7 +219,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"}, @@ -244,38 +246,37 @@ 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 emqx: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 emqx:subscribe(bin(Topic), bin(ClientId), [{qos, IntQos}]) of - ok -> - ?PRINT_MSG("ok~n"); - {error, already_existed} -> - ?PRINT_MSG("Error: already existed~n"); - {error, Reason} -> - ?PRINT("Error: ~p~n", [Reason]) - end - end, - if_valid_qos(QoS, Add); - -subscriptions(["del", ClientId]) -> - Ok = emqx_server:subscriber_down(bin(ClientId)), - ?PRINT("~p~n", [Ok]); + if_valid_qos(QoS, fun(IntQos) -> + case emqx_sm:lookup_session(bin(ClientId)) of + undefined -> + ?PRINT_MSG("Error: Session not found!"); + #mqtt_session{sess_pid = SessPid} -> + {Topic1, Options} = emqx_topic:parse(bin(Topic)), + emqx_session:subscribe(SessPid, [{Topic1, [{qos, IntQos}|Options]}]), + ?PRINT_MSG("ok~n") + end + end); subscriptions(["del", ClientId, Topic]) -> - Ok = emqx:unsubscribe(bin(Topic), bin(ClientId)), - ?PRINT("~p~n", [Ok]); + case emqx_sm:lookup_session(bin(ClientId)) of + undefined -> + ?PRINT_MSG("Error: Session not found!"); + #mqtt_session{sess_pid = SessPid} -> + emqx_session:unsubscribe(SessPid, [emqx_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) -> @@ -582,14 +583,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) -> emqx_time:now_secs(Val); diff --git a/src/emqx_cm.erl b/src/emqx_cm.erl index f56ce2929..cc73155ba 100644 --- a/src/emqx_cm.erl +++ b/src/emqx_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/emqx_gen_mod.erl b/src/emqx_gen_mod.erl index 24c0036b8..e808db1f6 100644 --- a/src/emqx_gen_mod.erl +++ b/src/emqx_gen_mod.erl @@ -20,9 +20,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/emqx_http.erl b/src/emqx_http.erl index 109c1e6ee..efc24e8fc 100644 --- a/src/emqx_http.erl +++ b/src/emqx_http.erl @@ -61,8 +61,10 @@ handle_request(Req, State) -> inner_handle_request(Req, State) -> Path = Req:get(path), - handle_request(Path, Req, State). - + case Path of + "/api/v2/auth" -> handle_request(Path, Req, State); + _ -> if_authorized(Req, fun() -> handle_request(Path, Req, State) end) + end. handle_request("/api/v2/" ++ Url, Req, #state{dispatch = Dispatch}) -> Dispatch(Req, Url); diff --git a/src/emqx_keepalive.erl b/src/emqx_keepalive.erl index 44dac454c..ddffbe611 100644 --- a/src/emqx_keepalive.erl +++ b/src/emqx_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/emqx_metrics.erl b/src/emqx_metrics.erl index 032969234..166173d18 100644 --- a/src/emqx_metrics.erl +++ b/src/emqx_metrics.erl @@ -97,8 +97,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/emqx_mgmt.erl.bk b/src/emqx_mgmt.erl.bk index 3d1141018..14345c019 100644 --- a/src/emqx_mgmt.erl.bk +++ b/src/emqx_mgmt.erl.bk @@ -494,29 +494,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/emqx_mod_sup.erl b/src/emqx_mod_sup.erl index a3f4f19ff..e5ccbb1ef 100644 --- a/src/emqx_mod_sup.erl +++ b/src/emqx_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/emqx_parser.erl b/src/emqx_parser.erl index ba1624af1..3af9f984b 100644 --- a/src/emqx_parser.erl +++ b/src/emqx_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/emqx_plugins.erl b/src/emqx_plugins.erl index ad5523e54..4cd4f997b 100644 --- a/src/emqx_plugins.erl +++ b/src/emqx_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 emqx: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 emqx: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/emqx_pool_sup.erl b/src/emqx_pool_sup.erl index afed06e87..984e9180e 100644 --- a/src/emqx_pool_sup.erl +++ b/src/emqx_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() | tuple(), atom(), pos_integer(), mfa()) -> {ok, pid()} | {error, any()}). +-spec(start_link(atom() | tuple(), 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/emqx_pooler.erl b/src/emqx_pooler.erl index c60a22e27..b5c780883 100644 --- a/src/emqx_pooler.erl +++ b/src/emqx_pooler.erl @@ -42,7 +42,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/emqx_protocol.erl b/src/emqx_protocol.erl index b03213310..532fd49e6 100644 --- a/src/emqx_protocol.erl +++ b/src/emqx_protocol.erl @@ -145,7 +145,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), @@ -409,7 +409,11 @@ shutdown(conflict, #proto_state{client_id = _ClientId}) -> shutdown(Error, State = #proto_state{will_msg = WillMsg}) -> ?LOG(info, "Shutdown for ~p", [Error], State), Client = client(State), - send_willmsg(Client, WillMsg), + %% Auth failure not publish the will message + case Error =:= auth_failure of + true -> ok; + false -> send_willmsg(Client, WillMsg) + end, emqx_hooks:run('client.disconnected', [Error], Client), %% let it down %% emqx_cm:unreg(ClientId). diff --git a/src/emqx_pubsub.erl b/src/emqx_pubsub.erl index f67c3c8e4..06f1ea3bc 100644 --- a/src/emqx_pubsub.erl +++ b/src/emqx_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,8 +54,8 @@ start_link(Pool, Id, Env) -> %% PubSub API %%-------------------------------------------------------------------- -%% @doc Subscribe a Topic --spec(subscribe(binary(), emqx:subscriber(), [emqx:suboption()]) -> ok). +%% @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 Message to a Topic --spec(publish(binary(), any()) -> {ok, mqtt_delivery()} | ignore). +%% @doc Publish MQTT Message to a Topic +-spec(publish(binary(), mqtt_message()) -> {ok, mqtt_delivery()} | ignore). publish(Topic, Msg) -> route(lists:append(emqx_router:match(Topic), emqx_router:match_local(Topic)), delivery(Msg)). @@ -73,7 +73,7 @@ route([], #mqtt_delivery{message = Msg}) -> emqx_hooks:run('message.dropped', [undefined, Msg]), dropped(Msg#mqtt_message.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]}); @@ -83,8 +83,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 = []}. @@ -93,7 +93,7 @@ delivery(Msg) -> #mqtt_delivery{sender = self(), message = Msg, flows = []}. forward(Node, To, Delivery) -> emqx_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 @@ -109,16 +109,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) -> - emqx_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) -> @@ -128,8 +128,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), @@ -157,8 +157,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 @@ -171,22 +171,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). @@ -207,39 +207,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 emqx_router:add_route(Topic), +add_global_subscriber(Share, Topic, Subscriber) -> + case ets:member(mqtt_subscriber, Topic) and emqx_router:has_route(Topic) of + true -> ok; + false -> emqx_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 emqx_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 emqx_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 emqx_router:del_local_route(Topic). shared(undefined, Subscriber) -> Subscriber; shared(Share, Subscriber) -> - {Share, Subscriber}. + {share, Share, Subscriber}. setstats(State) -> emqx_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/emqx_rest_api.erl.bk b/src/emqx_rest_api.erl.bk index f405f1ce5..f04a7c538 100644 --- a/src/emqx_rest_api.erl.bk +++ b/src/emqx_rest_api.erl.bk @@ -67,9 +67,9 @@ -http_api({"^users/?$", 'GET', users, []}). -http_api({"^users/?$", 'POST', users, [{<<"username">>, binary}, {<<"password">>, binary}, - {<<"tag">>, binary}]}). + {<<"tags">>, binary}]}). -http_api({"^users/(.+?)/?$", 'GET', users, []}). --http_api({"^users/(.+?)/?$", 'PUT', users, []}). +-http_api({"^users/(.+?)/?$", 'PUT', users, [{<<"tags">>, binary}]}). -http_api({"^users/(.+?)/?$", 'DELETE', users, []}). -http_api({"^auth/?$", 'POST', auth, [{<<"username">>, binary}, {<<"password">>, binary}]}). @@ -213,9 +213,10 @@ session_list('GET', Params, Node, ClientId) -> {ok, [{objects, [session_row(Row) || Row <- Data]}]}. session_row({ClientId, _Pid, _Persistent, Session}) -> - InfoKeys = [clean_sess, max_inflight, inflight_queue, message_queue, - message_dropped, awaiting_rel, awaiting_ack, awaiting_comp, created_at], - [{client_id, ClientId} | [{Key, format(Key, get_value(Key, Session))} || Key <- InfoKeys]]. + Data = lists:append(Session, emqttd_stats:get_session_stats(ClientId)), + InfoKeys = [clean_sess, subscriptions, max_inflight, inflight_len, mqueue_len, + mqueue_dropped, awaiting_rel_len, deliver_msg,enqueue_msg, created_at], + [{client_id, ClientId} | [{Key, format(Key, get_value(Key, Data))} || Key <- InfoKeys]]. %%-------------------------------------------------------------------------- %% subscription @@ -241,10 +242,14 @@ subscription_list('GET', Params, Node, Key) -> Data = emqx_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/emqx_router.erl b/src/emqx_router.erl index b16d8af1f..ac356f7de 100644 --- a/src/emqx_router.erl +++ b/src/emqx_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 emqx_trie:match/1, [Topic]), + %% Optimize: ets??? + Matched = mnesia:ets(fun emqx_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 emqx_topic:wildcard(Topic) of - true -> emqx_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 emqx_topic:wildcard(Topic) of - true -> emqx_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), - emqx_topic:match(Name, Filter)]. + case ets:info(mqtt_local_route, size) of + 0 -> []; + _ -> ets:foldl( + fun({Filter, Node}, Matched) -> + case emqx_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/emqx_rpc.erl b/src/emqx_rpc.erl index 17525c065..a39b1c784 100644 --- a/src/emqx_rpc.erl +++ b/src/emqx_rpc.erl @@ -1,8 +1,22 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2013-2017 EMQ Enterprise, Inc. (http://emqtt.io) %% -%% Copyright (c) 2013-2017 EMQ Enterprise, Inc. All Rights Reserved. +%% 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. +%%-------------------------------------------------------------------- %% %% @doc EMQ X Distributed RPC. %% +%%-------------------------------------------------------------------- -module(emqx_rpc). diff --git a/src/emqx_server.erl b/src/emqx_server.erl index 0ffc99eb6..12886656d 100644 --- a/src/emqx_server.erl +++ b/src/emqx_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 :: emqx_pmon:pmon() }). +-record(state, {pool, id, env, subids :: map(), submon :: emqx_pmon:pmon()}). -%% @doc Start a 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 | {error, any()}). +%% @doc Subscribe to a Topic. +-spec(subscribe(binary()) -> ok | {error, term()}). subscribe(Topic) when is_binary(Topic) -> subscribe(Topic, self()). --spec(subscribe(binary(), emqx:subscriber()) -> ok | {error, any()}). +-spec(subscribe(binary(), emqttd:subscriber()) -> ok | {error, term()}). subscribe(Topic, Subscriber) when is_binary(Topic) -> subscribe(Topic, Subscriber, []). --spec(subscribe(binary(), emqx:subscriber(), [emqx:suboption()]) -> - ok | {error, any()}). +-spec(subscribe(binary(), emqttd:subscriber(), [emqttd:suboption()]) -> + 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(), emqx:subscriber(), [emqx: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 a message -spec(publish(mqtt_message()) -> {ok, mqtt_delivery()} | ignore). @@ -109,14 +108,14 @@ trace(publish, From, #mqtt_message{topic = Topic, payload = Payload}) when is_bi "~s PUBLISH to ~s: ~p", [From, Topic, Payload]). %% @doc Unsubscribe --spec(unsubscribe(binary()) -> ok | {error, any()}). +-spec(unsubscribe(binary()) -> ok | {error, term()}). unsubscribe(Topic) when is_binary(Topic) -> unsubscribe(Topic, self()). %% @doc Unsubscribe --spec(unsubscribe(binary(), emqx:subscriber()) -> ok | {error, any()}). +-spec(unsubscribe(binary(), emqx: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(), emqx: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(), emqx: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(emqx:subscriber()) -> [{binary(), list(emqx: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(subscriber()) -> [{subscriber(), binary(), list(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(emqx:subscriber())). +subscribers(Topic) when is_binary(Topic) -> emqx_pubsub:subscribers(Topic). --spec(is_subscribed(binary(), emqx:subscriber()) -> boolean()). -is_subscribed(Topic, Subscriber) when is_binary(Topic) -> - ets:member(mqtt_subproperty, {Topic, Subscriber}). - --spec(subscriber_down(emqx: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 = emqx_pmon:new()}}. + State = #state{pool = Pool, id = Id, env = Env, + subids = #{}, submon = emqx_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 [] -> emqx_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}] -> emqx_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) -> emqx_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/emqx_session.erl b/src/emqx_session.erl index 1a9bfb65e..d716388b7 100644 --- a/src/emqx_session.erl +++ b/src/emqx_session.erl @@ -173,7 +173,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], []). @@ -193,7 +193,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 emqx_server:publish(Msg), ok; @@ -391,6 +391,7 @@ handle_cast({subscribe, From, TopicTable, AckFun}, SubMap; {ok, OldQos} -> emqx_server:setqos(Topic, ClientId, NewQos), + emqx_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); @@ -545,7 +546,7 @@ handle_info({dispatch, _Topic, #mqtt_message{from = {ClientId, _}}}, %% Dispatch Message handle_info({dispatch, Topic, Msg}, State) when is_record(Msg, mqtt_message) -> - hibernate(gc(dispatch(tune_qos(Topic, Msg, State), State))); + hibernate(gc(dispatch(tune_qos(Topic, reset_dup(Msg), State), State))); %% Do nothing if the client has been disconnected. handle_info({timeout, _Timer, retry_delivery}, State = #state{client_pid = undefined}) -> @@ -588,9 +589,9 @@ handle_info(Info, Session) -> ?UNEXPECTED_INFO(Info, Session). terminate(Reason, #state{client_id = ClientId, username = Username}) -> - emqx_stats:del_session_stats(ClientId), + %% Move to emqx_sm to avoid race condition + %% emqx_stats:del_session_stats(ClientId), emqx_hooks:run('session.terminated', [ClientId, Username, Reason]), - emqx_server:subscriber_down(ClientId), emqx_sm:unregister_session(ClientId). code_change(_OldVsn, Session, _Extra) -> @@ -811,6 +812,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/emqx_sm.erl b/src/emqx_sm.erl index 8c9b1c8eb..eac7743bc 100644 --- a/src/emqx_sm.erl +++ b/src/emqx_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 @@ -184,17 +185,15 @@ handle_cast(Msg, State) -> handle_info({'DOWN', MRef, process, DownPid, _Reason}, State) -> case dict:find(MRef, State#state.monitors) of {ok, ClientId} -> - mnesia:transaction(fun() -> + mnesia:transaction( + fun() -> case mnesia:wread({mqtt_session, ClientId}) of - [] -> - ok; [Sess = #mqtt_session{sess_pid = DownPid}] -> - emqx_stats:del_session_stats(ClientId), mnesia:delete_object(mqtt_session, Sess, write); - [_Sess] -> - ok - end - end), + [_Sess] -> ok; + [] -> ok + end + end), {noreply, erase_monitor(MRef, State), hibernate}; error -> lager:error("MRef of session ~p not found", [DownPid]), diff --git a/src/emqx_sm_helper.erl b/src/emqx_sm_helper.erl index 4d0e7a9cb..90f43be0b 100644 --- a/src/emqx_sm_helper.erl +++ b/src/emqx_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/emqx_trace.erl b/src/emqx_trace.erl index 7b09b55a8..22726336f 100644 --- a/src/emqx_trace.erl +++ b/src/emqx_trace.erl @@ -46,7 +46,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}); @@ -56,7 +56,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}) -> @@ -71,7 +71,7 @@ all_traces() -> gen_server:call(?MODULE, all_traces). %%-------------------------------------------------------------------- init([]) -> - {ok, #state{level = info, traces = #{}}}. + {ok, #state{level = debug, traces = #{}}}. handle_call({start_trace, Who, LogFile}, _From, State = #state{level = Level, traces = Traces}) -> case lager:trace_file(LogFile, [Who], Level, ?TRACE_OPTIONS) of diff --git a/src/emqx_trie.erl b/src/emqx_trie.erl index 4ad525a90..b4e19bbc8 100644 --- a/src/emqx_trie.erl +++ b/src/emqx_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, emqx_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, emqx_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, emqx_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(emqx_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(emqx_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_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]). + diff --git a/test/emqx_SUITE.erl b/test/emqx_SUITE.erl index c803244ab..48753c1f5 100644 --- a/test/emqx_SUITE.erl +++ b/test/emqx_SUITE.erl @@ -100,7 +100,8 @@ groups() -> run_hooks]}, {http, [sequence], [request_status, - request_publish + request_publish, + get_api_lists % websocket_test ]}, {alarms, [sequence], @@ -126,7 +127,8 @@ groups() -> [cleanSession_validate, cleanSession_validate1] }, - {rest_api, [sequence], [get_api_lists]} + {rest_api, [sequence], + [get_api_lists]} ]. init_per_suite(Config) -> @@ -239,14 +241,15 @@ pubsub(_) -> emqx:unsubscribe(<<"a/b/c">>). t_local_subscribe(_) -> - emqx:subscribe("$local/topic0"), - emqx:subscribe("$local/topic1", <<"x">>), - emqx:subscribe("$local/topic2", <<"x">>, [{qos, 2}]), + ok = emqx:subscribe("$local/topic0"), + ok = emqx:subscribe("$local/topic1", <<"x">>), + ok = emqx:subscribe("$local/topic2", <<"x">>, [{qos, 2}]), timer:sleep(10), ?assertEqual([self()], emqx:subscribers("$local/topic0")), - ?assertEqual([<<"x">>], emqx:subscribers("$local/topic1")), - ?assertEqual([{<<"$local/topic1">>,<<"x">>,[]},{<<"$local/topic2">>,<<"x">>,[{qos,2}]}], emqx:subscriptions(<<"x">>)), - + ?assertEqual([{<<"x">>, self()}], emqx:subscribers("$local/topic1")), + ?assertEqual([{{<<"x">>, self()}, <<"$local/topic1">>, []}, + {{<<"x">>, self()}, <<"$local/topic2">>, [{qos,2}]}], + emqx:subscriptions(<<"x">>)), ?assertEqual(ok, emqx:unsubscribe("$local/topic0")), ?assertMatch({error, {subscription_not_found, _}}, emqx:unsubscribe("$local/topic0")), ?assertEqual(ok, emqx:unsubscribe("$local/topic1", <<"x">>)), @@ -260,9 +263,9 @@ t_shared_subscribe(_) -> emqx:subscribe("$queue/topic3"), timer:sleep(10), ?assertEqual([self()], emqx: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(emqx:subscriptions(self()))), emqx:unsubscribe("$local/$share/group1/topic1"), emqx:unsubscribe("$share/group2/topic2"), @@ -302,7 +305,7 @@ router_add_del(_) -> %% Add emqx_router:add_route(<<"#">>), emqx_router:add_route(<<"a/b/c">>), - emqx_router:add_route(<<"+/#">>, node()), + emqx_router:add_route(<<"+/#">>), Routes = [R1, R2 | _] = [ #mqtt_route{topic = <<"#">>, node = node()}, #mqtt_route{topic = <<"+/#">>, node = node()}, @@ -310,7 +313,7 @@ router_add_del(_) -> Routes = lists:sort(emqx_router:match(<<"a/b/c">>)), %% Batch Add - emqx_router:add_routes(Routes), + lists:foreach(fun(R) -> emqx_router:add_route(R) end, Routes), Routes = lists:sort(emqx_router:match(<<"a/b/c">>)), %% Del @@ -321,7 +324,8 @@ router_add_del(_) -> %% Batch Del R3 = #mqtt_route{topic = <<"#">>, node = 'a@127.0.0.1'}, emqx_router:add_route(R3), - emqx_router:del_routes([R1, R2]), + emqx_router:del_route(R1), + emqx_router:del_route(R2), emqx_router:del_route(R3), [] = lists:sort(emqx_router:match(<<"a/b/c">>)). @@ -329,7 +333,7 @@ router_print(_) -> Routes = [#mqtt_route{topic = <<"a/b/c">>, node = node()}, #mqtt_route{topic = <<"#">>, node = node()}, #mqtt_route{topic = <<"+/#">>, node = node()}], - emqx_router:add_routes(Routes), + lists:foreach(fun(R) -> emqx_router:add_route(R) end, Routes), emqx_router:print(<<"a/b/c">>). router_unused(_) -> @@ -452,19 +456,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_emqx_publish_(Method, Api, Params, Auth) -> Url = "http://127.0.0.1:8080/" ++ Api, @@ -483,7 +488,9 @@ auth_header_(User, Pass) -> Encoded = base64:encode_to_string(lists:append([User,":",Pass])), {"Authorization","Basic " ++ Encoded}. -%%TODO: ... +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}, @@ -594,9 +601,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(_) -> @@ -821,7 +828,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;