调整协议格式，支持tls

bin/ca.crt

@@ -0,0 +1,23 @@
+-----BEGIN CERTIFICATE-----
+MIID2TCCAsGgAwIBAgIJAJ9gdUQAa53yMA0GCSqGSIb3DQEBCwUAMIGCMQswCQYD
+VQQGEwJDTjELMAkGA1UECAwCSEExCzAJBgNVBAcMAlpaMQ8wDQYDVQQKDAZ0b3Bu
+ZXQxETAPBgNVBAsMCGludGVybmV0MRQwEgYDVQQDDAt3d3cucW14LnRvcDEfMB0G
+CSqGSIb3DQEJARYQMzk0OTIyODE0QHFxLmNvbTAeFw0xODA3MDcwMjQyMzNaFw0y
+ODA3MDQwMjQyMzNaMIGCMQswCQYDVQQGEwJDTjELMAkGA1UECAwCSEExCzAJBgNV
+BAcMAlpaMQ8wDQYDVQQKDAZ0b3BuZXQxETAPBgNVBAsMCGludGVybmV0MRQwEgYD
+VQQDDAt3d3cucW14LnRvcDEfMB0GCSqGSIb3DQEJARYQMzk0OTIyODE0QHFxLmNv
+bTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMfUV+rQuTwk1n149uVa
+VruoaHHCsymukurMximtbkvnGFe8MrGx8txw/DKhOekCj0hKKMtw45JE2NF2Xsch
+otiCcrSDbW2GoTaphUGoDsdCjpXca5/nxXUIjtO9a29Hn2jcGZmeDqkH+qgw/Z3D
+M2x6Q8oMmI55bcW7/UlB6gOQKTZF2nKt8byeIZq719LXwr+tWLm54f8a45tr0cYO
+k8UuuwDTVt9aXqNnhWVJzFG4hEI3uTOZPKisH8qu4e3uyAZjX3Ms9y8wB4wR7Mzv
+gi0qLoyYVRINVi8wFMUL5WPMEVG4QT3zSyMrI0anYMZQmNS9EStkT7X5YlZM8/oo
+9m0CAwEAAaNQME4wHQYDVR0OBBYEFEZUOcGhhjZR51Jhzq/E9BQ0bX4gMB8GA1Ud
+IwQYMBaAFEZUOcGhhjZR51Jhzq/E9BQ0bX4gMAwGA1UdEwQFMAMBAf8wDQYJKoZI
+hvcNAQELBQADggEBAJjb/qEH9Oj52KPZ08cdpUa36xWuHWKRzE44tO7WMF44dHcu
+wQy66fA0xiIu4ZM8gBB8T1pSYL6jCkA9czqwkRl/KUmwW+vloRhmbLB0Pga0CcAl
+r/R4p0g+OKVW5wSGZispnrkcIAvtiaI5e/+YIt5G7wNut2g3nsbhzRvspzLwmrdc
+x60xoE/o9qCD4/AvL4h/lmhTIeK9UGWOgNww3xfd8w+Kso6ObJDlO00ghSN82uXC
+1zt0j5lawlJnbH7OGYZTqF9tkeUb6AJZhGFaoHczqtVg5YR1VAElijBAxD7ZSq8r
+UWIGHuQvR3vnzTgVhQ8hO7yuDjHR6jY2u+Xj3W4=
+-----END CERTIFICATE-----

bin/demo.xiaoa7.com.crt

@@ -0,0 +1,71 @@
+Certificate:
+    Data:
+        Version: 3 (0x2)
+        Serial Number: 146 (0x92)
+    Signature Algorithm: sha256WithRSAEncryption
+        Issuer: C=CN, ST=HA, L=ZZ, O=topnet, OU=internet, CN=www.qmx.top/emailAddress=394922814@qq.com
+        Validity
+            Not Before: Oct 10 08:25:11 2018 GMT
+            Not After : Oct  9 08:25:11 2022 GMT
+        Subject: C=CN, ST=HA, L=ZZ, O=topnet, OU=internet, CN=demo.xiaoa7.com/emailAddress=demo@xiaoa7.com
+        Subject Public Key Info:
+            Public Key Algorithm: rsaEncryption
+                Public-Key: (1024 bit)
+                Modulus:
+                    00:d7:d2:64:4d:a7:c1:d6:b8:b0:9b:7d:4d:90:1f:
+                    eb:80:7e:fe:ed:65:da:a5:e9:26:83:7c:bf:07:26:
+                    1f:7f:9d:0a:7c:16:2e:9d:b3:c7:73:c8:75:1e:01:
+                    e9:2e:fe:d1:8a:88:21:c6:09:c0:a5:4d:8f:a6:22:
+                    b3:3d:57:96:03:68:c1:58:55:5c:36:0c:df:80:9a:
+                    41:f9:8a:99:2a:e4:3a:45:08:a9:e8:d7:b8:c6:33:
+                    75:57:05:23:6a:22:dd:34:1f:c9:d2:02:be:89:01:
+                    6c:c2:2a:5e:5a:aa:24:bd:e2:43:18:59:6a:42:cf:
+                    b5:4d:ff:53:df:3f:31:fd:81
+                Exponent: 65537 (0x10001)
+        X509v3 extensions:
+            X509v3 Basic Constraints: 
+                CA:FALSE
+            Netscape Comment: 
+                OpenSSL Generated Certificate
+            X509v3 Subject Key Identifier: 
+                F0:70:1C:4A:3F:10:71:EB:F8:CF:E1:D3:B4:C4:60:B7:78:51:4B:C9
+            X509v3 Authority Key Identifier: 
+                keyid:46:54:39:C1:A1:86:36:51:E7:52:61:CE:AF:C4:F4:14:34:6D:7E:20
+
+    Signature Algorithm: sha256WithRSAEncryption
+         68:25:2e:6b:27:ef:73:da:42:0f:a8:59:7b:f4:99:ff:14:3e:
+         ba:05:f6:b5:c9:e0:5e:ef:27:25:4b:37:ea:f2:27:b1:f4:7f:
+         69:f0:75:3c:3b:c0:44:d9:50:42:9c:ea:d5:1b:07:48:df:9e:
+         c5:1e:9f:10:de:f2:91:b1:d6:41:fc:80:32:06:2e:d4:67:13:
+         24:4b:65:62:3e:53:a2:77:8e:1c:c0:c7:50:10:f3:e1:97:19:
+         dc:99:17:e7:ea:d4:72:e9:52:36:88:36:f2:45:9a:7b:07:df:
+         7b:0e:99:df:ab:c1:a9:2d:00:a8:b7:b4:4f:6d:29:53:5c:c5:
+         0d:8d:6a:61:d8:ac:3b:78:74:20:08:95:01:bc:4d:a1:03:31:
+         3b:6f:65:b1:e1:1f:e9:6d:fb:bb:32:e8:05:4a:f3:45:ba:40:
+         12:6d:03:29:b4:db:bc:9c:15:fb:58:cb:ad:2e:d6:70:39:53:
+         c1:b0:d5:d6:ef:c2:7f:27:2a:50:1d:11:4f:1d:94:4e:6b:69:
+         65:5f:dc:ce:4b:77:57:df:8f:92:39:b0:4f:6a:65:e7:7a:07:
+         95:1c:ef:de:70:43:ec:19:bd:da:62:2a:3f:78:d8:f4:56:0c:
+         a0:74:1d:65:1f:9a:f5:ed:dc:31:94:e8:f7:2d:40:80:a9:44:
+         66:79:74:01
+-----BEGIN CERTIFICATE-----
+MIIDfDCCAmSgAwIBAgICAJIwDQYJKoZIhvcNAQELBQAwgYIxCzAJBgNVBAYTAkNO
+MQswCQYDVQQIDAJIQTELMAkGA1UEBwwCWloxDzANBgNVBAoMBnRvcG5ldDERMA8G
+A1UECwwIaW50ZXJuZXQxFDASBgNVBAMMC3d3dy5xbXgudG9wMR8wHQYJKoZIhvcN
+AQkBFhAzOTQ5MjI4MTRAcXEuY29tMB4XDTE4MTAxMDA4MjUxMVoXDTIyMTAwOTA4
+MjUxMVowgYUxCzAJBgNVBAYTAkNOMQswCQYDVQQIDAJIQTELMAkGA1UEBwwCWlox
+DzANBgNVBAoMBnRvcG5ldDERMA8GA1UECwwIaW50ZXJuZXQxGDAWBgNVBAMMD2Rl
+bW8ueGlhb2E3LmNvbTEeMBwGCSqGSIb3DQEJARYPZGVtb0B4aWFvYTcuY29tMIGf
+MA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDX0mRNp8HWuLCbfU2QH+uAfv7tZdql
+6SaDfL8HJh9/nQp8Fi6ds8dzyHUeAeku/tGKiCHGCcClTY+mIrM9V5YDaMFYVVw2
+DN+AmkH5ipkq5DpFCKno17jGM3VXBSNqIt00H8nSAr6JAWzCKl5aqiS94kMYWWpC
+z7VN/1PfPzH9gQIDAQABo3sweTAJBgNVHRMEAjAAMCwGCWCGSAGG+EIBDQQfFh1P
+cGVuU1NMIEdlbmVyYXRlZCBDZXJ0aWZpY2F0ZTAdBgNVHQ4EFgQU8HAcSj8Qcev4
+z+HTtMRgt3hRS8kwHwYDVR0jBBgwFoAURlQ5waGGNlHnUmHOr8T0FDRtfiAwDQYJ
+KoZIhvcNAQELBQADggEBAGglLmsn73PaQg+oWXv0mf8UProF9rXJ4F7vJyVLN+ry
+J7H0f2nwdTw7wETZUEKc6tUbB0jfnsUenxDe8pGx1kH8gDIGLtRnEyRLZWI+U6J3
+jhzAx1AQ8+GXGdyZF+fq1HLpUjaINvJFmnsH33sOmd+rwaktAKi3tE9tKVNcxQ2N
+amHYrDt4dCAIlQG8TaEDMTtvZbHhH+lt+7sy6AVK80W6QBJtAym027ycFftYy60u
+1nA5U8Gw1dbvwn8nKlAdEU8dlE5raWVf3M5Ld1ffj5I5sE9qZed6B5Uc795wQ+wZ
+vdpiKj942PRWDKB0HWUfmvXt3DGU6PctQICpRGZ5dAE=
+-----END CERTIFICATE-----

bin/demo.xiaoa7.com.key

@@ -0,0 +1,15 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIICXQIBAAKBgQDX0mRNp8HWuLCbfU2QH+uAfv7tZdql6SaDfL8HJh9/nQp8Fi6d
+s8dzyHUeAeku/tGKiCHGCcClTY+mIrM9V5YDaMFYVVw2DN+AmkH5ipkq5DpFCKno
+17jGM3VXBSNqIt00H8nSAr6JAWzCKl5aqiS94kMYWWpCz7VN/1PfPzH9gQIDAQAB
+AoGBAMOcpdzcmAwOCcRkpdBjnggxEs4I60xFLesvPdU08IexxrcBQ+8b60M+BSjV
+goxd2X4+NJ5SaJFq4zApc1u97vAuAop8Xxixvkemjb2X1GJ//EORNlqa6weysuhc
+9lmIvOezdk78NdDZ2R2grl0tEJZse7YJCIA3tlq+WazzdfgBAkEA6y6lvcBdrJff
++cWsav5jbB9FPvaWp/tCOIm6imTeNysJ6b40/d5YdwjyXgtxaR2T/E87OKsS1TDQ
+XPA3eYiS0QJBAOrtBzTJpRGsksfEcjKpmD/yKWnyS810EN4V3kOnYj2WlJvm/K9Z
+bWVaF1UdptWak7llzZeiOGy/66I7eWB9i7ECQQDlYSnmWKkEnVymkFYKRPNTkWxI
+bNmkHTwJSMvYD2ltBV01zX6B3UsLTn6VmzOSAndDr0NjkfdcV6/h9aklsD8BAkBu
+Zywey7DOiIqGZTTByF/rlVOPHbYv/J68sAw/QRdzGCQIzFnt0hoOKnOSpZe1BKrt
+0wD2hN79hIgpOsgE55tRAkAGwABpkFAVFkMwJG/pUS1wrTRnv0P6mIBPGMp1ZxFE
+UOnEhN4SWMXvymB0ou4NEEjv8N+K0E+XyU+XdvZMq1Le
+-----END RSA PRIVATE KEY-----

bin/msg.xiaoa7.com.crt

@@ -0,0 +1,94 @@
+Certificate:
+    Data:
+        Version: 3 (0x2)
+        Serial Number: 145 (0x91)
+    Signature Algorithm: sha256WithRSAEncryption
+        Issuer: C=CN, ST=HA, L=ZZ, O=topnet, OU=internet, CN=www.qmx.top/emailAddress=394922814@qq.com
+        Validity
+            Not Before: Oct 10 08:23:38 2018 GMT
+            Not After : Oct  9 08:23:38 2022 GMT
+        Subject: C=CN, ST=HA, L=ZZ, O=topnet, OU=internet, CN=msg.xiaoa7.com/emailAddress=zhp5000@126.com
+        Subject Public Key Info:
+            Public Key Algorithm: rsaEncryption
+                Public-Key: (1024 bit)
+                Modulus:
+                    00:d0:ba:de:ca:b2:41:5e:cb:76:f7:b0:9d:8b:32:
+                    2c:04:bc:03:fe:ee:07:f7:a2:41:2d:f7:08:99:67:
+                    9e:b8:c7:6e:73:57:14:18:11:77:e0:7a:60:7e:db:
+                    24:20:8c:8a:f4:87:97:ab:d6:94:3f:2e:22:36:86:
+                    27:f2:a8:3d:46:d0:76:79:d0:e7:c3:a9:c8:ee:1f:
+                    4f:94:d0:86:82:82:1a:eb:43:7c:cd:01:94:c8:fa:
+                    4f:88:f8:78:25:42:b8:d8:e2:a9:d1:37:e4:bb:06:
+                    48:27:65:5b:29:10:67:11:30:24:78:d1:9b:ae:fc:
+                    b4:09:ac:e6:16:c6:56:a1:e3
+                Exponent: 65537 (0x10001)
+        X509v3 extensions:
+            X509v3 Basic Constraints: 
+                CA:FALSE
+            Netscape Comment: 
+                OpenSSL Generated Certificate
+            X509v3 Subject Key Identifier: 
+                E2:80:34:4F:62:FF:9F:04:C1:03:66:A3:40:6B:54:CA:6B:35:E1:7C
+            X509v3 Authority Key Identifier: 
+                keyid:46:54:39:C1:A1:86:36:51:E7:52:61:CE:AF:C4:F4:14:34:6D:7E:20
+
+    Signature Algorithm: sha256WithRSAEncryption
+         7d:de:cd:60:73:cf:ad:bd:78:99:3a:1c:1c:5d:95:c1:66:62:
+         57:57:94:ee:fe:c7:b1:f9:4b:4f:ac:a4:40:a0:85:e0:4a:58:
+         fc:00:b4:1d:33:c1:d0:d7:fa:c3:5a:e7:32:2b:26:23:47:74:
+         b5:19:06:a7:6d:51:72:f8:a4:6c:cb:59:9f:9a:20:cb:ba:e3:
+         9f:ed:26:d1:7b:b2:8b:2a:7c:10:cf:0a:a3:d8:6e:e5:04:c5:
+         93:30:12:5f:86:3a:27:b6:70:a0:b1:64:0f:64:db:4c:59:22:
+         4c:4b:3a:31:cc:bd:55:d4:49:26:55:fc:5d:6b:06:f2:b8:7e:
+         81:c8:0e:09:9c:40:1a:85:e6:bd:82:e1:af:c8:fd:d0:d6:95:
+         ca:1a:44:9f:53:ff:d1:76:34:87:a0:1c:2c:45:54:18:f9:ab:
+         4c:6b:4d:39:65:f8:63:ad:78:e1:6c:93:ef:5e:d3:5c:4b:84:
+         f9:95:5b:38:25:06:e6:7b:a8:4f:17:f8:3f:24:c9:32:8b:ab:
+         34:6d:2e:7c:e8:44:b0:f7:83:10:3e:0b:14:d4:29:5f:6f:db:
+         85:8f:08:57:0e:34:b6:fc:8d:71:5b:63:17:5c:fb:44:d6:e1:
+         89:e4:3c:30:71:e6:75:07:0b:4d:8a:1e:d0:6a:04:04:26:45:
+         a8:27:60:71
+-----BEGIN CERTIFICATE-----
+MIIDezCCAmOgAwIBAgICAJEwDQYJKoZIhvcNAQELBQAwgYIxCzAJBgNVBAYTAkNO
+MQswCQYDVQQIDAJIQTELMAkGA1UEBwwCWloxDzANBgNVBAoMBnRvcG5ldDERMA8G
+A1UECwwIaW50ZXJuZXQxFDASBgNVBAMMC3d3dy5xbXgudG9wMR8wHQYJKoZIhvcN
+AQkBFhAzOTQ5MjI4MTRAcXEuY29tMB4XDTE4MTAxMDA4MjMzOFoXDTIyMTAwOTA4
+MjMzOFowgYQxCzAJBgNVBAYTAkNOMQswCQYDVQQIDAJIQTELMAkGA1UEBwwCWlox
+DzANBgNVBAoMBnRvcG5ldDERMA8GA1UECwwIaW50ZXJuZXQxFzAVBgNVBAMMDm1z
+Zy54aWFvYTcuY29tMR4wHAYJKoZIhvcNAQkBFg96aHA1MDAwQDEyNi5jb20wgZ8w
+DQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBANC63sqyQV7LdvewnYsyLAS8A/7uB/ei
+QS33CJlnnrjHbnNXFBgRd+B6YH7bJCCMivSHl6vWlD8uIjaGJ/KoPUbQdnnQ58Op
+yO4fT5TQhoKCGutDfM0BlMj6T4j4eCVCuNjiqdE35LsGSCdlWykQZxEwJHjRm678
+tAms5hbGVqHjAgMBAAGjezB5MAkGA1UdEwQCMAAwLAYJYIZIAYb4QgENBB8WHU9w
+ZW5TU0wgR2VuZXJhdGVkIENlcnRpZmljYXRlMB0GA1UdDgQWBBTigDRPYv+fBMED
+ZqNAa1TKazXhfDAfBgNVHSMEGDAWgBRGVDnBoYY2UedSYc6vxPQUNG1+IDANBgkq
+hkiG9w0BAQsFAAOCAQEAfd7NYHPPrb14mTocHF2VwWZiV1eU7v7HsflLT6ykQKCF
+4EpY/AC0HTPB0Nf6w1rnMismI0d0tRkGp21RcvikbMtZn5ogy7rjn+0m0Xuyiyp8
+EM8Ko9hu5QTFkzASX4Y6J7ZwoLFkD2TbTFkiTEs6Mcy9VdRJJlX8XWsG8rh+gcgO
+CZxAGoXmvYLhr8j90NaVyhpEn1P/0XY0h6AcLEVUGPmrTGtNOWX4Y6144WyT717T
+XEuE+ZVbOCUG5nuoTxf4PyTJMourNG0ufOhEsPeDED4LFNQpX2/bhY8IVw40tvyN
+cVtjF1z7RNbhieQ8MHHmdQcLTYoe0GoEBCZFqCdgcQ==
+-----END CERTIFICATE-----
+-----BEGIN CERTIFICATE-----
+MIID2TCCAsGgAwIBAgIJAJ9gdUQAa53yMA0GCSqGSIb3DQEBCwUAMIGCMQswCQYD
+VQQGEwJDTjELMAkGA1UECAwCSEExCzAJBgNVBAcMAlpaMQ8wDQYDVQQKDAZ0b3Bu
+ZXQxETAPBgNVBAsMCGludGVybmV0MRQwEgYDVQQDDAt3d3cucW14LnRvcDEfMB0G
+CSqGSIb3DQEJARYQMzk0OTIyODE0QHFxLmNvbTAeFw0xODA3MDcwMjQyMzNaFw0y
+ODA3MDQwMjQyMzNaMIGCMQswCQYDVQQGEwJDTjELMAkGA1UECAwCSEExCzAJBgNV
+BAcMAlpaMQ8wDQYDVQQKDAZ0b3BuZXQxETAPBgNVBAsMCGludGVybmV0MRQwEgYD
+VQQDDAt3d3cucW14LnRvcDEfMB0GCSqGSIb3DQEJARYQMzk0OTIyODE0QHFxLmNv
+bTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMfUV+rQuTwk1n149uVa
+VruoaHHCsymukurMximtbkvnGFe8MrGx8txw/DKhOekCj0hKKMtw45JE2NF2Xsch
+otiCcrSDbW2GoTaphUGoDsdCjpXca5/nxXUIjtO9a29Hn2jcGZmeDqkH+qgw/Z3D
+M2x6Q8oMmI55bcW7/UlB6gOQKTZF2nKt8byeIZq719LXwr+tWLm54f8a45tr0cYO
+k8UuuwDTVt9aXqNnhWVJzFG4hEI3uTOZPKisH8qu4e3uyAZjX3Ms9y8wB4wR7Mzv
+gi0qLoyYVRINVi8wFMUL5WPMEVG4QT3zSyMrI0anYMZQmNS9EStkT7X5YlZM8/oo
+9m0CAwEAAaNQME4wHQYDVR0OBBYEFEZUOcGhhjZR51Jhzq/E9BQ0bX4gMB8GA1Ud
+IwQYMBaAFEZUOcGhhjZR51Jhzq/E9BQ0bX4gMAwGA1UdEwQFMAMBAf8wDQYJKoZI
+hvcNAQELBQADggEBAJjb/qEH9Oj52KPZ08cdpUa36xWuHWKRzE44tO7WMF44dHcu
+wQy66fA0xiIu4ZM8gBB8T1pSYL6jCkA9czqwkRl/KUmwW+vloRhmbLB0Pga0CcAl
+r/R4p0g+OKVW5wSGZispnrkcIAvtiaI5e/+YIt5G7wNut2g3nsbhzRvspzLwmrdc
+x60xoE/o9qCD4/AvL4h/lmhTIeK9UGWOgNww3xfd8w+Kso6ObJDlO00ghSN82uXC
+1zt0j5lawlJnbH7OGYZTqF9tkeUb6AJZhGFaoHczqtVg5YR1VAElijBAxD7ZSq8r
+UWIGHuQvR3vnzTgVhQ8hO7yuDjHR6jY2u+Xj3W4=
+-----END CERTIFICATE-----

bin/msg.xiaoa7.com.key

@@ -0,0 +1,15 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIICXgIBAAKBgQDQut7KskFey3b3sJ2LMiwEvAP+7gf3okEt9wiZZ564x25zVxQY
+EXfgemB+2yQgjIr0h5er1pQ/LiI2hifyqD1G0HZ50OfDqcjuH0+U0IaCghrrQ3zN
+AZTI+k+I+HglQrjY4qnRN+S7BkgnZVspEGcRMCR40Zuu/LQJrOYWxlah4wIDAQAB
+AoGAdy7MufaQpv0MoB+jxXBPfCCTM4Rc3kQybXmJ2nY1RhLx+T/2hEutJGscIDMm
+7MLzMO8fZoouueiwtmth8geofHfLFlFG1Ui4zTfmvLh7SVWjIW/X2Tjfv/17bknQ
+aH8ojMXe3QiddOwWuCMN+A0jt/+yuZdXsNRNO/pGMINDsYkCQQDsqi56+LunOEtv
+SfIpi89CGfzrEXSQF57orEkQGDk1hwehENwd5HzegJZYmwnXdrt1Fq1lBI3IOY9K
+NuZmAr3vAkEA4chvZjwnXaBBDAHo2NnIPSzSqOOTtaBSUfmj7fBGB0NYP5T0v9iK
+U7qmK3WYMwXZYKai0mk5kq39TntCgkWPTQJBALnXXU51Wa4TF7FZaM7VGbh3fdL9
+TwIDemaNO4Zm7y2oTgZdnevfS2rRoL4NaOoIM2Xbm+dybv+mq22EtOGwD4MCQQDF
+F4uvKgmPIEpH6g7R5WJ3jH4kMe0KGnZh2tLv7NOkmXQapLP2GFe/+Wn++/stZwi3
+5x6xIGvCkCFT9iM7xu7tAkEA0BTRYnN+lNheCJ1egA36mGuZrSYpwccdhwp04mCs
+x4apDSFp/UIA4d7jRUJu2H57vOkHUcRhygEN88YWJ7XvMg==
+-----END RSA PRIVATE KEY-----

readme.txt

@@ -1,6 +1,7 @@
 v3.0
 重点修改mfw框架；
 清除相关服务；
-框架支持tls，报文提供crc64校验，连接添加token验证；
+框架支持tls，报文提供crc64校验；
+支持集群式调用；
 接入流程变动：
 1.join；2-n.操作；3.levae；4.close

src/mfw/util/README.md

@@ -0,0 +1,5 @@
+数据包格式（协议）
+HEAD(4Byte)+包长度（4Byte)+FromID(8Byte)+ToID(8Byte)+消息ID(8Byte)+事件类型（4Byte)+消息发送模式（4Byte)+数据+CRC32校验码（4Byte)
+
+段索引：
+HEAD(0)+包长度（4)+FromID(8)+ToID(16)+消息ID(24)+事件类型（28)+消息发送模式（32)+数据(36)+CRC32校验码（n-4)

src/mfw/util/catch.go

@@ -3,8 +3,6 @@ package util
 import (
 	"log"
 	"runtime"
-
-	"github.com/donnie4w/go-logger/logger"
 )
 
 //出错拦截
@@ -20,17 +18,3 @@ func Catch() {
 		}
 	}
 }
-
-//出错拦截
-func CatchLogger() {
-	if r := recover(); r != nil {
-		logger.Error(r)
-		for skip := 0; ; skip++ {
-			_, file, line, ok := runtime.Caller(skip)
-			if !ok {
-				break
-			}
-			go logger.Error(file, ", ", line)
-		}
-	}
-}

src/mfw/util/client.go

@@ -1,52 +1,122 @@
-//客户端封装,支持短线重连接，支持客户端检测
+//客户端封装,支持断线重连接，支持客户端检测
 package util
 
 import (
+	"crypto/tls"
 	"encoding/binary"
-	"encoding/json"
 	"errors"
-	"fmt"
 	"io"
 	"math"
-	"os"
 	"sync"
 	"time"
 )
 
-var IsHiddenLog bool
+//
+type CustomConn interface {
+	io.Reader
+	io.Writer
+	io.Closer
+	Dial() error
+}
 
 //封装，可以链接多个消息总线
 type Client struct {
-	conn                 *TCPClient
+	conn                 CustomConn
 	myid                 string //
-	oldmyid              string
 	businessEventHandler func(*Packet)
 	canHandleEvent       []int
 	lastCheckHeart       int64 //最后一次心跳检测
 	messageQueue         chan *Packet
 	writeQueue           chan RawData
 	resultlock           sync.Map //map[string]chan []byte
-	myname               string
-	OnConnectSuccess     func()
 }
 
 //客户端配置
 type ClientConfig struct {
-	ClientName       string          //名称
-	EventHandler     func(p *Packet) //事件处理函数
-	MsgServerAddr    string          //消息总线地址
-	CanHandleEvents  []int           //我可以处理的事件列表
-	OnRequestConnect func()          //请求重连的事件处理函数
-	OnConnectSuccess func()          //重连成功的时间处理函数
-	ReadBufferSize   int             //读缓冲区大小
-	WriteBufferSize  int             //写缓冲区大小
+	ClientName      string          //名称
+	EventHandler    func(p *Packet) //事件处理函数
+	MsgServerAddr   string          //消息总线地址
+	CanHandleEvents []int           //我可以处理的事件列表
+	ReadBufferSize  int             //读缓冲区大小
+	WriteBufferSize int             //写缓冲区大小
+	UseTls          bool            //是否使用tls
+	TlsCertFile     string
+	TlsKeyFile      string
 }
 
-//
-func printlog(p ...interface{}) {
-	if !IsHiddenLog {
-		fmt.Print(p...)
+//启动客户端
+func NewClient(conf *ClientConfig) (*Client, error) {
+	if conf.MsgServerAddr == "" {
+		return nil, errors.New("配置项缺失(MsgServerAddr)")
+	} else if conf.EventHandler == nil {
+		return nil, errors.New("配置项缺失(EventHandler)")
+	}
+	if conf.ReadBufferSize == 0 {
+		conf.ReadBufferSize = 200
+	}
+	if conf.WriteBufferSize == 0 {
+		conf.WriteBufferSize = 150
+	}
+	client := &Client{}
+	client.resultlock = sync.Map{}
+	client.businessEventHandler = conf.EventHandler
+	client.canHandleEvent = conf.CanHandleEvents
+	client.messageQueue = make(chan *Packet, conf.ReadBufferSize) //接受消息
+	client.writeQueue = make(chan RawData, conf.WriteBufferSize)  //写入并发
+	client.myid = UUID(8)
+	//消息读取泵
+	go func(q <-chan *Packet, parseEvent func(*Packet)) {
+		for {
+			select {
+			case msg := <-q:
+				go parseEvent(msg)
+			}
+		}
+	}(client.messageQueue, client.baseEventHandle)
+	//消息写入泵
+	go client.writeDump(client.writeQueue)
+	//创建链接
+	var connectedcallbackfn = func() {
+		client.WriteObj("", "", EVENT_REQUEST_JOIN, SENDTO_TYPE_P2P, map[string]interface{}{
+			"handler": conf.CanHandleEvents,
+			"name":    conf.ClientName,
+			"myid":    client.myid,
+		})
+	}
+	//使用tls
+	if conf.UseTls {
+		cert, err := tls.LoadX509KeyPair(conf.TlsCertFile, conf.TlsKeyFile)
+		if err != nil {
+			Log.Fail.Fatalln(err)
+		}
+		config := &tls.Config{
+			InsecureSkipVerify: true,
+			Certificates:       []tls.Certificate{cert},
+		}
+		client.conn = NewTlsClient("tcp4", conf.MsgServerAddr,
+			math.MaxInt32, 2*time.Second, config, connectedcallbackfn)
+
+	} else {
+		client.conn = NewTCPClient("tcp4", conf.MsgServerAddr,
+			math.MaxInt32, 2*time.Second, connectedcallbackfn)
+	}
+	for {
+		if err := client.conn.Dial(); err != nil {
+			Log.Debug.Println("尝试等待连接到消息总线")
+			time.Sleep(5 * time.Second)
+		} else {
+			break
+		}
 	}
+	go client.readDump(client.messageQueue)
+	//发起加入
+	client.WriteObj("", "", EVENT_REQUEST_JOIN, SENDTO_TYPE_P2P, map[string]interface{}{
+		"handler": conf.CanHandleEvents,
+		"name":    conf.ClientName,
+		"myid":    client.myid,
+	})
+
+	return client, nil
 }
 
 //
@@ -70,43 +140,17 @@ func (client *Client) Call(to, msgid string, event, sendtotype int, obj interfac
 	}
 }
 
-func (client *Client) GetMyclient() string {
-	return client.myid
-}
-
 //底层事件处理
 func (client *Client) baseEventHandle(msg *Packet) {
-	printlog(".")
 	event := int(msg.Event)
 	switch event {
-	case EVENT_RETURN_MACHINE_ID: //服务端分配id，发布我能处理的时间
-		printlog("*", string(msg.GetBusinessData()))
-		client.myid = string(msg.GetBusinessData())
-		client.conn.MyId = client.myid
-
-		if client.oldmyid != "" {
-			client.WriteObj("", "", EVENT_REMOVE_CLIENT, SENDTO_TYPE_P2P, []byte(client.oldmyid))
-			client.oldmyid = ""
-		}
-
-		bs := make([]byte, 0)
-		for _, v := range client.canHandleEvent {
-			bs = append(bs, Int2Byte(int32(v))...)
-		}
-		client.WriteObj("", "", EVENT_PUBLISH_MYSERVICES, SENDTO_TYPE_P2P, bs)
-		client.WriteObj("", "", EVENT_UPDATE_MYNAME, SENDTO_TYPE_P2P, []byte(client.myname))
-		if client.OnConnectSuccess != nil { //调用成功回调
-			go client.OnConnectSuccess()
-		}
 	case EVENT_REQUEST_HEARTBEAT: //请求的心跳，回应心跳请求
-		client.WriteObj("", "", EVENT_RETURN_HEARTBEAT, SENDTO_TYPE_P2P, msg.GetBusinessData())
+		client.WriteObj("", "", EVENT_RETURN_HEARTBEAT, SENDTO_TYPE_P2P, msg.Raw)
 		client.lastCheckHeart = time.Now().Unix()
-	case EVENT_SYSTEM_COMMAND: //系统控制指令
-		go processSysCommand(client, msg.GetBusinessData())
-	case EVENT_RECIVE_CALLBACK: //回调处理
+	case EVENT_RETURN: //回调处理，一般用于同步操作
 		if v, ok := client.resultlock.Load(msg.Msgid); ok {
 			tmp := v.(chan []byte)
-			tmp <- msg.GetBusinessData()
+			tmp <- msg.Raw
 			return
 		}
 		fallthrough
@@ -121,18 +165,18 @@ func (client *Client) read4socket() (*Packet, error) {
 	defer Catch()
 	buffer := make([]byte, 4)
 	if _, err := io.ReadFull(client.conn, buffer); err != nil {
-		printlog("?_readFullerr", err)
+		Log.Debug.Println("读取包头失败", err)
 		return nil, errors.New("read packet error")
 	}
 	size := int32(binary.BigEndian.Uint32(buffer))
-	if size < 32 || size > 16384000 {
-		printlog("?_sizeerror[", size, "]")
+	if size < PACKET_MIN_LEN || size > PACKET_MAX_LEN {
+		Log.Debug.Println("数据包长度越界[", size, "]")
 		return nil, errors.New("read packet error")
 	}
 	var buf []byte = make([]byte, size)
 	readlen, err := io.ReadFull(client.conn, buf)
 	if err != nil || int32(readlen) != size {
-		printlog("?_readpacketerror", err)
+		Log.Debug.Println("读取包主体数据失败", err)
 		return nil, errors.New("read packet error")
 	}
 	//
@@ -184,122 +228,3 @@ func (client *Client) writeDump(p <-chan RawData) {
 		}
 	}
 }
-
-//启动客户端
-//@Deprecated 此方法已过期
-func StartClient(fn func(*Packet),
-	addr string, myname string,
-	handleevent []int) (*Client, error) {
-	//
-	client := &Client{}
-	client.resultlock = sync.Map{} //make(map[string]chan []byte)
-	client.businessEventHandler = fn
-	client.canHandleEvent = handleevent
-	client.myname = myname
-	client.messageQueue = make(chan *Packet, 500) //接受消息
-	client.writeQueue = make(chan RawData, 500)   //写入并发
-
-	go func(q <-chan *Packet, parseEvent func(*Packet)) {
-		for {
-			select {
-			case msg := <-q:
-				go parseEvent(msg)
-			}
-		}
-	}(client.messageQueue, client.baseEventHandle)
-	go client.writeDump(client.writeQueue)
-	//创建链接
-	var err error
-	for client.conn, err = Dial("tcp4", addr); err != nil; client.conn, err = Dial("tcp4", addr) {
-		printlog("?")
-		time.Sleep(5 * time.Second)
-	}
-	client.conn.SetMaxRetries(math.MaxInt32)      //重试次数
-	client.conn.SetRetryInterval(2 * time.Second) //重试间隔30秒
-	client.conn.OnRequestConnect = func() {
-		client.oldmyid = client.myid
-		client.myid = ""
-	}
-	go client.readDump(client.messageQueue)
-	return client, nil
-}
-
-//解析处理控制指令
-func processSysCommand(client *Client, data []byte) {
-	tmp := map[string]interface{}{}
-	err := json.Unmarshal(data, &tmp)
-	if err != nil {
-		return
-	}
-	cmd := tmp["act"].(string)
-	switch cmd {
-	case "pass": //暂停服务
-		t := int(tmp["passtime"].(float64))
-		client.WriteObj("", "", EVENT_PUBLISH_MYSERVICES, SENDTO_TYPE_P2P, []byte{})
-		time.Sleep(time.Duration(t) * time.Second)
-		bs := make([]byte, 0)
-		for _, v := range client.canHandleEvent {
-			bs = append(bs, Int2Byte(int32(v))...)
-		}
-		client.WriteObj("", "", EVENT_PUBLISH_MYSERVICES, SENDTO_TYPE_P2P, bs)
-	case "reconnect": //重新连接
-
-	case "quit": //退出
-		client.WriteObj("", "", EVENT_BYE, SENDTO_TYPE_P2P, []byte{})
-		os.Exit(0)
-	}
-}
-
-//启动客户端
-func NewClient(conf *ClientConfig) (*Client, error) {
-	if conf.MsgServerAddr == "" {
-		return nil, errors.New("配置项缺失(MsgServerAddr)")
-	} else if conf.EventHandler == nil {
-		return nil, errors.New("配置项缺失(EventHandler)")
-	}
-	if conf.ReadBufferSize == 0 {
-		conf.ReadBufferSize = 200
-	}
-	if conf.WriteBufferSize == 0 {
-		conf.WriteBufferSize = 150
-	}
-	client := &Client{}
-	client.resultlock = sync.Map{} // make(map[string]chan []byte)
-	client.businessEventHandler = conf.EventHandler
-	client.canHandleEvent = conf.CanHandleEvents
-	client.myname = conf.ClientName
-	client.messageQueue = make(chan *Packet, conf.ReadBufferSize) //接受消息
-	client.writeQueue = make(chan RawData, conf.WriteBufferSize)  //写入并发
-	go func(q <-chan *Packet, parseEvent func(*Packet)) {
-		for {
-			select {
-			case msg := <-q:
-				go parseEvent(msg)
-			}
-		}
-	}(client.messageQueue, client.baseEventHandle)
-	go client.writeDump(client.writeQueue)
-	//创建链接
-	var err error
-	for client.conn, err = Dial("tcp4", conf.MsgServerAddr); err != nil; client.conn, err = Dial("tcp4", conf.MsgServerAddr) {
-		printlog("?")
-		time.Sleep(5 * time.Second)
-	}
-	client.conn.SetMaxRetries(math.MaxInt32)      //重试次数
-	client.conn.SetRetryInterval(2 * time.Second) //重试间隔30秒
-	//事件绑定
-	client.conn.OnRequestConnect = func() {
-		client.oldmyid = client.myid
-		client.myid = ""
-		if conf.OnRequestConnect != nil {
-			conf.OnRequestConnect()
-		}
-	}
-	client.conn.OnConnectSuccess = func() {
-		if conf.OnConnectSuccess != nil {
-			conf.OnConnectSuccess()
-		}
-	}
-	go client.readDump(client.messageQueue)
-	return client, nil
-}

src/mfw/util/client.go1.5

@@ -1,301 +0,0 @@
-//客户端封装,支持短线重连接，支持客户端检测
-package util
-
-import (
-	"encoding/binary"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"io"
-	"math"
-	"os"
-	"time"
-)
-
-var IsHiddenLog bool
-
-//封装，可以链接多个消息总线
-type Client struct {
-	conn                 *TCPClient
-	myid                 string //
-	oldmyid              string
-	businessEventHandler func(*Packet)
-	canHandleEvent       []int
-	lastCheckHeart       int64 //最后一次心跳检测
-	messageQueue         chan *Packet
-	writeQueue           chan RawData
-	resultlock           map[string]chan []byte
-	myname               string
-	OnConnectSuccess     func()
-}
-
-//客户端配置
-type ClientConfig struct {
-	ClientName       string          //名称
-	EventHandler     func(p *Packet) //事件处理函数
-	MsgServerAddr    string          //消息总线地址
-	CanHandleEvents  []int           //我可以处理的事件列表
-	OnRequestConnect func()          //请求重连的事件处理函数
-	OnConnectSuccess func()          //重连成功的时间处理函数
-	ReadBufferSize   int             //读缓冲区大小
-	WriteBufferSize  int             //写缓冲区大小
-}
-
-//
-func printlog(p ...interface{}) {
-	if !IsHiddenLog {
-		fmt.Print(p...)
-	}
-}
-
-//
-func (client *Client) WriteObj(to, msgid string, event, sendtotype int, obj interface{}) {
-	client.writeQueue <- RawData(Enpacket(client.myid, to, msgid, event, sendtotype, obj))
-}
-
-//同步调用方法
-func (client *Client) Call(to, msgid string, event, sendtotype int, obj interface{}, calltimeout int64) ([]byte, error) {
-	client.resultlock[msgid] = make(chan []byte)
-	client.writeQueue <- RawData(Enpacket(client.myid, to, msgid, event, sendtotype, obj))
-	select {
-	case <-time.After(time.Duration(calltimeout) * time.Second):
-		delete(client.resultlock, msgid)
-		return nil, errors.New("timeout")
-	case ret := <-client.resultlock[msgid]:
-		delete(client.resultlock, msgid)
-		return ret, nil
-	}
-}
-
-func (client *Client) GetMyclient() string {
-	return client.myid
-}
-
-//底层事件处理
-func (client *Client) baseEventHandle(msg *Packet) {
-	printlog(".")
-	event := int(msg.Event)
-	switch event {
-	case EVENT_RETURN_MACHINE_ID: //服务端分配id，发布我能处理的时间
-		printlog("*", string(msg.GetBusinessData()))
-		client.myid = string(msg.GetBusinessData())
-		client.conn.MyId = client.myid
-
-		if client.oldmyid != "" {
-			client.WriteObj("", "", EVENT_REMOVE_CLIENT, SENDTO_TYPE_P2P, []byte(client.oldmyid))
-			client.oldmyid = ""
-		}
-
-		bs := make([]byte, 0)
-		for _, v := range client.canHandleEvent {
-			bs = append(bs, Int2Byte(int32(v))...)
-		}
-		client.WriteObj("", "", EVENT_PUBLISH_MYSERVICES, SENDTO_TYPE_P2P, bs)
-		client.WriteObj("", "", EVENT_UPDATE_MYNAME, SENDTO_TYPE_P2P, []byte(client.myname))
-		if client.OnConnectSuccess != nil { //调用成功回调
-			go client.OnConnectSuccess()
-		}
-	case EVENT_REQUEST_HEARTBEAT: //请求的心跳，回应心跳请求
-		client.WriteObj("", "", EVENT_RETURN_HEARTBEAT, SENDTO_TYPE_P2P, msg.GetBusinessData())
-		client.lastCheckHeart = time.Now().Unix()
-	case EVENT_SYSTEM_COMMAND: //系统控制指令
-		go processSysCommand(client, msg.GetBusinessData())
-	case EVENT_RECIVE_CALLBACK: //回调处理
-		if v, ok := client.resultlock[msg.Msgid]; ok {
-			v <- msg.GetBusinessData()
-			return
-		}
-		fallthrough
-	default: //业务处理
-		go client.businessEventHandler(msg)
-	}
-
-}
-
-//从套接字读取数据
-func (client *Client) read4socket() (*Packet, error) {
-	defer Catch()
-	buffer := make([]byte, 4)
-	if _, err := io.ReadFull(client.conn, buffer); err != nil {
-		printlog("?_readFullerr", err)
-		return nil, errors.New("read packet error")
-	}
-	size := int32(binary.BigEndian.Uint32(buffer))
-	if size < 32 || size > 16384000 {
-		printlog("?_sizeerror[", size, "]")
-		return nil, errors.New("read packet error")
-	}
-	var buf []byte = make([]byte, size)
-	readlen, err := io.ReadFull(client.conn, buf)
-	if err != nil || int32(readlen) != size {
-		printlog("?_readpacketerror", err)
-		return nil, errors.New("read packet error")
-	}
-	//
-	raw := append(Int2Byte(size), buf...)
-	//解析包
-	packet := &Packet{Length: size,
-		From:       string(buf[:8]),
-		To:         string(buf[8:16]),
-		Msgid:      string(buf[16:24]),
-		Event:      Byte2Int(buf[24:28]),
-		SentToType: Byte2Int(buf[28:32]),
-		Raw:        raw,
-	}
-	return packet, nil
-}
-
-//读数据到缓冲区
-func (client *Client) readDump(queue chan<- *Packet) {
-	defer Catch()
-	for {
-		if p, err := client.read4socket(); err == nil && p != nil {
-			client.lastCheckHeart = time.Now().Unix()
-			queue <- p
-		} else {
-			time.Sleep(4 * time.Second)
-		}
-	}
-}
-
-//写数据到套接字
-func (client *Client) write2socket(msg RawData) (ret bool) {
-	defer Catch()
-	_, err := client.conn.Write([]byte(msg))
-	if err == nil {
-		client.lastCheckHeart = time.Now().Unix()
-		return true
-	}
-	return false
-}
-
-//写数据到缓冲区
-func (client *Client) writeDump(p <-chan RawData) {
-	defer Catch()
-	for {
-		select {
-		case msg := <-p:
-			//写数据，直到写入成功
-			client.write2socket(msg)
-		}
-	}
-}
-
-//启动客户端
-//@Deprecated 此方法已过期
-func StartClient(fn func(*Packet),
-	addr string, myname string,
-	handleevent []int) (*Client, error) {
-	//
-	client := &Client{}
-	client.resultlock = make(map[string]chan []byte)
-	client.businessEventHandler = fn
-	client.canHandleEvent = handleevent
-	client.myname = myname
-	client.messageQueue = make(chan *Packet, 500) //接受消息
-	client.writeQueue = make(chan RawData, 500)   //写入并发
-
-	go func(q <-chan *Packet, parseEvent func(*Packet)) {
-		for {
-			select {
-			case msg := <-q:
-				go parseEvent(msg)
-			}
-		}
-	}(client.messageQueue, client.baseEventHandle)
-	go client.writeDump(client.writeQueue)
-	//创建链接
-	var err error
-	for client.conn, err = Dial("tcp4", addr); err != nil; client.conn, err = Dial("tcp4", addr) {
-		printlog("?")
-		time.Sleep(5 * time.Second)
-	}
-	client.conn.SetMaxRetries(math.MaxInt32)      //重试次数
-	client.conn.SetRetryInterval(2 * time.Second) //重试间隔30秒
-	client.conn.OnRequestConnect = func() {
-		client.oldmyid = client.myid
-		client.myid = ""
-	}
-	go client.readDump(client.messageQueue)
-	return client, nil
-}
-
-//解析处理控制指令
-func processSysCommand(client *Client, data []byte) {
-	tmp := map[string]interface{}{}
-	err := json.Unmarshal(data, &tmp)
-	if err != nil {
-		return
-	}
-	cmd := tmp["act"].(string)
-	switch cmd {
-	case "pass": //暂停服务
-		t := int(tmp["passtime"].(float64))
-		client.WriteObj("", "", EVENT_PUBLISH_MYSERVICES, SENDTO_TYPE_P2P, []byte{})
-		time.Sleep(time.Duration(t) * time.Second)
-		bs := make([]byte, 0)
-		for _, v := range client.canHandleEvent {
-			bs = append(bs, Int2Byte(int32(v))...)
-		}
-		client.WriteObj("", "", EVENT_PUBLISH_MYSERVICES, SENDTO_TYPE_P2P, bs)
-	case "reconnect": //重新连接
-
-	case "quit": //退出
-		client.WriteObj("", "", EVENT_BYE, SENDTO_TYPE_P2P, []byte{})
-		os.Exit(0)
-	}
-}
-
-//启动客户端
-func NewClient(conf *ClientConfig) (*Client, error) {
-	if conf.MsgServerAddr == "" {
-		return nil, errors.New("配置项缺失(MsgServerAddr)")
-	} else if conf.EventHandler == nil {
-		return nil, errors.New("配置项缺失(EventHandler)")
-	}
-	if conf.ReadBufferSize == 0 {
-		conf.ReadBufferSize = 200
-	}
-	if conf.WriteBufferSize == 0 {
-		conf.WriteBufferSize = 150
-	}
-	client := &Client{}
-	client.resultlock = make(map[string]chan []byte)
-	client.businessEventHandler = conf.EventHandler
-	client.canHandleEvent = conf.CanHandleEvents
-	client.myname = conf.ClientName
-	client.messageQueue = make(chan *Packet, conf.ReadBufferSize) //接受消息
-	client.writeQueue = make(chan RawData, conf.WriteBufferSize)  //写入并发
-	go func(q <-chan *Packet, parseEvent func(*Packet)) {
-		for {
-			select {
-			case msg := <-q:
-				go parseEvent(msg)
-			}
-		}
-	}(client.messageQueue, client.baseEventHandle)
-	go client.writeDump(client.writeQueue)
-	//创建链接
-	var err error
-	for client.conn, err = Dial("tcp4", conf.MsgServerAddr); err != nil; client.conn, err = Dial("tcp4", conf.MsgServerAddr) {
-		printlog("?")
-		time.Sleep(5 * time.Second)
-	}
-	client.conn.SetMaxRetries(math.MaxInt32)      //重试次数
-	client.conn.SetRetryInterval(2 * time.Second) //重试间隔30秒
-	//事件绑定
-	client.conn.OnRequestConnect = func() {
-		client.oldmyid = client.myid
-		client.myid = ""
-		if conf.OnRequestConnect != nil {
-			conf.OnRequestConnect()
-		}
-	}
-	client.conn.OnConnectSuccess = func() {
-		if conf.OnConnectSuccess != nil {
-			conf.OnConnectSuccess()
-		}
-	}
-	go client.readDump(client.messageQueue)
-	return client, nil
-}

src/mfw/util/event.go

@@ -2,41 +2,21 @@ package util
 
 //只有事件，发送类型
 const (
-	EVENT_RETURN_MACHINE_ID    = iota
-	EVENT_REQUEST_HEARTBEAT    //心跳
-	EVENT_RETURN_HEARTBEAT     //
-	EVENT_PUBLISH_MYSERVICES   //发布我的服务
-	EVENT_REQUEST_SPIDER_STATE //获取爬虫状态
-	EVENT_RECIVE_SPIDER_STATE  //接受爬虫状态
-	EVENT_RECIVE_CALLBACK      //调用返回，用于调用服务需要同步返回值
-	EVENT_VIEWALL_SERVICE      //查看所有的客户端
-	EVENT_REMOVE_CLIENT        //删除客户端
-	EVENT_UPDATE_MYNAME        //更新客户端名称
-	EVENT_SYSTEM_COMMAND       //系统控制指令，每个客户端默认都会实现
-	EVENT_BYE                  //客户端主动断开
-	EVENT_GET_ALLDOWNLOADER    //取所有的下载器
-	//---------
-	SERVICE_DOWNLOAD             = 7070 //下载服务
-	SERVICE_GETPROXY             = 7071 //获取代理
-	SERVICE_GETNAME              = 7072 //获取企业名录
-	SERVICE_SPIDER_ECPS          = 7073 //公示爬虫分发服务
-	SERVICE_DISTINGUISH          = 7074 //人工识别验证码
-	SERVICE_DOWNLOAD_DELETE_NODE = 7075 //删除节点
-	SERVICE_DOWNLOAD_APPEND_NODE = 7076 //追加节点
-	SERVICE_REPORT_SAVE          = 7077 //保存报表数据
-	SERVICE_INVNAME_ANALYSIS     = 7078 //分析股东名录
-	SERVICE_YCML_SAVE            = 7079 //异常名录保存入内存数据库
-	SERVICE_YCML_NOTICE          = 7080 //异常名录下载完成通知
-	SERVICE_ECPS_INC             = 7081 //公示增量数据保存
-	SERVICE_OFFICE_ANALYSIS      = 7082 //word pdf excel文件解析服务 由java提供的一个服务
+	EVENT_REQUEST_JOIN      = iota
+	EVENT_REQUEST_HEARTBEAT //心跳
 
-	SERVICE_LOG_CHECK = 7200 //检测日志记录
+	EVENT_RETURN_MACHINE_ID  //
+	EVENT_RETURN_HEARTBEAT   //
+	EVENT_PUBLISH_MYSERVICES //发布我的服务
+	EVENT_RETURN             //调用返回，用于调用服务需要同步返回值
+	EVENT_VIEWALL_CLIENT     //查看所有的客户端
+	EVENT_REMOVE_CLIENT      //删除客户端
+	EVENT_BYE                //客户端主动断开
 
 	//-------发送方式----------------
 	SENDTO_TYPE_RAND_RECIVER = 0 //发送给任一服务接收者，默认是这种模式
 	SENDTO_TYPE_ALL          = 1 //发送给所有客户端
 	SENDTO_TYPE_ALL_RECIVER  = 2 //发送给所有指定服务接收者
 	SENDTO_TYPE_P2P          = 3 //发送给指定客户端
-
-	SENDTO_TYPE_P2P_BYNAME = 4 //发送给指定客户，通过指定客户的名称
+	SENDTO_TYPE_P2P_BYNAME   = 4 //发送给指定客户，通过指定客户的名称
 )

src/mfw/util/key.go

@@ -3,28 +3,15 @@ package util
 import (
 	"crypto/rand"
 	"encoding/hex"
-	//	"encoding/hex"
-	//"fmt"
-	mr "math/rand"
 )
 
 const (
 	KEY = "()*+-_=~`!@#$%^&*{}[]|',.<>?abcdefghhijklmnopqrstuvwxyz0123456789ABCDEFJHIJKLMNOPQRSTUVWXYZ"
 )
 
-//
+//保证系统ID唯一
 func UUID(length int) string {
 	tmp := make([]byte, length>>1)
 	rand.Read(tmp)
 	return hex.EncodeToString(tmp)
 }
-
-//
-func OldUUID(length int) string {
-	ret := ""
-	for i := 0; i < length; i++ {
-		pos := mr.Intn(90)
-		ret += KEY[pos : pos+1]
-	}
-	return ret
-}

src/mfw/util/log.go

@@ -0,0 +1,143 @@
+/**
+日志工具,
+可以用NewLogger自定义，
+使用Log时，默认在当前目录创建logs目录
+*/
+package util
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"log"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+)
+
+const (
+	LOG_LEVEL_INFO = iota
+	LOG_LEVEL_DEBUG
+	LOG_LEVEL_WARING
+	LOG_LEVEL_ERROR
+	LOG_LEVEL_FAIL
+)
+
+type Logger struct {
+	Info      *log.Logger
+	Debug     *log.Logger
+	Error     *log.Logger
+	Waring    *log.Logger
+	Fail      *log.Logger
+	logpath   string
+	sizelimit int64
+}
+
+//实现io.writer接口
+type logwriter struct {
+	logpath     string
+	sizelimt    int64
+	currentsize int64
+	mux         sync.RWMutex
+	filewriter  *os.File
+}
+
+//
+func (lw *logwriter) Write(p []byte) (int, error) {
+	lw.mux.Lock()
+	if lw.filewriter == nil {
+		lw.filewriter, _ = os.OpenFile(lw.makeLogFile(lw.logpath), os.O_CREATE|os.O_TRUNC|os.O_SYNC|os.O_RDWR, 0777)
+	} else if lw.currentsize > lw.sizelimt {
+		lw.filewriter.Close()
+		lw.filewriter, _ = os.OpenFile(lw.makeLogFile(lw.logpath), os.O_CREATE|os.O_TRUNC|os.O_SYNC|os.O_RDWR, 0777)
+		lw.currentsize = 0
+	}
+	lw.mux.Unlock()
+	lw.currentsize = lw.currentsize + int64(len(p))
+	return lw.filewriter.Write(p)
+}
+
+//生成不重复的文件名
+func (lw *logwriter) makeLogFile(logpath string) string {
+	if _, err := os.Stat(logpath); err != nil {
+		os.MkdirAll(logpath, 0777)
+	}
+	var log_file_path string
+	for i := 0; i < 1000; i++ {
+		file_name := fmt.Sprintf("%s%clog_%s_%03d.log", logpath, filepath.Separator, time.Now().Format("2006_01_02"), i)
+		if _, err := os.Stat(file_name); err != nil {
+			log_file_path = file_name
+			break
+		}
+	}
+	return log_file_path
+}
+
+//
+func NewLogger(logpath string, level int, out2stdout bool, filesizelimit int64) *Logger {
+	log_flag := log.Ldate | log.Ltime | log.Lshortfile
+	logger := &Logger{
+		logpath:   logpath,
+		sizelimit: filesizelimit,
+		Info:      log.New(ioutil.Discard, "[info]", log_flag),
+		Debug:     log.New(ioutil.Discard, "[debug]", log_flag),
+		Error:     log.New(ioutil.Discard, "[error]", log_flag),
+		Waring:    log.New(ioutil.Discard, "[waring]", log_flag),
+		Fail:      log.New(ioutil.Discard, "[fail]", log_flag),
+	}
+	logger.SetLevel(level, out2stdout)
+	return logger
+}
+
+//
+func (l *Logger) SetLevel(level int, out2stdout bool) {
+	var lw io.Writer
+	if out2stdout {
+		lw = io.MultiWriter(&logwriter{logpath: l.logpath,
+			sizelimt:    l.sizelimit,
+			currentsize: 0,
+		}, os.Stdout)
+	} else {
+		lw = &logwriter{logpath: l.logpath,
+			sizelimt:    l.sizelimit,
+			currentsize: 0,
+		}
+	}
+
+	if LOG_LEVEL_INFO == level {
+		l.Info.SetOutput(lw)
+		l.Debug.SetOutput(lw)
+		l.Error.SetOutput(lw)
+		l.Waring.SetOutput(lw)
+		l.Fail.SetOutput(lw)
+	} else if LOG_LEVEL_DEBUG == level {
+		l.Info.SetOutput(ioutil.Discard)
+		l.Debug.SetOutput(lw)
+		l.Error.SetOutput(lw)
+		l.Waring.SetOutput(lw)
+		l.Fail.SetOutput(lw)
+	} else if LOG_LEVEL_ERROR == level {
+		l.Info.SetOutput(ioutil.Discard)
+		l.Debug.SetOutput(ioutil.Discard)
+		l.Error.SetOutput(lw)
+		l.Waring.SetOutput(lw)
+		l.Fail.SetOutput(lw)
+	} else if LOG_LEVEL_WARING == level {
+		l.Info.SetOutput(ioutil.Discard)
+		l.Debug.SetOutput(ioutil.Discard)
+		l.Error.SetOutput(ioutil.Discard)
+		l.Waring.SetOutput(lw)
+		l.Fail.SetOutput(lw)
+	} else if LOG_LEVEL_FAIL == level {
+		l.Info.SetOutput(ioutil.Discard)
+		l.Debug.SetOutput(ioutil.Discard)
+		l.Error.SetOutput(ioutil.Discard)
+		l.Waring.SetOutput(ioutil.Discard)
+		l.Fail.SetOutput(lw)
+	}
+
+}
+
+//默认日志
+var Log = NewLogger("./logs", LOG_LEVEL_INFO, true, 1024*1024*50)

src/mfw/util/log_test.go

@@ -0,0 +1,21 @@
+package util
+
+import (
+	"testing"
+)
+
+//
+func BenchmarkLog(b *testing.B) {
+	//mylog := NewLogger("../../../logs", LOG_LEVEL_DEBUG, 1024*1024*5)
+	Log.SetLevel(LOG_LEVEL_ERROR)
+	for i := 0; i < b.N; i++ {
+		Log.Debug.Println("hello world")
+	}
+}
+
+func TestLog(b *testing.T) {
+	mylog := NewLogger("../../../logs", LOG_LEVEL_DEBUG, 1024*50)
+	for i := 0; i < 5; i++ {
+		mylog.Debug.Println("hello world")
+	}
+}

src/mfw/util/protocol.go

@@ -2,29 +2,29 @@ package util
 
 //底层网络通信协议
 import (
-	"bufio"
 	"bytes"
 	"encoding/binary"
 	"encoding/json"
-	"fmt"
-	"io"
+	"errors"
+	"hash/crc32"
 	"net"
 )
 
 const (
-	EMPTY_ADDR = "00000000"
+	PACKET_MAX_LEN = 16384000
+	PACKET_MIN_LEN = 36
 )
 
-//读取
-type Reader struct {
-	conn   net.Conn
-	reader *bufio.Reader
-	buffer [4]byte
-}
+var (
+	EMPTY_ADDR  = "00000000"
+	PACKET_HEAD = []byte("HEAD")
+	crc         = crc32.NewIEEE()
+)
 
 //数据包，实际上就是协议定义
 type Packet struct {
 	//头开始
+	Head       []byte //头标志
 	Length     int32  //4bit 包长度
 	From       string //8bit
 	To         string //8bit
@@ -32,92 +32,14 @@ type Packet struct {
 	Event      int32  //4bit
 	SentToType int32  //4bit
 	//头结束
-	Raw []byte //数据包，头+数据
-}
-
-//返回
-func (p *Packet) GetBusinessData() []byte {
-	return p.Raw[36:]
+	Raw  []byte   //数据区域
+	Sum  int32    //crc32校验和 （从长度标志到数据区域）
+	Conn net.Conn //
 }
 
 //
 type RawData []byte //流数据
 
-//
-func NewReader(c net.Conn) *Reader {
-	return &Reader{
-		conn:   c,
-		reader: bufio.NewReader(c),
-	}
-}
-
-//读取头部
-func (p *Reader) readHeader() (int32, error) {
-	buf := p.buffer[:4]
-	if _, err := io.ReadFull(p.reader, buf); err != nil {
-		return 0, err
-	}
-	size := int32(binary.BigEndian.Uint32(buf))
-	if size < 32 || size > 16384000 {
-		return 0, fmt.Errorf("Incorrect frame size (%d)", size)
-	}
-	return size, nil
-}
-
-//读取完整一帧数据，防止粘包
-func (p *Reader) readFrame(size int32) (*Packet, error) {
-	var buf []byte
-	if int(size) <= len(p.buffer) {
-		buf = p.buffer[0:size]
-	} else {
-		buf = make([]byte, size)
-	}
-	_, err := io.ReadFull(p.reader, buf)
-	raw := append(Int2Byte(size), buf...)
-	if err == nil {
-		//解析包
-		packet := &Packet{
-			Length:     size,
-			From:       string(buf[:8]),
-			To:         string(buf[8:16]),
-			Msgid:      string(buf[16:24]),
-			Event:      Byte2Int(buf[24:28]),
-			SentToType: Byte2Int(buf[28:32]),
-			//Data:       buf[32:],
-			Raw: raw,
-		}
-		return packet, nil
-	}
-	return nil, err
-}
-
-//读取数据写入队列
-func forwardMessage(c net.Conn, queue chan<- *Packet) {
-	defer c.Close()
-	logReader := NewReader(c)
-	for {
-		size, err := logReader.readHeader()
-		if err != nil {
-			break
-		}
-		packet, err := logReader.readFrame(size)
-		if err != nil {
-			break
-		}
-		queue <- packet
-	}
-}
-
-//从队列中读取数据，定期处理
-func processMsg(q <-chan *Packet, parseEvent func(*Packet)) {
-	for {
-		select {
-		case msg := <-q:
-			go parseEvent(msg)
-		}
-	}
-}
-
 //封包，输出直接可以写到流中的数据包
 func Enpacket(from, to, msgid string, event, sendtotype int, obj interface{}) []byte {
 	if len(from) != 8 {
@@ -129,8 +51,8 @@ func Enpacket(from, to, msgid string, event, sendtotype int, obj interface{}) []
 	if len(msgid) != 8 {
 		msgid = EMPTY_ADDR
 	}
-	var ret []byte
-	var bs []byte
+	var body, ret, bs []byte
+
 	if v, ok := obj.([]byte); ok {
 		bs = v
 	} else if v, ok := obj.(string); ok {
@@ -138,14 +60,41 @@ func Enpacket(from, to, msgid string, event, sendtotype int, obj interface{}) []
 	} else {
 		bs, _ = json.Marshal(obj)
 	}
-	ret = append(Int2Byte(int32(32 + len(bs))))
-	ret = append(ret, []byte(from+to+msgid)...)
-	ret = append(ret, Int2Byte(int32(event))...)
-	ret = append(ret, Int2Byte(int32(sendtotype))...)
-	ret = append(ret, bs...)
+	body = append([]byte(from + to + msgid))
+	body = append(body, Int2Byte(int32(event))...)
+	body = append(body, Int2Byte(int32(sendtotype))...)
+	body = append(body, bs...)
+	crc.Write(body)
+	body = append(body, Uint322Byte(crc.Sum32())...)
+	ret = append(PACKET_HEAD)
+	ret = append(ret, Int2Byte(int32(36+len(bs)))...)
+	ret = append(ret, body...)
+	crc.Write(body)
+	crc.Reset()
 	return ret
 }
 
+//
+func Depacket(bs []byte) (*Packet, error) {
+	//校验
+	if bytes.Compare(bs[:4], PACKET_HEAD) != 0 {
+		return nil, errors.New("无效的包头")
+	}
+
+	//crc
+	o_sum := bs[len(bs)-4:]
+	crc.Write(bs[8 : len(bs)-4])
+	n_sum := Uint322Byte(crc.Sum32())
+	if bytes.Compare(o_sum, n_sum) != 0 {
+		return nil, errors.New("无效包校验")
+	}
+	//全部校验通过
+	return &Packet{
+		Head: PACKET_HEAD,
+		Raw:  bs[36 : len(bs)-4],
+	}, nil
+}
+
 //
 func Byte2Int(src []byte) int32 {
 	var ret int32
@@ -159,3 +108,10 @@ func Int2Byte(src int32) []byte {
 	binary.Write(buf, binary.BigEndian, src)
 	return buf.Bytes()
 }
+
+//
+func Uint322Byte(src uint32) []byte {
+	buf := bytes.NewBuffer([]byte{})
+	binary.Write(buf, binary.BigEndian, src)
+	return buf.Bytes()
+}

src/mfw/util/protocol_test.go

@@ -0,0 +1,19 @@
+package util
+
+import (
+	"encoding/hex"
+	"fmt"
+	"testing"
+)
+
+func TestEnpacket(t *testing.T) {
+	bs := Enpacket("", "", "", 2, 1, "1")
+	fmt.Printf("len %d   % X ", len(bs), bs)
+}
+
+func TestExpacket(t *testing.T) {
+	msg := "484541440000004530303030303030303030303030303030623961646562393900000000000000037B2268616E646C6572223A5B312C322C335D2C226E616D65223A2274657374227D6E9C8595"
+	bs, _ := hex.DecodeString(msg)
+	p, err := Depacket(bs)
+	fmt.Println(p, err)
+}

src/mfw/util/server.go

@@ -2,33 +2,159 @@
 package util
 
 import (
-	"fmt"
+	"bufio"
+	"bytes"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/binary"
+	"errors"
+	"io"
+	"io/ioutil"
 	"net"
+	"strconv"
 )
 
+type ServerConfig struct {
+	ProcessEvent func(*Packet)
+	ListenAddr   string
+	UseTls       bool
+	TlsCA        string
+	TlsCert      string
+	TlsKey       string
+}
+
 //启动服务端
-func StartServer(parseEvent func(*Packet),
-	processconnection func(conn net.Conn),
-	addr string) {
-	netListen, err := net.Listen("tcp", addr)
-	if err != nil {
-		fmt.Println(err.Error())
+func StartServer(conf *ServerConfig) {
+	var err error
+	var netListen net.Listener
+	if conf.UseTls {
+		buf, err := ioutil.ReadFile(conf.TlsCA)
+		if err != nil {
+			Log.Fail.Fatalln(err.Error())
+		}
+		pool := x509.NewCertPool()
+		pool.AppendCertsFromPEM(buf)
+		cert, err := tls.LoadX509KeyPair(conf.TlsCert, conf.TlsKey)
+		if err != nil {
+			Log.Fail.Fatalln(err.Error())
+		}
+
+		tlsConfig := &tls.Config{
+			Certificates: []tls.Certificate{cert},
+			ClientAuth:   tls.RequireAndVerifyClientCert,
+			ClientCAs:    pool,
+		}
+		netListen, err = tls.Listen("tcp", conf.ListenAddr, tlsConfig)
+		if err != nil {
+			Log.Fail.Fatalln(err.Error())
+		}
+	} else {
+		netListen, err = net.Listen("tcp", conf.ListenAddr)
+		if err != nil {
+			Log.Fail.Fatalln(err.Error())
+		}
 	}
 	defer netListen.Close()
-	fmt.Println("等待连接，服务地址:", addr)
+	Log.Debug.Println("等待连接，服务地址", conf.ListenAddr)
 	//接受消息
 	messageQueue := make(chan *Packet, 5000) //并发1000
-	go processMsg(messageQueue, parseEvent)
+	go processMsg(messageQueue, conf.ProcessEvent)
 	for {
 		conn, err := netListen.Accept()
 		if err != nil {
 			continue
 		} else {
-			//告诉客户端ID，每次发消息都要携带自己ID
-			go func() {
-				processconnection(conn)
-				forwardMessage(conn, messageQueue)
-			}()
+			go forwardMessage(conn, messageQueue)
+
+		}
+	}
+}
+
+//读取
+type Reader struct {
+	conn   net.Conn
+	reader *bufio.Reader
+	buf    []byte
+}
+
+//
+func NewReader(c net.Conn) *Reader {
+	return &Reader{
+		conn:   c,
+		reader: bufio.NewReader(c),
+		buf:    make([]byte, 8),
+	}
+}
+
+//读取头部
+func (p *Reader) readHeader() (int32, error) {
+	if _, err := io.ReadFull(p.conn, p.buf); err != nil {
+		return 0, err
+	}
+	if bytes.Compare(p.buf[:4], PACKET_HEAD) != 0 {
+		return 0, errors.New("包头错误")
+	}
+	size := int32(binary.BigEndian.Uint32(p.buf[4:]))
+	if size < PACKET_MIN_LEN || size > PACKET_MAX_LEN {
+		return 0, errors.New("包长度越界[" + strconv.Itoa(int(size)) + "]")
+	}
+	return size, nil
+}
+
+//读取完整一帧数据，防止粘包
+func (p *Reader) readFrame(size int32) (*Packet, error) {
+	buf := make([]byte, size, size)
+	_, err := io.ReadFull(p.reader, buf)
+	if err == nil {
+		body := buf[:len(buf)-4]
+		o_sum := buf[len(buf)-4:]
+		crc.Write(body)
+		n_sum := Uint322Byte(crc.Sum32())
+		crc.Reset()
+		if bytes.Compare(o_sum, n_sum) != 0 {
+			return nil, errors.New("包文校验失败")
+		}
+		//解析包
+		packet := &Packet{
+			Length:     size,
+			From:       string(body[:8]),
+			To:         string(body[8:16]),
+			Msgid:      string(body[16:24]),
+			Event:      Byte2Int(body[24:28]),
+			SentToType: Byte2Int(body[28:32]),
+			Raw:        body[32:],
+			Conn:       p.conn,
+		}
+		return packet, nil
+	}
+	return nil, err
+}
+
+//读取数据写入队列
+func forwardMessage(c net.Conn, queue chan<- *Packet) {
+	defer c.Close()
+	logReader := NewReader(c)
+	for {
+		size, err := logReader.readHeader()
+		if err != nil {
+			Log.Debug.Println("读取头失败", err.Error())
+			break
+		}
+		packet, err := logReader.readFrame(size)
+		if err != nil {
+			Log.Debug.Println("读取包体失败", err.Error())
+			break
+		}
+		queue <- packet
+	}
+}
+
+//从队列中读取数据，定期处理
+func processMsg(q <-chan *Packet, parseEvent func(*Packet)) {
+	for {
+		select {
+		case msg := <-q:
+			go parseEvent(msg)
 		}
 	}
 }

src/mfw/util/tcpclient.go

@@ -1,7 +1,6 @@
 package util
 
 import (
-	"io"
 	"net"
 	"sync"
 	"sync/atomic"
@@ -22,13 +21,26 @@ type Error int
 //
 type TCPClient struct {
 	*net.TCPConn
-	MyId             string
-	lock             sync.RWMutex
-	status           int32
-	maxRetries       int
-	retryInterval    time.Duration
-	OnConnectSuccess func() //重连成功后的回调
-	OnRequestConnect func() //请求重连
+	lock          sync.RWMutex
+	status        int32
+	maxRetries    int
+	retryInterval time.Duration
+	network       string
+	serveraddr    string
+	onconnectedfn func()
+}
+
+//
+func NewTCPClient(network, serveraddr string, maxretries int, retryinterval time.Duration, connectedcallback func()) *TCPClient {
+	return &TCPClient{
+		network:       network,
+		serveraddr:    serveraddr,
+		lock:          sync.RWMutex{},
+		maxRetries:    maxretries,
+		status:        statusOffline,
+		retryInterval: retryinterval,
+		onconnectedfn: connectedcallback,
+	}
 }
 
 //
@@ -42,70 +54,18 @@ func (e Error) Error() string {
 }
 
 //
-func Dial(network, addr string) (*TCPClient, error) {
-	raddr, err := net.ResolveTCPAddr(network, addr)
+func (t *TCPClient) Dial() error {
+	raddr, err := net.ResolveTCPAddr(t.network, t.serveraddr)
 	if err != nil {
-		return nil, err
+		return err
 	}
-	return DialTCP(network, nil, raddr)
-}
-
-//
-func DialTCP(network string, laddr, raddr *net.TCPAddr) (*TCPClient, error) {
-	conn, err := net.DialTCP(network, laddr, raddr)
+	conn, err := net.DialTCP(t.network, nil, raddr)
 	if err != nil {
-		return nil, err
+		return err
 	}
-	return &TCPClient{
-		TCPConn:       conn,
-		lock:          sync.RWMutex{},
-		status:        0,
-		maxRetries:    10,
-		retryInterval: 10 * time.Millisecond,
-	}, nil
-}
-
-//建议用这个
-func DialTCPTimeout(network, raddr string, timeout time.Duration) (*TCPClient, error) {
-	conn, err := net.DialTimeout(network, raddr, timeout)
-	if err != nil {
-		return nil, err
-	}
-	return &TCPClient{
-		TCPConn:       conn.(*net.TCPConn),
-		lock:          sync.RWMutex{},
-		status:        0,
-		maxRetries:    10,
-		retryInterval: 10 * time.Millisecond,
-	}, nil
-}
-
-//设置最大尝试次数
-func (c *TCPClient) SetMaxRetries(maxRetries int) {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-	c.maxRetries = maxRetries
-}
-
-//
-func (c *TCPClient) GetMaxRetries() int {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-	return c.maxRetries
-}
-
-//重连间隔时间
-func (c *TCPClient) SetRetryInterval(retryInterval time.Duration) {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-	c.retryInterval = retryInterval
-}
-
-//
-func (c *TCPClient) GetRetryInterval() time.Duration {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-	return c.retryInterval
+	t.TCPConn = conn
+	t.status = statusOnline
+	return nil
 }
 
 //
@@ -113,10 +73,7 @@ func (c *TCPClient) reconnect() error {
 	if !atomic.CompareAndSwapInt32(&c.status, statusOffline, statusReconnecting) {
 		return nil
 	}
-	//
-	if c.OnRequestConnect != nil {
-		c.OnRequestConnect()
-	}
+
 	raddr := c.TCPConn.RemoteAddr()
 	conn, err := net.DialTCP(raddr.Network(), nil, raddr.(*net.TCPAddr))
 	if err != nil {
@@ -128,11 +85,12 @@ func (c *TCPClient) reconnect() error {
 			return err
 		}
 	}
+
 	c.TCPConn.Close()
-	c.TCPConn = conn
+	c.TCPConn = conn //tlscon.(*net.TCPConn)
 	atomic.StoreInt32(&c.status, statusOnline)
-	if c.OnConnectSuccess != nil {
-		c.OnConnectSuccess() //执行后续操作
+	if c.onconnectedfn != nil {
+		c.onconnectedfn() //执行后续操作
 	}
 	return nil
 }
@@ -169,53 +127,12 @@ func (c *TCPClient) Read(b []byte) (int, error) {
 	return -1, ErrMaxRetries
 }
 
-//
-func (c *TCPClient) ReadFrom(r io.Reader) (int64, error) {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-	for i := 0; i < c.maxRetries; i++ {
-		if atomic.LoadInt32(&c.status) == statusOnline {
-			n, err := c.TCPConn.ReadFrom(r)
-			if err == nil {
-				return n, err
-			}
-			switch err.(type) {
-			case *net.OpError:
-				atomic.StoreInt32(&c.status, statusOffline)
-			default:
-				if err.Error() == "EOF" {
-					atomic.StoreInt32(&c.status, statusOffline)
-				} else {
-					return n, err
-				}
-			}
-		} else if atomic.LoadInt32(&c.status) == statusOffline {
-			if err := c.reconnect(); err != nil {
-				return -1, err
-			}
-		}
-		if i < (c.maxRetries - 1) {
-			time.Sleep(c.retryInterval)
-		}
-	}
-	return -1, ErrMaxRetries
-}
-
 //
 func (c *TCPClient) Write(b []byte) (int, error) {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 	for i := 0; i < c.maxRetries; i++ {
 		if atomic.LoadInt32(&c.status) == statusOnline {
-			for c.MyId == "" {
-				time.Sleep(1 * time.Second)
-			}
-			if (c.MyId != "" && string(b[4:12]) != c.MyId) || i == 1 { //重连过了，老数据要切换来源（仅修一次）
-				tmp := append(b[:4])
-				tmp = append(tmp, []byte(c.MyId)...)
-				tmp = append(tmp, b[12:]...)
-				b = append(tmp)
-			}
 			n, err := c.TCPConn.Write(b)
 			if err == nil {
 				return n, err
@@ -223,7 +140,6 @@ func (c *TCPClient) Write(b []byte) (int, error) {
 			switch err.(type) {
 			case *net.OpError:
 				atomic.StoreInt32(&c.status, statusOffline)
-
 			default:
 				return n, err
 			}

src/mfw/util/thread_test.go

@@ -1,35 +0,0 @@
-package util
-
-import (
-	"fmt"
-	"log"
-	"testing"
-	"time"
-)
-
-//
-func TestThread(t *testing.T) {
-	NewGoThread(4).Run(func(p ...interface{}) {
-		fmt.Println("vvv", p[0])
-	}, []interface{}{1}, []interface{}{2}, []interface{}{3}, []interface{}{4, 5, 6, 8, 9, 0})
-	//
-	time.Sleep(5 * time.Second)
-}
-
-func Test_error(t *testing.T) {
-	defer func() {
-		log.Println("1111")
-	}()
-	defer Catch()
-	defer func() {
-		log.Println("222")
-	}()
-
-	type ss struct {
-		B string
-		C map[string]string
-	}
-	s := ss{}
-	s.C["sss"] = "ssss"
-	log.Println(s.C)
-}

src/mfw/util/tlsclient.go

@@ -0,0 +1,135 @@
+/**
+支持tls的自动重连tcp连接，作为对tcpclient的补充
+*/
+package util
+
+import (
+	"crypto/tls"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+//
+type TLSClient struct {
+	*tls.Conn
+	lock          sync.RWMutex
+	status        int32
+	maxRetries    int
+	retryInterval time.Duration
+	onconnectedfn func()
+	tlsconfig     *tls.Config
+	network       string
+	serveraddr    string
+}
+
+func NewTlsClient(network, serveraddr string, maxretries int, retryinterval time.Duration, tlsconfig *tls.Config, onconnectedfn func()) *TLSClient {
+	return &TLSClient{
+		network:       network,
+		serveraddr:    serveraddr,
+		lock:          sync.RWMutex{},
+		maxRetries:    maxretries,
+		status:        statusOffline,
+		retryInterval: retryinterval,
+		tlsconfig:     tlsconfig,
+		onconnectedfn: onconnectedfn,
+	}
+}
+
+//连接
+func (t *TLSClient) Dial() error {
+	conn, err := tls.Dial(t.network, t.serveraddr, t.tlsconfig)
+	if err != nil {
+		return err
+	}
+	t.Conn = conn
+	t.status = statusOnline
+	return nil
+}
+
+//
+func (c *TLSClient) reconnect() error {
+	if !atomic.CompareAndSwapInt32(&c.status, statusOffline, statusReconnecting) {
+		return nil
+	}
+
+	conn, err := tls.Dial(c.network, c.serveraddr, c.tlsconfig)
+	if err != nil {
+		defer atomic.StoreInt32(&c.status, statusOffline)
+		switch err.(type) {
+		case *net.OpError:
+			return err
+		default:
+			return err
+		}
+	}
+
+	c.Conn.Close()
+	c.Conn = conn
+	atomic.StoreInt32(&c.status, statusOnline)
+	if c.onconnectedfn != nil {
+		c.onconnectedfn() //执行后续操作
+	}
+	return nil
+}
+
+//读操作
+func (c *TLSClient) Read(b []byte) (int, error) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	for i := 0; i < c.maxRetries; i++ {
+		if atomic.LoadInt32(&c.status) == statusOnline {
+			n, err := c.Conn.Read(b)
+			if err == nil {
+				return n, err
+			}
+			switch err.(type) {
+			case *net.OpError:
+				atomic.StoreInt32(&c.status, statusOffline)
+			default:
+				if err.Error() == "EOF" {
+					atomic.StoreInt32(&c.status, statusOffline)
+				} else {
+					return n, err
+				}
+			}
+		} else if atomic.LoadInt32(&c.status) == statusOffline {
+			if err := c.reconnect(); err != nil {
+				return -1, err
+			}
+		}
+		if i < (c.maxRetries - 1) {
+			time.Sleep(c.retryInterval)
+		}
+	}
+	return -1, ErrMaxRetries
+}
+
+//
+func (c *TLSClient) Write(b []byte) (int, error) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	for i := 0; i < c.maxRetries; i++ {
+		if atomic.LoadInt32(&c.status) == statusOnline {
+			n, err := c.Conn.Write(b)
+			if err == nil {
+				return n, err
+			}
+			switch err.(type) {
+			case *net.OpError:
+				atomic.StoreInt32(&c.status, statusOffline)
+			default:
+				return n, err
+			}
+		} else if atomic.LoadInt32(&c.status) == statusOffline {
+			if err := c.reconnect(); err != nil {
+				return -1, err
+			}
+		}
+		if i < (c.maxRetries - 1) {
+			time.Sleep(c.retryInterval)
+		}
+	}
+	return -1, ErrMaxRetries
+}

src/发布服务样例/main.go

@@ -4,10 +4,8 @@
 package main
 
 import (
+	"flag"
 	"mfw/util"
-	_ "os"
-	_ "runtime/pprof"
-	_ "time"
 )
 
 const (
@@ -22,18 +20,35 @@ func processEvent(p *util.Packet) {
 	event := int(p.Event)
 	switch event {
 	case SERVICE_SAYHEOLLO: //同步调用，原数据返回
-		client.WriteObj(p.From, p.Msgid, util.EVENT_RECIVE_CALLBACK, util.SENDTO_TYPE_P2P, "ok")
+		client.WriteObj(p.From, p.Msgid, util.EVENT_RETURN, util.SENDTO_TYPE_P2P, "ok")
 	case SERVICE_SAVEUSER: //异步
-		client.WriteObj(p.From, "", util.EVENT_RECIVE_CALLBACK, util.SENDTO_TYPE_P2P, "成功了啊")
+		client.WriteObj(p.From, "", util.EVENT_RETURN, util.SENDTO_TYPE_P2P, "成功了啊")
 	}
 }
 
 var client *util.Client
 
 func main() {
-	client, _ = util.StartClient(processEvent, MSG_SERVER, "这是一个发布的测试服务样例",
-		[]int{SERVICE_SAVEUSER, SERVICE_SAYHEOLLO})
-
+	addr := flag.String("addr", "127.0.0.1:7070", "消息服务地址")
+	name := flag.String("name", "myname", "客户端名称")
+	usetls := flag.Bool("usetls", false, "是否使用tls")
+	tlscert := flag.String("cert", "", "cert文件")
+	tlskey := flag.String("key", "", "key文件")
+	flag.Parse()
+	util.Log.SetLevel(util.LOG_LEVEL_INFO, true)
+	var err error
+	client, err = util.NewClient(&util.ClientConfig{
+		MsgServerAddr:   *addr,
+		ClientName:      *name,
+		CanHandleEvents: []int{1, 2, 3},
+		EventHandler:    processEvent,
+		UseTls:          *usetls,
+		TlsCertFile:     *tlscert,
+		TlsKeyFile:      *tlskey,
+	})
+	if err != nil {
+		util.Log.Debug.Println(err.Error())
+	}
 	//
 	flag := make(chan bool)
 	<-flag

src/服务端/main.go

@@ -13,7 +13,7 @@ import (
 )
 
 //
-type Client struct {
+type ClientItem struct {
 	conn            net.Conn
 	timestamp       int64
 	name            string //
@@ -24,9 +24,10 @@ type Client struct {
 
 //
 var lock sync.Mutex
+var gcinterval int64
 
 //所有的请求
-var allclient map[string]*Client = make(map[string]*Client)
+var allclient map[string]*ClientItem = make(map[string]*ClientItem)
 
 //服务与提供者对应表
 var allservice map[int][]string = make(map[int][]string)
@@ -35,12 +36,11 @@ var allservice map[int][]string = make(map[int][]string)
 //每次读写都会刷新心跳时间，
 //长时间未发生读写时发送心跳查询，
 //3倍检测时间后仍未响应的，自动剔除队列
-func GC() {
+func gc() {
 	now := time.Now().Unix()
 	now_bs := util.Int2Byte(int32(now))
 	for k, v := range allclient {
 		if now-v.timestamp > gcinterval*3 {
-			//3次GC未回应心跳
 			v.conn.Close()
 			removeClient(k)
 			continue
@@ -54,7 +54,7 @@ func GC() {
 		}
 
 	}
-	time.AfterFunc(time.Duration(gcinterval)*time.Second, GC)
+	time.AfterFunc(time.Duration(gcinterval)*time.Second, gc)
 }
 
 //删除服务节点
@@ -74,13 +74,11 @@ func removeClient(myid string) {
 			}
 		}
 	}
-
-	fmt.Println()
-	fmt.Println("删除节点", myid, "全部服务", allservice)
+	util.Log.Debug.Println("删除节点", myid, "全部服务", allservice)
 }
 
 //通过名称查找客户端
-func findClientByName(name string) (*Client, error) {
+func findClientByName(name string) (*ClientItem, error) {
 	for _, c := range allclient {
 		if c.name == name {
 			return c, nil
@@ -110,14 +108,13 @@ func publishservice(myid string, s []int) {
 		}
 		//追加新发布的服务
 		for _, v := range s {
-			if v == util.EVENT_RECIVE_CALLBACK { //回调的不再作为服务注册
+			if v == util.EVENT_RETURN { //回调的不再作为服务注册
 				continue
 			}
 			allservice[v] = append(allservice[v], myid)
 		}
 	}
-	fmt.Println()
-	fmt.Println("全部服务", allservice)
+	util.Log.Debug.Println("全部服务", allservice)
 }
 
 //更新心跳
@@ -146,18 +143,42 @@ func viewallservice(p *util.Packet) {
 	}
 	lock.Unlock()
 	bs, _ := json.Marshal(ret)
-	allclient[p.From].conn.Write(util.Enpacket("", p.From, p.Msgid, util.EVENT_RECIVE_CALLBACK, util.SENDTO_TYPE_P2P, bs))
+	allclient[p.From].conn.Write(util.Enpacket("", p.From, p.Msgid, util.EVENT_RETURN, util.SENDTO_TYPE_P2P, bs))
 }
 
-//更新客户端名称
-func updatemyname(myid, name string) {
-	fmt.Println("updatename", myid, name)
-	lock.Lock()
-	defer lock.Unlock()
-	if v, ok := allclient[myid]; ok {
-		v.name = name
+//加入网络
+func join(msg *util.Packet) {
+	util.Log.Debug.Printf("% X\n", msg.Raw)
+	data := map[string]interface{}{}
+	err := json.Unmarshal(msg.Raw, &data)
+	if err != nil { //解析join指令失败
+		msg.Conn.Close()
+		return
 	}
-
+	//
+	uuid := data["myid"].(string)
+	if allclient[uuid] != nil {
+		util.Log.Debug.Println(uuid, "已经存在")
+		msg.Conn.Close()
+		return
+	}
+	//
+	clientname := data["name"].(string)
+	handers_o_arr := data["handler"].([]interface{})
+	handers := []int{}
+	for _, v := range handers_o_arr {
+		handers = append(handers, int(v.(float64)))
+	}
+	//
+	allclient[uuid] = &ClientItem{conn: msg.Conn,
+		timestamp: time.Now().Unix(),
+		name:      clientname,
+		ip:        msg.Conn.RemoteAddr().String(),
+		id:        uuid,
+	}
+	//
+	publishservice(uuid, handers)
+	util.Log.Debug.Println(allservice)
 }
 
 //处理客户端发过来的消息
@@ -169,38 +190,20 @@ func processmsg(msg *util.Packet) {
 	updateheartbeat(from)
 	switch event {
 	//TODO 只写需要特殊处理的时间，其他都走default
+	case util.EVENT_REQUEST_JOIN:
+		//TODO 加入指令，直接返回
+		join(msg)
 	case util.EVENT_RETURN_HEARTBEAT: //心跳回应包处理
-		fmt.Print(".")
-	case util.EVENT_PUBLISH_MYSERVICES: //客户端发布了自己的服务
-		data := msg.GetBusinessData()
+	case util.EVENT_PUBLISH_MYSERVICES: //客户端发布了自己的服务,支持二次发布服务，用于暂停、更新服务能力
 		services := []int{}
-		for i := 0; i < len(data)/4; i++ {
-			start := i * 4
-			service := int(util.Byte2Int(data[start : start+4]))
-			services = append(services, service)
-		}
+		json.Unmarshal(msg.Raw, &services)
 		publishservice(from, services)
-	case util.EVENT_VIEWALL_SERVICE: //
+	case util.EVENT_VIEWALL_CLIENT: //
 		go viewallservice(msg)
 	case util.EVENT_REMOVE_CLIENT: //服务端强制删除节点
-		removeClient(string(msg.GetBusinessData()))
+		removeClient(string(msg.Raw))
 	case util.EVENT_BYE: //客户端主动要求断开
 		removeClient(from)
-	case util.EVENT_SYSTEM_COMMAND: //转发系统消息
-		if v, ok := allclient[msg.To]; ok {
-			_, err := v.conn.Write(msg.Raw)
-			if err == nil {
-				updateheartbeat(from)
-			}
-		}
-	case util.EVENT_UPDATE_MYNAME:
-		updatemyname(from, string(msg.GetBusinessData()))
-	case util.EVENT_GET_ALLDOWNLOADER:
-		if v, ok := allservice[util.SERVICE_DOWNLOAD]; ok {
-			if v1, ok1 := allclient[msg.From]; ok1 {
-				v1.conn.Write(util.Enpacket("", msg.From, msg.Msgid, util.EVENT_RECIVE_CALLBACK, util.SENDTO_TYPE_P2P, v))
-			}
-		}
 	default: //处理业务事件
 		//识别发送类型
 		sttype := int(msg.SentToType)
@@ -272,38 +275,30 @@ func processmsg(msg *util.Packet) {
 	}
 }
 
-var port string
-var gcinterval int64
-
 //
 func main() {
-	flag.StringVar(&port, "p", "7070", "开放端口")
-	flag.Int64Var(&gcinterval, "g", 10, "GC间隔时间")
+	addr := flag.String("addr", ":7070", "开放端口")
+	interval := flag.Int64("gc", 10, "GC间隔时间")
+	usetls := flag.Bool("usetls", false, "是否使用tls")
+	tlsca := flag.String("ca", "", "ca文件")
+	tlscert := flag.String("cert", "", "cert文件")
+	tlskey := flag.String("key", "", "key文件")
 	flag.Parse()
+	gcinterval = *interval
+	//初始化日志
+	util.Log = util.NewLogger("./logs", util.LOG_LEVEL_INFO, true, 1024*1024*50)
 	//心跳检测
-	go GC()
+	go gc()
 	//启动服务
-	util.StartServer(func(data *util.Packet) {
-		//接受消息处理
-		processmsg(data)
-	}, func(c net.Conn) { //连接后返回UUID
-		uuid := util.UUID(8)
-		//防止重复
-		for {
-			if allclient[uuid] == nil {
-				break
-			} else {
-				uuid = util.UUID(8)
-			}
-		}
-
-		c.Write(util.Enpacket("", "", "", util.EVENT_RETURN_MACHINE_ID, util.SENDTO_TYPE_P2P, []byte(uuid)))
-		allclient[uuid] = &Client{conn: c,
-			timestamp: time.Now().Unix(),
-			name:      c.RemoteAddr().String(),
-			ip:        c.RemoteAddr().String(),
-			id:        uuid,
-		}
-	}, ":"+port)
+	util.StartServer(&util.ServerConfig{
+		ListenAddr: *addr,
+		UseTls:     *usetls,
+		TlsCA:      *tlsca,
+		TlsCert:    *tlscert,
+		TlsKey:     *tlskey,
+		ProcessEvent: func(data *util.Packet) {
+			//接受消息处理
+			processmsg(data)
+		}})
 
 }

src/服务端/server.go

@@ -1,26 +0,0 @@
-package main
-
-import (
-	"log"
-	mu "mfw/util"
-	"net"
-	"time"
-)
-
-const (
-	WRITE_TIMEOUT = 10 //写超时2秒
-)
-
-func afterwrite(l time.Time) {
-	mu.Catch()
-	log.Printf("write taketime %.4fm \n", time.Now().Sub(l).Seconds())
-}
-
-//支持写超时
-func write2socket(con net.Conn, data []byte) (int, error) {
-	l1 := time.Now()
-	defer afterwrite(l1)
-	con.SetWriteDeadline(time.Now().Add(WRITE_TIMEOUT * time.Second))
-	n, err := con.Write(data)
-	return n, err
-}