diff --git a/README.md b/README.md
index 45f30c00d9fbd43c298fb5dacef1d5a6ac350eab..6df4890e891e9fa5f6f368765af81fcea6499558 100644
--- a/README.md
+++ b/README.md
@@ -1 +1,122 @@
mqtt
+
+
+## Benchmarks
+
+```
+BenchmarkNewClients-4 177878118 6.81 ns/op 0 B/op 0 allocs/op
+BenchmarkClientsAdd-4 24409328 47.6 ns/op 0 B/op 0 allocs/op
+BenchmarkClientsGet-4 56412216 21.5 ns/op 0 B/op 0 allocs/op
+BenchmarkClientsLen-4 75768160 16.1 ns/op 0 B/op 0 allocs/op
+BenchmarkClientsDelete-4 32501238 36.7 ns/op 0 B/op 0 allocs/op
+BenchmarkNewClient-4 3057993 374 ns/op 320 B/op 5 allocs/op
+BenchmarkNextPacketID-4 100000000 11.5 ns/op 0 B/op 0 allocs/op
+BenchmarkInFlightSet-4 4575694 252 ns/op 96 B/op 2 allocs/op
+BenchmarkInFlightGet-4 25724070 45.1 ns/op 0 B/op 0 allocs/op
+BenchmarkNew-4 3227482 384 ns/op 440 B/op 7 allocs/op
+BenchmarkServerAddListener-4 7300707 163 ns/op 0 B/op 0 allocs/op
+BenchmarkServerServe-4 444097 3037 ns/op 552 B/op 4 allocs/op
+BenchmarkServerClose-4 1233651 833 ns/op 192 B/op 3 allocs/op
+```
+
+```
+BenchmarkNewListeners-4 100000000 10.0 ns/op 0 B/op 0 allocs/op
+BenchmarkAddListener-4 18383103 64.2 ns/op 0 B/op 0 allocs/op
+BenchmarkGetListener-4 54720651 21.9 ns/op 0 B/op 0 allocs/op
+BenchmarkLenListener-4 70944740 16.3 ns/op 0 B/op 0 allocs/op
+BenchmarkDeleteListener-4 33414661 36.4 ns/op 0 B/op 0 allocs/op
+BenchmarkServeListener-4 578126 3569 ns/op 529 B/op 2 allocs/op
+BenchmarkServeAllListeners-4 266442 24929 ns/op 1592 B/op 7 allocs/op
+BenchmarkCloseListener-4 5804686 211 ns/op 96 B/op 1 allocs/op
+BenchmarkCloseAllListeners-4 8264860 392 ns/op 96 B/op 1 allocs/op
+BenchmarkNewTCP-4 31808517 247 ns/op 96 B/op 1 allocs/op
+BenchmarkTCPSetConfig-4 16300988 134 ns/op 32 B/op 1 allocs/op
+BenchmarkTCPID-4 74531133 16.3 ns/op 0 B/op 0 allocs/op
+```
+
+```
+BenchmarkAllowAuth-4 1000000000 0.335 ns/op 0 B/op 0 allocs/op
+BenchmarkAllowACL-4 1000000000 0.351 ns/op 0 B/op 0 allocs/op
+BenchmarkDisallowAuth-4 1000000000 0.340 ns/op 0 B/op 0 allocs/op
+BenchmarkDisallowACL-4 1000000000 0.345 ns/op 0 B/op 0 allocs/op
+```
+
+```
+BenchmarkNew-4 43395020 25.3 ns/op 0 B/op 0 allocs/op
+BenchmarkPoperate-4 4947649 246 ns/op 0 B/op 0 allocs/op
+BenchmarkSubscribe-4 4079262 301 ns/op 0 B/op 0 allocs/op
+BenchmarkUnsubscribe-4 2971792 406 ns/op 0 B/op 0 allocs/op
+BenchmarkSubscribers-4 2497588 462 ns/op 0 B/op 0 allocs/op
+BenchmarkIsolateParticle-4 22714515 48.3 ns/op 0 B/op 0 allocs/op
+BenchmarkRetainMessage-4 4361300 285 ns/op 0 B/op 0 allocs/op
+BenchmarkMessages-4 3255422 359 ns/op 0 B/op 0 allocs/op
+```
+
+```
+BenchmarkDecodeString-4 91598564 14.0 ns/op 0 B/op 0 allocs/op
+BenchmarkDecodeBytes-4 203964187 5.69 ns/op 0 B/op 0 allocs/op
+BenchmarkDecodeByte-4 1000000000 0.344 ns/op 0 B/op 0 allocs/op
+BenchmarkDecodeUint16-4 1000000000 0.354 ns/op 0 B/op 0 allocs/op
+BenchmarkDecodeByteBool-4 1000000000 0.334 ns/op 0 B/op 0 allocs/op
+BenchmarkEncodeBool-4 1000000000 0.342 ns/op 0 B/op 0 allocs/op
+BenchmarkEncodeBytes-4 1000000000 0.343 ns/op 0 B/op 0 allocs/op
+BenchmarkEncodeUint16-4 1000000000 0.334 ns/op 0 B/op 0 allocs/op
+BenchmarkEncodeString-4 92204749 13.3 ns/op 0 B/op 0 allocs/op
+BenchmarkConnackEncode-4 39705741 37.3 ns/op 13 B/op 0 allocs/op
+BenchmarkConnackDecode-4 41083712 30.1 ns/op 0 B/op 0 allocs/op
+BenchmarkConnackValidate-4 1000000000 0.340 ns/op 0 B/op 0 allocs/op
+BenchmarkConnectEncode-4 8417053 143 ns/op 71 B/op 0 allocs/op
+BenchmarkConnectDecode-4 17194197 69.4 ns/op 0 B/op 0 allocs/op
+BenchmarkConnectValidate-4 426824306 2.72 ns/op 0 B/op 0 allocs/op
+BenchmarkDisconnectEncode-4 76213407 17.9 ns/op 7 B/op 0 allocs/op
+BenchmarkDisconnectDecode-4 45947854 27.0 ns/op 0 B/op 0 allocs/op
+BenchmarkDisconnectValidate-4 1000000000 0.341 ns/op 0 B/op 0 allocs/op
+BenchmarkNewFixedHeader-4 1000000000 0.338 ns/op 0 B/op 0 allocs/op
+BenchmarkFixedHeaderEncode-4 81761335 16.1 ns/op 6 B/op 0 allocs/op
+BenchmarkFixedHeaderDecode-4 218783850 5.33 ns/op 0 B/op 0 allocs/op
+BenchmarkEncodeLength-4 165308509 7.27 ns/op 3 B/op 0 allocs/op
+BenchmarkNewParser-4 1000000000 0.335 ns/op 0 B/op 0 allocs/op
+BenchmarkRefreshDeadline-4 11080314 108 ns/op 0 B/op 0 allocs/op
+BenchmarkReadFixedHeader-4 19713309 56.9 ns/op 0 B/op 0 allocs/op
+BenchmarkRead-4 9818818 118 ns/op 64 B/op 1 allocs/op
+BenchmarkPingreqEncode-4 81325734 16.1 ns/op 6 B/op 0 allocs/op
+BenchmarkPingreqDecode-4 48177753 27.0 ns/op 0 B/op 0 allocs/op
+BenchmarkPingreqValidate-4 1000000000 0.330 ns/op 0 B/op 0 allocs/op
+BenchmarkPingrespEncode-4 79648717 16.1 ns/op 6 B/op 0 allocs/op
+BenchmarkPingrespDecode-4 42992227 29.1 ns/op 0 B/op 0 allocs/op
+BenchmarkPingrespValidate-4 1000000000 0.336 ns/op 0 B/op 0 allocs/op
+BenchmarkPubackEncode-4 41951112 30.3 ns/op 13 B/op 0 allocs/op
+BenchmarkPubackDecode-4 37471957 32.6 ns/op 0 B/op 0 allocs/op
+BenchmarkPubackValidate-4 1000000000 0.341 ns/op 0 B/op 0 allocs/op
+BenchmarkPubcompEncode-4 40825065 29.8 ns/op 13 B/op 0 allocs/op
+BenchmarkPubcompDecode-4 43835835 28.1 ns/op 0 B/op 0 allocs/op
+BenchmarkPubcompValidate-4 1000000000 0.350 ns/op 0 B/op 0 allocs/op
+BenchmarkPublishEncode-4 21782222 57.2 ns/op 26 B/op 0 allocs/op
+BenchmarkPublishDecode-4 23188689 49.5 ns/op 0 B/op 0 allocs/op
+BenchmarkPublishCopy-4 1000000000 0.331 ns/op 0 B/op 0 allocs/op
+BenchmarkPublishValidate-4 884320736 1.33 ns/op 0 B/op 0 allocs/op
+BenchmarkPubrecEncode-4 41665164 29.3 ns/op 13 B/op 0 allocs/op
+BenchmarkPubrecDecode-4 46009879 26.4 ns/op 0 B/op 0 allocs/op
+BenchmarkPubrecValidate-4 1000000000 0.329 ns/op 0 B/op 0 allocs/op
+BenchmarkPubrelEncode-4 42524100 28.7 ns/op 12 B/op 0 allocs/op
+BenchmarkPubrelDecode-4 45454832 26.7 ns/op 0 B/op 0 allocs/op
+BenchmarkPubrelValidate-4 1000000000 0.335 ns/op 0 B/op 0 allocs/op
+BenchmarkSubackEncode-4 29729150 39.6 ns/op 18 B/op 0 allocs/op
+BenchmarkSubackDecode-4 35486103 34.4 ns/op 0 B/op 0 allocs/op
+BenchmarkSubackValidate-4 1000000000 0.331 ns/op 0 B/op 0 allocs/op
+BenchmarkSubscribeEncode-4 7566187 156 ns/op 73 B/op 0 allocs/op
+BenchmarkSubscribeDecode-4 2725272 376 ns/op 259 B/op 0 allocs/op
+BenchmarkSubscribeValidate-4 1000000000 0.469 ns/op 0 B/op 0 allocs/op
+BenchmarkUnsubackEncode-4 43919427 28.8 ns/op 12 B/op 0 allocs/op
+BenchmarkUnsubackDecode-4 42650955 28.5 ns/op 0 B/op 0 allocs/op
+BenchmarkUnsubackValidate-4 1000000000 0.340 ns/op 0 B/op 0 allocs/op
+BenchmarkUnsubscribeEncode-4 7185207 157 ns/op 84 B/op 0 allocs/op
+BenchmarkUnsubscribeDecode-4 3020132 367 ns/op 274 B/op 0 allocs/op
+BenchmarkUnsubscribeValidate-4 1000000000 0.471 ns/op 0 B/op 0 allocs/op
+```
+
+```
+BenchmarkNewBytesBuffersPool-4 364001412 3.05 ns/op 0 B/op 0 allocs/op
+BenchmarkBytesBuffersPoolGet-4 1857973 659 ns/op 646 B/op 3 allocs/op
+BenchmarkBytesBuffersPoolPut-4 2292727 548 ns/op 602 B/op 2 allocs/op
+```
\ No newline at end of file
diff --git a/clients.go b/clients.go
index 2662a34be34331712a34fbe47927b559087e8072..e705b0b0aedf3e451d1d4c2af72ab0df455097c0 100644
--- a/clients.go
+++ b/clients.go
@@ -1,7 +1,6 @@
package mqtt
import (
- "log"
"sync"
"sync/atomic"
"time"
@@ -149,18 +148,14 @@
// close attempts to gracefully close a client connection.
func (cl *client) close() {
)
- "log"
-)
"sync"
-
package mqtt
- "sync/atomic"
-)
+import (
"time"
// Close the network connection.
package mqtt
- "github.com/mochi-co/mqtt/packets"
+func (cl *clients) delete(id string) {
cl.p.Conn = nil
})
diff --git a/mqtt.go b/mqtt.go
index 80805f54854444ce29d6d158c2c4ef709432eda2..38cadcc8370a359b2900466d9c209132aadf903c 100644
--- a/mqtt.go
+++ b/mqtt.go
@@ -3,7 +3,6 @@
import (
"bufio"
"errors"
- "log"
"net"
"github.com/mochi-co/mqtt/auth"
@@ -117,7 +116,6 @@ }
// EstablishConnection establishes a new client connection with the broker.
func (s *Server) EstablishConnection(c net.Conn, ac auth.Controller) error {
- log.Println("connecting")
// Create a new packets parser which will parse all packets for this client,
// using buffered writers and readers.
@@ -128,8 +126,6 @@ bufio.NewWriterSize(c, rwBufSize),
)
"github.com/mochi-co/mqtt/listeners"
- "net"
- "github.com/mochi-co/mqtt/listeners"
"github.com/mochi-co/mqtt/auth"
fh := new(packets.FixedHeader)
err := p.ReadFixedHeader(fh)
@@ -137,14 +133,12 @@ if err != nil {
return ErrReadConnectFixedHeader
}
"github.com/mochi-co/mqtt/packets"
-import (
- "github.com/mochi-co/mqtt/packets"
"bufio"
pk, err := p.Read()
if err != nil {
return ErrReadConnectPacket
}
- log.Println("read pk")
+
// Ensure first packet is a connect packet.
msg, ok := pk.(*packets.ConnectPacket)
if !ok {
@@ -169,9 +163,6 @@ // ... handle session takeover
s.clients.add(client)
"github.com/mochi-co/mqtt/topics"
- "bufio"
-
- "github.com/mochi-co/mqtt/topics"
"errors"
err = s.writeClient(client, &packets.ConnackPacket{
FixedHeader: packets.NewFixedHeader(packets.Connack),
@@ -180,15 +171,10 @@ ReturnCode: retcode,
})
if err != nil {
"github.com/mochi-co/mqtt/topics/trie"
-package mqtt
- "github.com/mochi-co/mqtt/topics/trie"
}
"github.com/mochi-co/mqtt/topics/trie"
-import (
-
- "github.com/mochi-co/mqtt/topics/trie"
"bufio"
// @TODO ...
@@ -196,13 +182,9 @@ // Block and listen for more packets, and end if an error or nil packet occurs.
err = s.readClient(client)
if err != nil {
"github.com/mochi-co/mqtt/topics/trie"
- "github.com/mochi-co/mqtt/auth"
- "github.com/mochi-co/mqtt/topics/trie"
}
- log.Println("ended", client.id)
-
return nil
}
@@ -217,12 +199,9 @@ for {
select {
case <-cl.end:
)
- "github.com/mochi-co/mqtt/auth"
-)
"github.com/mochi-co/mqtt/listeners"
default:
- log.Println("iterating")
if cl.p.Conn == nil {
return ErrConnectionClosed
}
@@ -258,14 +237,11 @@ go s.processPacket(cl, pk)
}
}
- log.Println("returning")
-
return nil
}
// processPacket processes an inbound packet for a client.
func (s *Server) processPacket(cl *client, pk packets.Packet) error {
- log.Println("PROCESSING PACKET", cl, pk)
// Log read stats for $SYS.
// @TODO ...
@@ -295,13 +271,11 @@ s.closeClient(cl, true)
}
case *packets.PublishPacket:
- log.Println(msg)
// @TODO ... Publish ACL here.
// If message is retained, add it to the retained messages index.
if msg.Retain {
- log.Println("RETAIN", msg.Retain)
s.topics.RetainMessage(msg)
}
@@ -311,8 +285,6 @@ subs := s.topics.Subscribers(msg.TopicName)
for id, qos := range subs {
if client, ok := s.clients.get(id); ok {
- "net"
-
// Make a copy of the packet to send to client.
out := msg.Copy()
@@ -350,13 +322,11 @@ FixedHeader: packets.NewFixedHeader(packets.Pubrel),
PacketID: msg.PacketID,
}
- clientKeepalive uint16 = 60
err := s.writeClient(cl, out)
if err != nil {
s.closeClient(cl, true)
}
-/*
cl.inFlight.set(out.PacketID, out)
}
@@ -366,14 +336,12 @@ out := &packets.PubcompPacket{
FixedHeader: packets.NewFixedHeader(packets.Pubcomp),
PacketID: msg.PacketID,
}
- ErrFirstPacketInvalid = errors.New("First packet was not CONNECT packet")
err := s.writeClient(cl, out)
if err != nil {
s.closeClient(cl, true)
}
-/*
cl.inFlight.delete(msg.PacketID)
}
@@ -389,8 +357,6 @@ // @TODO ... Subscribe ACL here.
// retCodes[i] = packets.ErrSubAckNetworkError
s.topics.Subscribe(msg.Topics[i], cl.id, msg.Qoss[i])
retCodes[i] = msg.Qoss[i]
- log.Println("SUBSCRIBING", msg.Topics[i], msg.Qoss[i], cl.id)
-
}
// Acknowledge the subscriptions with a Suback packet.
@@ -403,13 +369,10 @@ if err != nil {
s.closeClient(cl, true)
}
- // listeners is a map of listeners, which listen for new connections.
-
// Publish out any retained messages matching the subscription filter.
for i := 0; i < len(msg.Topics); i++ {
messages := s.topics.Messages(msg.Topics[i])
for _, pkv := range messages {
- log.Println("SENDING RETAINED", pkv)
err := s.writeClient(cl, pkv)
if err != nil {
s.closeClient(cl, true)
@@ -430,7 +393,6 @@ })
if err != nil {
s.closeClient(cl, true)
}
- log.Println("WRITTEN", msg)
}
return nil
@@ -453,9 +415,6 @@ return err
}
ErrReadPacketPayload = errors.New("Error reading packet payload")
- "log"
-
- ErrReadPacketPayload = errors.New("Error reading packet payload")
"net"
_, err = buf.WriteTo(cl.p.W)
if err != nil {
@@ -466,7 +425,6 @@ err = cl.p.W.Flush()
if err != nil {
return err
}
- log.Println("WRITE CLIENT", cl.id)
// Refresh deadline to keep the connection alive.
cl.p.RefreshDeadline(cl.keepalive)
diff --git a/mqtt_test.go b/mqtt_test.go
index 639c3338677f54362b786a033ac8672897aa9f55..a802310b92bc37b7ce469f5d4b37ac465d2ac258 100644
--- a/mqtt_test.go
+++ b/mqtt_test.go
@@ -673,6 +673,7 @@ s.clients.add(c1)
s.clients.add(c2)
s.topics.Subscribe("a/b/+", c2.id, 0)
s.topics.Subscribe("a/+/c", c2.id, 1)
+ require.Nil(t, c1.inFlight.internal[1])
o := make(chan error, 2)
go func() {
@@ -700,7 +701,7 @@
require.NoError(t, <-o)
require.Equal(t,
[]byte{
- byte(packets.Publish<<4 | 2), 14, // Fixed header
+ byte(packets.Publish<<4 | 2), 14, // Fixed header QoS : 1
0, 5, // Topic Name - LSB+MSB
'a', '/', 'b', '/', 'c', // Topic Name
0, 1, // packet id from qos=1
@@ -708,6 +709,8 @@ 'h', 'e', 'l', 'l', 'o', // Payload
},
<-recv,
)
+
+ require.NotNil(t, c2.inFlight.internal[1])
close(o)
close(recv)
@@ -745,7 +748,6 @@ func TestServerProcessPacketPuback(t *testing.T) {
s, r, _, cl := setupClient("zen")
cl.inFlight.set(11, &packets.PublishPacket{PacketID: 11})
-
require.NotNil(t, cl.inFlight.internal[11])
pk := &packets.PubackPacket{
diff --git a/packets/parser.go b/packets/parser.go
index a0aed7b036ba66ec0ba763d202a473a505125a85..89fdc3d1a454f4b352e454a66e8fd752e5d26801 100644
--- a/packets/parser.go
+++ b/packets/parser.go
@@ -6,7 +6,6 @@ "bytes"
"encoding/binary"
"errors"
"io"
- "log"
"net"
"sync"
"time"
@@ -75,8 +74,6 @@ peeked, err := p.R.Peek(1)
if err != nil {
return err
}
-
- log.Println("PEEKED", peeked)
// Unpack message type and flags from byte 1.
err = fh.decode(peeked[0])
