Add a persistence hook based on cockroachdb/pebble. (#378)

* Implementing Pebble as a persistence database hook.

* Fixed failing test cases.

* Add Pebble DB configuration for file-based configuration, optimize part of the code.

* Resolve test failure issues and perform code optimization.

* Optimized the test cases.

Author: werbenhu

README-CN.md

@@ -45,7 +45,7 @@ MQTT 代表 MQ Telemetry Transport。它是一种发布/订阅、非常简单和
     - 通过所有 [Paho互操作性测试](https://github.com/eclipse/paho.mqtt.testing/tree/master/interoperability)（MQTT v5 和 MQTT v3）。
     - 超过一千多个经过仔细考虑的单元测试场景。
 - 支持 TCP、Websocket（包括 SSL/TLS）和$SYS 服务状态监控。
-- 内置 基于Redis、Badger 和 Bolt 的持久化（使用Hook钩子，你也可以自己创建）。
+- 内置 基于Redis、Badger、Pebble 和 Bolt 的持久化（使用Hook钩子，你也可以自己创建）。
 - 内置基于规则的认证和 ACL 权限管理（使用Hook钩子，你也可以自己创建）。
 
 ### 兼容性说明(Compatibility Notes)
@@ -227,6 +227,7 @@ server := mqtt.New(&mqtt.Options{
 | 访问控制 | [mochi-mqtt/server/hooks/auth . Auth](hooks/auth/auth.go)                | 基于规则的访问权限控制。  | 
 | 数据持久性    | [mochi-mqtt/server/hooks/storage/bolt](hooks/storage/bolt/bolt.go)       | 使用 [BoltDB](https://dbdb.io/db/boltdb) 进行持久性存储（已弃用）。 | 
 | 数据持久性    | [mochi-mqtt/server/hooks/storage/badger](hooks/storage/badger/badger.go) | 使用 [BadgerDB](https://github.com/dgraph-io/badger) 进行持久性存储。   | 
+| 数据持久性    | [mochi-mqtt/server/hooks/storage/pebble](hooks/storage/pebble/pebble.go) | 使用 [PebbleDB](https://github.com/cockroachdb/pebble) 进行持久性存储。   | 
 | 数据持久性    | [mochi-mqtt/server/hooks/storage/redis](hooks/storage/redis/redis.go)    | 使用 [Redis](https://redis.io) 进行持久性存储。                         | 
 | 调试跟踪      | [mochi-mqtt/server/hooks/debug](hooks/debug/debug.go)                    | 调试输出以查看数据包在服务端的链路追踪。   |
 
@@ -329,9 +330,25 @@ if err != nil {
 ```
 有关 Redis 钩子的工作原理或如何使用它的更多信息，请参阅  [examples/persistence/redis/main.go](examples/persistence/redis/main.go) 或 [hooks/storage/redis](hooks/storage/redis) 。
 
+#### Pebble DB
+
+如果您更喜欢基于文件的存储，还有一个 PebbleDB 存储钩子(Hook)可用。它可以以与其他钩子大致相同的方式添加和配置（具有较少的选项）。
+
+```go
+err := server.AddHook(new(pebble.Hook), &pebble.Options{
+  Path: pebblePath,
+  Mode: pebble.NoSync,
+})
+if err != nil {
+  log.Fatal(err)
+}
+```
+
+有关 pebble 钩子(Hook)的工作原理或如何使用它的更多信息，请参阅 [examples/persistence/pebble/main.go](examples/persistence/pebble/main.go) 或 [hooks/storage/pebble](hooks/storage/pebble)。
+
 #### Badger DB
 
-如果您更喜欢基于文件的存储，还有一个 BadgerDB 存储钩子(Hook)可用。它可以以与其他钩子大致相同的方式添加和配置（具有较少的选项）。
+同样是基于文件的存储，还有一个 BadgerDB 存储钩子(Hook)可用。它可以以与其他钩子大致相同的方式添加和配置。
 
 ```go
 err := server.AddHook(new(badger.Hook), &badger.Options{

README.md

@@ -45,7 +45,7 @@ MQTT stands for [MQ Telemetry Transport](https://en.wikipedia.org/wiki/MQTT). It
     - Passes all [Paho Interoperability Tests](https://github.com/eclipse/paho.mqtt.testing/tree/master/interoperability) for MQTT v5 and MQTT v3.
     - Over a thousand carefully considered unit test scenarios.
 - TCP, Websocket (including SSL/TLS), and $SYS Dashboard listeners.
-- Built-in Redis, Badger, and Bolt Persistence using Hooks (but you can also make your own).
+- Built-in Redis, Badger, Pebble and Bolt Persistence using Hooks (but you can also make your own).
 - Built-in Rule-based Authentication and ACL Ledger using Hooks (also make your own).
 
 ### Compatibility Notes
@@ -228,6 +228,7 @@ Hooks are stackable - you can add multiple hooks to a server, and they will be r
 | Access Control | [mochi-mqtt/server/hooks/auth . Auth](hooks/auth/auth.go)                | Rule-based access control ledger.                                          | 
 | Persistence    | [mochi-mqtt/server/hooks/storage/bolt](hooks/storage/bolt/bolt.go)       | Persistent storage using [BoltDB](https://dbdb.io/db/boltdb) (deprecated). | 
 | Persistence    | [mochi-mqtt/server/hooks/storage/badger](hooks/storage/badger/badger.go) | Persistent storage using [BadgerDB](https://github.com/dgraph-io/badger).  | 
+| Persistence    | [mochi-mqtt/server/hooks/storage/pebble](hooks/storage/pebble/pebble.go) | Persistent storage using [PebbleDB](https://github.com/cockroachdb/pebble).  | 
 | Persistence    | [mochi-mqtt/server/hooks/storage/redis](hooks/storage/redis/redis.go)    | Persistent storage using [Redis](https://redis.io).                        | 
 | Debugging      | [mochi-mqtt/server/hooks/debug](hooks/debug/debug.go)                    | Additional debugging output to visualise packet flow.                      | 
 
@@ -322,8 +323,21 @@ if err != nil {
 ```
 For more information on how the redis hook works, or how to use it, see the [examples/persistence/redis/main.go](examples/persistence/redis/main.go) or [hooks/storage/redis](hooks/storage/redis) code.
 
+#### Pebble DB
+There's also a Pebble Db storage hook if you prefer file-based storage. It can be added and configured in much the same way as the other hooks (with somewhat less options).
+```go
+err := server.AddHook(new(pebble.Hook), &pebble.Options{
+  Path: pebblePath,
+  Mode: pebble.NoSync,
+})
+if err != nil {
+  log.Fatal(err)
+}
+```
+For more information on how the pebble hook works, or how to use it, see the [examples/persistence/pebble/main.go](examples/persistence/pebble/main.go) or [hooks/storage/pebble](hooks/storage/pebble) code.
+
 #### Badger DB
-There's also a BadgerDB storage hook if you prefer file based storage. It can be added and configured in much the same way as the other hooks (with somewhat less options).
+Similarly, for file-based storage, there is also a BadgerDB storage hook available. It can be added and configured in much the same way as the other hooks.
 ```go
 err := server.AddHook(new(badger.Hook), &badger.Options{
   Path: badgerPath,

config/config.go

@@ -6,10 +6,12 @@ package config
 
 import (
 	"encoding/json"
+
 	"github.com/mochi-mqtt/server/v2/hooks/auth"
 	"github.com/mochi-mqtt/server/v2/hooks/debug"
 	"github.com/mochi-mqtt/server/v2/hooks/storage/badger"
 	"github.com/mochi-mqtt/server/v2/hooks/storage/bolt"
+	"github.com/mochi-mqtt/server/v2/hooks/storage/pebble"
 	"github.com/mochi-mqtt/server/v2/hooks/storage/redis"
 	"github.com/mochi-mqtt/server/v2/listeners"
 	"gopkg.in/yaml.v3"
@@ -41,6 +43,7 @@ type HookAuthConfig struct {
 type HookStorageConfig struct {
 	Badger *badger.Options `yaml:"badger" json:"badger"`
 	Bolt   *bolt.Options   `yaml:"bolt" json:"bolt"`
+	Pebble *pebble.Options `yaml:"pebble" json:"pebble"`
 	Redis  *redis.Options  `yaml:"redis" json:"redis"`
 }
 
@@ -111,6 +114,13 @@ func (hc HookConfigs) toHooksStorage() []mqtt.HookLoadConfig {
 			Config: hc.Storage.Redis,
 		})
 	}
+
+	if hc.Storage.Pebble != nil {
+		hlc = append(hlc, mqtt.HookLoadConfig{
+			Hook:   new(pebble.Hook),
+			Config: hc.Storage.Pebble,
+		})
+	}
 	return hlc
 }
 

config/config_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/mochi-mqtt/server/v2/hooks/auth"
 	"github.com/mochi-mqtt/server/v2/hooks/storage/badger"
 	"github.com/mochi-mqtt/server/v2/hooks/storage/bolt"
+	"github.com/mochi-mqtt/server/v2/hooks/storage/pebble"
 	"github.com/mochi-mqtt/server/v2/hooks/storage/redis"
 	"github.com/mochi-mqtt/server/v2/listeners"
 
@@ -210,3 +211,23 @@ func TestToHooksStorageRedis(t *testing.T) {
 
 	require.Equal(t, expect, th)
 }
+
+func TestToHooksStoragePebble(t *testing.T) {
+	hc := HookConfigs{
+		Storage: &HookStorageConfig{
+			Pebble: &pebble.Options{
+				Path: "pebble",
+			},
+		},
+	}
+
+	th := hc.toHooksStorage()
+	expect := []mqtt.HookLoadConfig{
+		{
+			Hook:   new(pebble.Hook),
+			Config: hc.Storage.Pebble,
+		},
+	}
+
+	require.Equal(t, expect, th)
+}

examples/config/config.json

@@ -21,6 +21,10 @@
       "enable": true
     },
     "storage": {
+      "pebble": {
+        "path": "pebble.db",
+        "mode": "NoSync"
+      },
       "badger": {
         "path": "badger.db",
         "gc_interval": 3,

examples/config/config.yaml

@@ -16,6 +16,9 @@ hooks:
       path: badger.db
       gc_interval: 3
       gc_discard_ratio: 0.5
+    pebble:
+      path: pebble.db
+      mode: "NoSync"
     bolt:
       path: bolt.db
     redis:

examples/persistence/pebble/main.go

@@ -0,0 +1,62 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2022 mochi-mqtt, mochi-co, werbenhu
+// SPDX-FileContributor: werbenhu
+
+package main
+
+import (
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	mqtt "github.com/mochi-mqtt/server/v2"
+	"github.com/mochi-mqtt/server/v2/hooks/auth"
+	"github.com/mochi-mqtt/server/v2/hooks/storage/pebble"
+	"github.com/mochi-mqtt/server/v2/listeners"
+)
+
+func main() {
+	pebblePath := ".pebble"
+	defer os.RemoveAll(pebblePath) // remove the example pebble files at the end
+
+	sigs := make(chan os.Signal, 1)
+	done := make(chan bool, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		<-sigs
+		done <- true
+	}()
+
+	server := mqtt.New(nil)
+	_ = server.AddHook(new(auth.AllowHook), nil)
+
+	err := server.AddHook(new(pebble.Hook), &pebble.Options{
+		Path: pebblePath,
+		Mode: pebble.NoSync,
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	tcp := listeners.NewTCP(listeners.Config{
+		ID:      "t1",
+		Address: ":1883",
+	})
+	err = server.AddListener(tcp)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	go func() {
+		err := server.Serve()
+		if err != nil {
+			log.Fatal(err)
+		}
+	}()
+
+	<-done
+	server.Log.Warn("caught signal, stopping...")
+	_ = server.Close()
+	server.Log.Info("main.go finished")
+}

go.mod

@@ -6,32 +6,51 @@ require (
 	github.com/alicebob/miniredis/v2 v2.23.0
 	github.com/asdine/storm v2.1.2+incompatible
 	github.com/asdine/storm/v3 v3.2.1
+	github.com/cockroachdb/pebble v1.1.0
+	github.com/dgraph-io/badger v1.6.0
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/gorilla/websocket v1.5.0
 	github.com/jinzhu/copier v0.3.5
 	github.com/rs/xid v1.4.0
-	github.com/stretchr/testify v1.7.1
+	github.com/stretchr/testify v1.8.1
 	github.com/timshannon/badgerhold v1.0.0
 	go.etcd.io/bbolt v1.3.5
 	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
 	github.com/AndreasBriese/bbloom v0.0.0-20190825152654-46b345b51c96 // indirect
+	github.com/DataDog/zstd v1.4.5 // indirect
 	github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a // indirect
-	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cockroachdb/errors v1.11.1 // indirect
+	github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect
+	github.com/cockroachdb/redact v1.1.5 // indirect
+	github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/dgraph-io/badger v1.6.0 // 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/golang/protobuf v1.5.0 // indirect
-	github.com/golang/snappy v0.0.3 // indirect
-	github.com/google/go-cmp v0.5.8 // indirect
+	github.com/getsentry/sentry-go v0.18.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/klauspost/compress v1.15.15 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_golang v1.12.0 // indirect
+	github.com/prometheus/client_model v0.2.1-0.20210607210712-147c58e9608a // indirect
+	github.com/prometheus/common v0.32.1 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	github.com/yuin/gopher-lua v0.0.0-20210529063254-f4c35e4016d9 // indirect
+	golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df // indirect
 	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
 )