Added support for MQTT retain and rewritten in-memory storage

go.mod

@@ -2,7 +2,7 @@ module github.com/emitter-io/emitter
 
 require (
 	github.com/AndreasBriese/bbloom v0.0.0-20180913140656-343706a395b7 // indirect
-	github.com/aws/aws-sdk-go v0.0.0-20181019205654-b2427922671b
+	github.com/aws/aws-sdk-go v0.0.0-20181019205654-b2427922671b // indirect
 	github.com/axiomhq/hyperloglog v0.0.0-20180317131949-fe9507de0228
 	github.com/dgraph-io/badger v0.0.0-20181020042726-fbb27786246d
 	github.com/dgryski/go-farm v0.0.0-20180109070241-2de33835d102 // indirect
@@ -13,23 +13,26 @@ require (
 	github.com/emitter-io/stats v1.0.1
 	github.com/golang/protobuf v0.0.0-20181005181728-ddf22928ea3c // indirect
 	github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db
-	github.com/gopherjs/gopherjs v0.0.0-20181103185306-d547d1d9531e // indirect
 	github.com/gorilla/websocket v1.4.0
 	github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af // indirect
-	github.com/karlseguin/ccache v0.0.0-20180722041702-a3174167557e
-	github.com/karlseguin/expect v1.0.1 // indirect
 	github.com/kelindar/binary v1.0.1
 	github.com/kelindar/tcp v0.0.0-20170901060004-951ea01d26dd
 	github.com/pkg/errors v0.0.0-20181008045315-2233dee583dc // indirect
 	github.com/stretchr/objx v0.0.0-20180825064932-ef50b0de2877 // indirect
 	github.com/stretchr/testify v1.2.2
+	github.com/tidwall/btree v0.0.0-20170113224114-9876f1454cf0 // indirect
+	github.com/tidwall/buntdb v1.0.0
+	github.com/tidwall/gjson v1.2.1 // indirect
+	github.com/tidwall/grect v0.0.0-20161006141115-ba9a043346eb // indirect
+	github.com/tidwall/match v1.0.1 // indirect
+	github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51 // indirect
+	github.com/tidwall/rtree v0.0.0-20180113144539-6cd427091e0e // indirect
+	github.com/tidwall/tinyqueue v0.0.0-20180302190814-1e39f5511563 // indirect
 	github.com/valyala/fasthttp v0.0.0-20181014131909-996610f021ff
 	github.com/weaveworks/mesh v0.0.0-20180416113225-61ba45522f8a
-	github.com/wsxiaoys/terminal v0.0.0-20160513160801-0940f3fc43a0 // indirect
 	golang.org/x/crypto v0.0.0-20181015023909-0c41d7ab0a0e
 	golang.org/x/net v0.0.0-20181017193950-04a2e542c03f // indirect
 	golang.org/x/sys v0.0.0-20181019160139-8e24a49d80f8 // indirect
 	golang.org/x/text v0.3.0 // indirect
 	gopkg.in/alexcesaro/statsd.v2 v2.0.0
-	gopkg.in/karlseguin/expect.v1 v1.0.1 // indirect
 )

go.sum

@@ -25,16 +25,11 @@ github.com/golang/protobuf v0.0.0-20181005181728-ddf22928ea3c/go.mod h1:Qd/q+1AK
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db h1:woRePGFeVFfLKN/pOkfl+p/TAqKOfFu+7KPlMVpok/w=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/gopherjs/gopherjs v0.0.0-20181103185306-d547d1d9531e/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
-github.com/karlseguin/ccache v0.0.0-20180722041702-a3174167557e h1:D6upEoXsbRNEe3uNPzRf4rYFbxn076khDpywaNZxHYY=
-github.com/karlseguin/ccache v0.0.0-20180722041702-a3174167557e/go.mod h1:CM9tNPzT6EdRh14+jiW8mEF9mkNZuuE51qmgGYUB93w=
-github.com/karlseguin/expect v1.0.1 h1:z4wy4npwwHSWKjGWH85WNJO42VQhovxTCZDSzhjo8hY=
-github.com/karlseguin/expect v1.0.1/go.mod h1:zNBxMY8P21owkeogJELCLeHIt+voOSduHYTFUbwRAV8=
 github.com/kelindar/binary v1.0.1 h1:618UzNUN9ckT2CeGcRF4N8q/0WWtvcpVL6O1DDfCXUM=
 github.com/kelindar/binary v1.0.1/go.mod h1:wMMTJccK3HLFh6eH9bD4VRIjUTFVW7pC1q0vTUQ7vV8=
 github.com/kelindar/process v0.0.0-20170730150328-69a29e249ec3 h1:6If+E1dikQbdT7DlhZqLplfGkEt6dSoz7+MK+TFC7+U=
@@ -53,15 +48,29 @@ github.com/stretchr/objx v0.0.0-20180825064932-ef50b0de2877 h1:Uk9JZ6L7yGoJ0V/gy
 github.com/stretchr/objx v0.0.0-20180825064932-ef50b0de2877/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/tidwall/btree v0.0.0-20170113224114-9876f1454cf0 h1:QnyrPZZvPmR0AtJCxxfCtI1qN+fYpKTKJ/5opWmZ34k=
+github.com/tidwall/btree v0.0.0-20170113224114-9876f1454cf0/go.mod h1:huei1BkDWJ3/sLXmO+bsCNELL+Bp2Kks9OLyQFkzvA8=
+github.com/tidwall/buntdb v1.0.0 h1:urIJqQ8OR9fibXXtFSu8sR5arMqZK8ZnNq22yWl+A+8=
+github.com/tidwall/buntdb v1.0.0/go.mod h1:Y39xhcDW10WlyYXeLgGftXVbjtM0QP+/kpz8xl9cbzE=
+github.com/tidwall/gjson v1.2.1 h1:j0efZLrZUvNerEf6xqoi0NjWMK5YlLrR7Guo/dxY174=
+github.com/tidwall/gjson v1.2.1/go.mod h1:c/nTNbUr0E0OrXEhq1pwa8iEgc2DOt4ZZqAt1HtCkPA=
+github.com/tidwall/grect v0.0.0-20161006141115-ba9a043346eb h1:5NSYaAdrnblKByzd7XByQEJVT8+9v0W/tIY0Oo4OwrE=
+github.com/tidwall/grect v0.0.0-20161006141115-ba9a043346eb/go.mod h1:lKYYLFIr9OIgdgrtgkZ9zgRxRdvPYsExnYBsEAd8W5M=
+github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=
+github.com/tidwall/match v1.0.1/go.mod h1:LujAq0jyVjBy028G1WhWfIzbpQfMO8bBZ6Tyb0+pL9E=
+github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51 h1:BP2bjP495BBPaBcS5rmqviTfrOkN5rO5ceKAMRZCRFc=
+github.com/tidwall/pretty v0.0.0-20180105212114-65a9db5fad51/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/tidwall/rtree v0.0.0-20180113144539-6cd427091e0e h1:+NL1GDIUOKxVfbp2KoJQD9cTQ6dyP2co9q4yzmT9FZo=
+github.com/tidwall/rtree v0.0.0-20180113144539-6cd427091e0e/go.mod h1:/h+UnNGt0IhNNJLkGikcdcJqm66zGD/uJGMRxK/9+Ao=
+github.com/tidwall/tinyqueue v0.0.0-20180302190814-1e39f5511563 h1:Otn9S136ELckZ3KKDyCkxapfufrqDqwmGjcHfAyXRrE=
+github.com/tidwall/tinyqueue v0.0.0-20180302190814-1e39f5511563/go.mod h1:mLqSmt7Dv/CNneF2wfcChfN1rvapyQr01LGKnKex0DQ=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/valyala/fasthttp v0.0.0-20181014131909-996610f021ff h1:KdfHkiQLtYASN2INA+Pvk0a53ozqlqNs0sfv4Ci1RcQ=
 github.com/valyala/fasthttp v0.0.0-20181014131909-996610f021ff/go.mod h1:4vX61m6KN+xDduDNwXrhIAVZaZaZiQ1luJk8LWSxF3s=
 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
 github.com/weaveworks/mesh v0.0.0-20180416113225-61ba45522f8a h1:dcGmKbrZRO6/VcV9fdo8wmlXi90b8E50dPy7pyOKzXE=
 github.com/weaveworks/mesh v0.0.0-20180416113225-61ba45522f8a/go.mod h1:mcON9Ws1aW0crSErpXWp7U1ErCDEKliDX2OhVlbWRKk=
-github.com/wsxiaoys/terminal v0.0.0-20160513160801-0940f3fc43a0 h1:3UeQBvD0TFrlVjOeLOBz+CPAI8dnbqNSVwUwRrkp7vQ=
-github.com/wsxiaoys/terminal v0.0.0-20160513160801-0940f3fc43a0/go.mod h1:IXCdmsXIht47RaVFLEdVnh1t+pgYtTAhQGj73kz+2DM=
 golang.org/x/crypto v0.0.0-20181015023909-0c41d7ab0a0e h1:IzypfodbhbnViNUO/MEh0FzCUooG97cIGfdggUrUSyU=
 golang.org/x/crypto v0.0.0-20181015023909-0c41d7ab0a0e/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -77,5 +86,3 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 google.golang.org/genproto v0.0.0-20180831171423-11092d34479b/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 gopkg.in/alexcesaro/statsd.v2 v2.0.0 h1:FXkZSCZIH17vLCO5sO2UucTHsH9pc+17F6pl3JVCwMc=
 gopkg.in/alexcesaro/statsd.v2 v2.0.0/go.mod h1:i0ubccKGzBVNBpdGV5MocxyA/XlLUJzA7SLonnE4drU=
-gopkg.in/karlseguin/expect.v1 v1.0.1 h1:9u0iUltnhFbJTHaSIH0EP+cuTU5rafIgmcsEsg2JQFw=
-gopkg.in/karlseguin/expect.v1 v1.0.1/go.mod h1:uB7QIJBcclvYbwlUDkSCsGjAOMis3fP280LyhuDEf2I=

internal/broker/conn.go

@@ -193,7 +193,7 @@ func (c *Conn) onReceive(msg mqtt.Message) error {
 
 	case mqtt.TypeOfPublish:
 		packet := msg.(*mqtt.Publish)
-		if err := c.onPublish(packet.Topic, packet.Payload, packet.MessageID); err != nil {
+		if err := c.onPublish(packet); err != nil {
 			logging.LogError("conn", "publish received", err)
 			c.notifyError(err, packet.MessageID)
 		}

internal/broker/handlers.go

@@ -91,8 +91,15 @@ func (c *Conn) onSubscribe(mqttTopic []byte) *Error {
 	ssid := message.NewSsid(key.Contract(), channel.Query)
 	c.Subscribe(ssid, channel.Channel)
 
-	// In case of ttl, check the key provides the permission to store (soft permission)
-	if limit, ok := channel.Last(); ok && key.HasPermission(security.AllowLoad) {
+	// Use limit = 1 if not specified, otherwise use the limit option. The limit now
+	// defaults to one as per MQTT spec we always need to send retained messages.
+	limit := int64(1)
+	if v, ok := channel.Last(); ok {
+		limit = v
+	}
+
+	// Check if the key has a load permission (also applies for retained)
+	if key.HasPermission(security.AllowLoad) {
 		t0, t1 := channel.Window() // Get the window
 		msgs, err := c.service.storage.Query(ssid, t0, t1, int(limit))
 		if err != nil {
@@ -139,8 +146,8 @@ func (c *Conn) onUnsubscribe(mqttTopic []byte) *Error {
 // ------------------------------------------------------------------------------------
 
 // OnPublish is a handler for MQTT Publish events.
-func (c *Conn) onPublish(mqttTopic []byte, payload []byte, requestID uint16) *Error {
-	exclude := ""
+func (c *Conn) onPublish(packet *mqtt.Publish) *Error {
+	mqttTopic := packet.Topic
 	if len(mqttTopic) <= 2 && c.links != nil {
 		mqttTopic = []byte(c.links[string(mqttTopic)])
 	}
@@ -158,7 +165,7 @@ func (c *Conn) onPublish(mqttTopic []byte, payload []byte, requestID uint16) *Er
 
 	// Check whether the key is 'emitter' which means it's an API request
 	if len(channel.Key) == 7 && string(channel.Key) == "emitter" {
-		c.onEmitterRequest(channel, payload, requestID)
+		c.onEmitterRequest(channel, packet.Payload, packet.MessageID)
 		return nil
 	}
 
@@ -172,16 +179,26 @@ func (c *Conn) onPublish(mqttTopic []byte, payload []byte, requestID uint16) *Er
 	msg := message.New(
 		message.NewSsid(key.Contract(), channel.Query),
 		channel.Channel,
-		payload,
+		packet.Payload,
 	)
 
-	// In case of ttl, check the key provides the permission to store (soft permission)
-	if ttl, ok := channel.TTL(); ok && key.HasPermission(security.AllowStore) {
-		msg.TTL = uint32(ttl) // Add the TTL to the message
+	// If a user have specified a retain flag, retain with a default TTL
+	if packet.Header.Retain {
+		msg.TTL = message.RetainedTTL
+	}
+
+	// If a user have specified a TTL, use that value
+	if ttl, ok := channel.TTL(); ok && ttl > 0 {
+		msg.TTL = uint32(ttl)
+	}
+
+	// Store the message if needed
+	if msg.Stored() && key.HasPermission(security.AllowStore) {
 		c.service.storage.Store(msg)
 	}
 
 	// Check whether an exclude me option was set (i.e.: 'me=0')
+	var exclude string
 	if channel.Exclude() {
 		exclude = c.ID()
 	}
@@ -191,7 +208,7 @@ func (c *Conn) onPublish(mqttTopic []byte, payload []byte, requestID uint16) *Er
 
 	// Write the monitoring information
 	c.track(contract)
-	contract.Stats().AddIngress(int64(len(payload)))
+	contract.Stats().AddIngress(int64(len(packet.Payload)))
 	contract.Stats().AddEgress(size)
 	return nil
 }

internal/broker/handlers_test.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/emitter-io/emitter/internal/message"
 	netmock "github.com/emitter-io/emitter/internal/network/mock"
+	"github.com/emitter-io/emitter/internal/network/mqtt"
 	"github.com/emitter-io/emitter/internal/provider/contract"
 	secmock "github.com/emitter-io/emitter/internal/provider/contract/mock"
 	"github.com/emitter-io/emitter/internal/provider/usage"
@@ -320,7 +321,11 @@ func TestHandlers_onPublish(t *testing.T) {
 		nc := s.newConn(conn.Client)
 		s.Cipher, _ = s.License.Cipher()
 
-		err := nc.onPublish([]byte(tc.channel), []byte(tc.payload), 0)
+		err := nc.onPublish(&mqtt.Publish{
+			Header:  new(mqtt.StaticHeader),
+			Topic:   []byte(tc.channel),
+			Payload: []byte(tc.payload),
+		})
 
 		assert.Equal(t, tc.err, err, tc.msg)
 	}

internal/broker/service_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/emitter-io/emitter/internal/network/mqtt"
 	"github.com/emitter-io/emitter/internal/provider/contract"
 	secmock "github.com/emitter-io/emitter/internal/provider/contract/mock"
+	"github.com/emitter-io/emitter/internal/provider/storage"
 	"github.com/emitter-io/emitter/internal/provider/usage"
 	"github.com/emitter-io/emitter/internal/security"
 	"github.com/stretchr/testify/assert"
@@ -143,6 +144,8 @@ func TestPubsub(t *testing.T) {
 	// Start the broker asynchronously
 	broker, svcErr := NewService(context.Background(), cfg)
 	broker.contracts = contract.NewSingleContractProvider(broker.License, usage.NewNoop())
+	broker.storage = storage.NewInMemory(broker)
+	broker.storage.Configure(nil)
 	assert.NoError(t, svcErr)
 	defer broker.Close()
 	go broker.Listen()
@@ -178,17 +181,38 @@ func TestPubsub(t *testing.T) {
 		assert.Equal(t, mqtt.TypeOfPingresp, pkt.Type())
 	}
 
+	{ // Publish a retained message
+		msg := mqtt.Publish{
+			Header:  &mqtt.StaticHeader{QOS: 0, Retain: true},
+			Topic:   []byte("EbUlduEbUssgWueAWjkEZwdYG5YC0dGh/a/b/c/"),
+			Payload: []byte("retained message"),
+		}
+		_, err := msg.EncodeTo(cli)
+		assert.NoError(t, err)
+	}
+
 	{ // Subscribe to a topic
 		sub := mqtt.Subscribe{
 			Header: &mqtt.StaticHeader{QOS: 0},
 			Subscriptions: []mqtt.TopicQOSTuple{
-				{Topic: []byte("0Nq8SWbL8qoOKEDqh_ebBepug6cLLlWO/a/b/c/"), Qos: 0},
+				{Topic: []byte("EbUlduEbUssgWueAWjkEZwdYG5YC0dGh/a/b/c/"), Qos: 0},
 			},
 		}
 		_, err := sub.EncodeTo(cli)
 		assert.NoError(t, err)
 	}
 
+	{ // Read the retained message
+		pkt, err := mqtt.DecodePacket(cli)
+		assert.NoError(t, err)
+		assert.Equal(t, mqtt.TypeOfPublish, pkt.Type())
+		assert.Equal(t, &mqtt.Publish{
+			Header:  &mqtt.StaticHeader{QOS: 0},
+			Topic:   []byte("a/b/c/"),
+			Payload: []byte("retained message"),
+		}, pkt)
+	}
+
 	{ // Read suback
 		pkt, err := mqtt.DecodePacket(cli)
 		assert.NoError(t, err)
@@ -198,7 +222,7 @@ func TestPubsub(t *testing.T) {
 	{ // Publish a message
 		msg := mqtt.Publish{
 			Header:  &mqtt.StaticHeader{QOS: 0},
-			Topic:   []byte("0Nq8SWbL8qoOKEDqh_ebBepug6cLLlWO/a/b/c/"),
+			Topic:   []byte("EbUlduEbUssgWueAWjkEZwdYG5YC0dGh/a/b/c/"),
 			Payload: []byte("hello world"),
 		}
 		_, err := msg.EncodeTo(cli)
@@ -219,7 +243,7 @@ func TestPubsub(t *testing.T) {
 	{ // Publish a message but ignore ourselves
 		msg := mqtt.Publish{
 			Header:  &mqtt.StaticHeader{QOS: 0},
-			Topic:   []byte("0Nq8SWbL8qoOKEDqh_ebBepug6cLLlWO/a/b/c/?me=0"),
+			Topic:   []byte("EbUlduEbUssgWueAWjkEZwdYG5YC0dGh/a/b/c/?me=0"),
 			Payload: []byte("hello world"),
 		}
 		_, err := msg.EncodeTo(cli)
@@ -230,7 +254,7 @@ func TestPubsub(t *testing.T) {
 		sub := mqtt.Unsubscribe{
 			Header: &mqtt.StaticHeader{QOS: 0},
 			Topics: []mqtt.TopicQOSTuple{
-				{Topic: []byte("0Nq8SWbL8qoOKEDqh_ebBepug6cLLlWO/a/b/c/"), Qos: 0},
+				{Topic: []byte("EbUlduEbUssgWueAWjkEZwdYG5YC0dGh/a/b/c/"), Qos: 0},
 			},
 		}
 		_, err := sub.EncodeTo(cli)

internal/message/id.go

@@ -28,6 +28,9 @@ const (
 	fixed = 16
 )
 
+// RetainedTTL represents a TTL value to use for retained messages (max TTL).
+const RetainedTTL = math.MaxUint32
+
 var (
 	next   uint32
 	unique = newUnique()
@@ -62,7 +65,6 @@ func NewID(ssid Ssid) ID {
 // NewPrefix creates a new message identifier only containing the prefix.
 func NewPrefix(ssid Ssid, from int64) ID {
 	id := make(ID, 8)
-
 	binary.BigEndian.PutUint32(id[0:4], ssid[0]^ssid[1])
 	binary.BigEndian.PutUint32(id[4:8], math.MaxUint32-uint32(from-offset))
 	return id

internal/message/message.go

@@ -59,6 +59,11 @@ func (m *Message) Contract() uint32 {
 	return m.ID.Contract()
 }
 
+// Stored returns whether the message is or should be stored.
+func (m *Message) Stored() bool {
+	return m.TTL > 0
+}
+
 // Expires calculates the expiration time.
 func (m *Message) Expires() time.Time {
 	return time.Unix(m.Time(), 0).Add(time.Second * time.Duration(m.TTL))

internal/message/message_test.go

@@ -54,6 +54,7 @@ func TestNewMessage(t *testing.T) {
 	assert.Equal(t, Ssid{1, 2, 3}, m.Ssid())
 	assert.Equal(t, uint32(1), m.Contract())
 	assert.NotNil(t, m.Expires().String())
+	assert.False(t, m.Stored())
 }
 
 func TestNewFrame(t *testing.T) {