Merge pull request #40 from c16a/feature/persistence

Add persistence provider

.github/workflows/docker.yml

@@ -29,7 +29,7 @@ jobs:
         with:
           context: .
           file: ./Dockerfile
-          platforms: linux/amd64
+          platforms: linux/amd64,linux/arm64
           username: $GITHUB_ACTOR
           password: ${{ secrets.CR_PAT }}
           push: true

app/main.go

@@ -4,6 +4,7 @@ import (
 	"github.com/c16a/hermes/lib/config"
 	"github.com/c16a/hermes/lib/mqtt"
 	"github.com/c16a/hermes/lib/transports"
+	"go.uber.org/zap"
 	"log"
 	"os"
 )
@@ -12,16 +13,21 @@ func main() {
 
 	configFilePath := os.Getenv("CONFIG_FILE_PATH")
 
+	logger, err := zap.NewProduction()
+	if err != nil {
+		log.Fatal(err)
+	}
+
 	serverConfig, err := config.ParseConfig(configFilePath)
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	ctx, err := mqtt.NewServerContext(serverConfig)
+	ctx, err := mqtt.NewServerContext(serverConfig, logger)
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	go transports.StartWebSocketServer(serverConfig, ctx)
-	transports.StartTcpServer(serverConfig, ctx)
+	go transports.StartWebSocketServer(serverConfig, ctx, logger)
+	transports.StartTcpServer(serverConfig, ctx, logger)
 }

docs/java.md

@@ -48,7 +48,12 @@ repositories {
 ```java
 var persistence = new MemoryPersistence();
 var client = new MqttClient(broker, clientID, persistence);
+
 var connOpts = new MqttConnectionOptions();
+// Setting clean start to "false" enables the client 
+// to receive offline messages send while it was disconnected.
+connOpts.setCleanStart(false);
+
 client.connect(connOpts);
 ```
 
@@ -71,13 +76,21 @@ client.setCallback(new MqttCallback(){
     public void messageArrived(String topic, MqttMessage message) throws Exception{
         // Do something awesome
     }
+
+    // other implemented methods
 });
 client.connect(connOpts);
 
 // Provide a topic and Quality of Service (QoS)
 client.subscribe("my-topic", 0);
 ```
 
+### Closing the connection
+
+```java
+client.disconnect();
+```
+
 ## Spring Integration
 
 Spring Integration provides inbound and outbound channel adapters to support the MQTT protocol.

go.mod

@@ -6,24 +6,30 @@ require (
 	github.com/dgraph-io/badger/v2 v2.2007.4
 	github.com/eclipse/paho.golang v0.10.0
 	github.com/go-ldap/ldap/v3 v3.4.1
+	github.com/go-redis/redis/v8 v8.11.3
 	github.com/gorilla/websocket v1.4.2
 	github.com/satori/go.uuid v1.2.0
-	github.com/sirupsen/logrus v1.8.1
+	go.uber.org/zap v1.19.1
 )
 
 require (
 	github.com/Azure/go-ntlmssp v0.0.0-20200615164410-66371956d46c // indirect
 	github.com/cespare/xxhash v1.1.0 // indirect
+	github.com/cespare/xxhash/v2 v2.1.1 // indirect
 	github.com/dgraph-io/ristretto v0.0.3-0.20200630154024-f66de99634de // indirect
 	github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/dustin/go-humanize v1.0.0 // indirect
 	github.com/go-asn1-ber/asn1-ber v1.5.1 // indirect
-	github.com/golang/protobuf v1.3.1 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.3 // indirect
 	github.com/klauspost/compress v1.12.3 // indirect
 	github.com/pkg/errors v0.8.1 // indirect
-	golang.org/x/crypto v0.0.0-20200604202706-70a84ac30bf9 // indirect
-	golang.org/x/net v0.0.0-20190620200207-3b0461eec859 // indirect
-	golang.org/x/sync v0.0.0-20201207232520-09787c993a3a // indirect
-	golang.org/x/sys v0.0.0-20191026070338-33540a1f6037 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.6.0 // indirect
+	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 // indirect
+	golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 // indirect
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
+	golang.org/x/sys v0.0.0-20210510120138-977fb7262007 // indirect
+	google.golang.org/protobuf v1.26.0 // indirect
 )

lib/config/config.go

@@ -7,12 +7,12 @@ type Config struct {
 
 // Server stores all server related configuration
 type Server struct {
-	Tls         *Tls     `json:"tls" yaml:"tls"`
-	TcpAddress  string   `json:"tcp,omitempty" yaml:"tcp,omitempty"`
-	HttpAddress string   `json:"http,omitempty" yaml:"http,omitempty"`
-	MaxQos      byte     `json:"max_qos,omitempty" yaml:"max_qos,omitempty"`
-	Auth        *Auth    `json:"auth,omitempty" yaml:"auth,omitempty"`
-	Offline     *Offline `json:"offline,omitempty"`
+	Tls         *Tls         `json:"tls" yaml:"tls"`
+	TcpAddress  string       `json:"tcp,omitempty" yaml:"tcp,omitempty"`
+	HttpAddress string       `json:"http,omitempty" yaml:"http,omitempty"`
+	MaxQos      byte         `json:"max_qos,omitempty" yaml:"max_qos,omitempty"`
+	Auth        *Auth        `json:"auth,omitempty" yaml:"auth,omitempty"`
+	Persistence *Persistence `json:"persistence,omitempty" yaml:"persistence,omitempty"`
 }
 
 // Tls stores the TLS config for the server
@@ -28,8 +28,20 @@ type Auth struct {
 	LdapDn   string `json:"ldap_dn,omitempty" yaml:"ldap_dn,omitempty"`
 }
 
-type Offline struct {
+type Badger struct {
 	Path         string `json:"path,omitempty"`
 	MaxTableSize int64  `json:"max_table_size,omitempty"`
 	NumTables    int    `json:"num_tables,omitempty"`
 }
+
+type Redis struct {
+	Url      string `json:"url" yaml:"url"`
+	Password string `json:"password" yaml:"password"`
+}
+
+type Persistence struct {
+	Type string `json:"type" yaml:"type"`
+
+	Badger *Badger `json:"badger" yaml:"badger"`
+	Redis  *Redis  `json:"redis" yaml:"redis"`
+}

lib/mqtt/conn_handler.go

@@ -3,7 +3,7 @@ package mqtt
 import "net"
 
 func HandleMqttConnection(conn net.Conn, ctx *ServerContext) {
-	handler := &MqttHandler{base: ctx}
+	handler := &MqttHandler{base: ctx, logger: ctx.logger}
 
 	for true {
 		handler.Handle(conn)