feature(persistence): adjust storage (#24)

* feature(goroutine): add goroutine

* feature(persistence): adjust storage

* feature(persistence): adjust storage

* feature(persistence): update storage

Author: chenquan

cmd/lighthouse/config.yaml

@@ -3,4 +3,10 @@ mqtt:
 log:
   level: debug
   format: json
-  filename: log.log
+  filename: log.log
+persistence:
+  session:
+    type: memory
+  queue:
+    type: memory
+

cmd/lighthouse/main.go

@@ -4,10 +4,13 @@ import (
 	_ "embed"
 	"github.com/go-playground/validator/v10"
 	"github.com/yunqi/lighthouse/config"
+	_ "github.com/yunqi/lighthouse/internal/persistence/session/memory"
+	_ "github.com/yunqi/lighthouse/internal/persistence/session/redis"
 	"github.com/yunqi/lighthouse/internal/server"
 	"github.com/yunqi/lighthouse/internal/xlog"
 	"gopkg.in/yaml.v3"
 	"net/http"
+	_ "net/http/pprof"
 )
 
 //go:embed config.yaml
@@ -30,9 +33,9 @@ func main() {
 		panic(err)
 	}
 	go func() {
-		http.ListenAndServe("localhost:6060", nil)
+		_ = http.ListenAndServe("localhost:6060", nil)
 	}()
 
-	newServer := server.NewServer(server.WithTcpListen(":1883"))
+	newServer := server.NewServer(server.WithTcpListen(":1883"), server.WithPersistence(&c.Persistence))
 	newServer.ServeTCP()
 }

config/config.go

@@ -21,61 +21,62 @@ import (
 )
 
 type Config struct {
-	Mqtt Mqtt `yaml:"mqtt"`
-	Log  Log  `yaml:"log"`
+	Mqtt        Mqtt        `yaml:"mqtt"`
+	Log         Log         `yaml:"log"`
+	Persistence Persistence `yaml:"persistence"`
 }
 
 type Mqtt struct {
 	// SessionExpiry is the maximum session expiry interval in seconds.
-	SessionExpiry time.Duration `yaml:"session_expiry"`
+	SessionExpiry time.Duration `yaml:"sessionExpiry"`
 	// SessionExpiryCheckInterval is the interval time for session expiry checker to check whether there
 	// are expired sessions.
-	SessionExpiryCheckInterval time.Duration `yaml:"session_expiry_check_interval"`
+	SessionExpiryCheckInterval time.Duration `yaml:"sessionExpiryCheckInterval"`
 	// MessageExpiry is the maximum lifetime of the message in seconds.
 	// If a message in the queue is not sent in MessageExpiry time, it will be removed, which means it will not be sent to the subscriber.
-	MessageExpiry time.Duration `yaml:"message_expiry"`
+	MessageExpiry time.Duration `yaml:"messageExpiry"`
 	// InflightExpiry is the lifetime of the "inflight" message in seconds.
 	// If a "inflight" message is not acknowledged by a client in InflightExpiry time, it will be removed when the message queue is full.
-	InflightExpiry time.Duration `yaml:"inflight_expiry"`
+	InflightExpiry time.Duration `yaml:"inflightExpiry"`
 	// MaxPacketSize is the maximum packet size that the server is willing to accept from the client
-	MaxPacketSize uint32 `yaml:"max_packet_size"`
+	MaxPacketSize uint32 `yaml:"maxPacketSize"`
 	// ReceiveMax limits the number of QoS 1 and QoS 2 publications that the server is willing to process concurrently for the client.
-	ReceiveMax uint16 `yaml:"server_receive_maximum"`
+	ReceiveMax uint16 `yaml:"serverReceiveMaximum"`
 	// MaxKeepAlive is the maximum keep alive time in seconds allows by the server.
 	// If the client requests a keepalive time bigger than MaxKeepalive,
 	// the server will use MaxKeepAlive as the keepalive time.
 	// In this case, if the client version is v5, the server will set MaxKeepalive into CONNACK to inform the client.
 	// But if the client version is 3.x, the server has no way to inform the client that the keepalive time has been changed.
-	MaxKeepAlive uint16 `yaml:"max_keepalive"`
+	MaxKeepAlive uint16 `yaml:"maxKeepalive"`
 	// TopicAliasMax indicates the highest value that the server will accept as a Topic Alias sent by the client.
 	// No-op if the client version is MQTTv3.x
-	TopicAliasMax uint16 `yaml:"topic_alias_maximum"`
+	TopicAliasMax uint16 `yaml:"topicAliasMaximum"`
 	// SubscriptionIDAvailable indicates whether the server supports Subscription Identifiers.
 	// No-op if the client version is MQTTv3.x .
-	SubscriptionIDAvailable bool `yaml:"subscription_identifier_available"`
+	SubscriptionIDAvailable bool `yaml:"subscriptionIdentifierAvailable"`
 	// SharedSubAvailable indicates whether the server supports Shared Subscriptions.
-	SharedSubAvailable bool `yaml:"shared_subscription_available"`
+	SharedSubAvailable bool `yaml:"sharedSubscriptionAvailable"`
 	// WildcardSubAvailable indicates whether the server supports Wildcard Subscriptions.
-	WildcardAvailable bool `yaml:"wildcard_subscription_available"`
+	WildcardAvailable bool `yaml:"wildcardSubscriptionAvailable"`
 	// RetainAvailable indicates whether the server supports retained messages.
-	RetainAvailable bool `yaml:"retain_available"`
+	RetainAvailable bool `yaml:"retainAvailable"`
 	// MaxQueuedMsg is the maximum queue length of the outgoing messages.
 	// If the queue is full, some message will be dropped.
 	// The message dropping strategy is described in the document of the persistence/queue.Store interface.
-	MaxQueueMessages int `yaml:"max_queue_messages"`
+	MaxQueueMessages int `yaml:"maxQueueMessages"`
 	// MaxInflight limits inflight message length of the outgoing messages.
 	// Inflight message is also stored in the message queue, so it must be less than or equal to MaxQueuedMsg.
 	// Inflight message is the QoS 1 or QoS 2 message that has been sent out to a client but not been acknowledged yet.
-	MaxInflight uint16 `yaml:"max_inflight"`
+	MaxInflight uint16 `yaml:"maxInflight"`
 	// MaximumQoS is the highest QOS level permitted for a Publish.
-	MaximumQoS uint8 `yaml:"maximum_qos"`
+	MaximumQoS uint8 `yaml:"maximumQos"`
 	// QueueQos0Msg indicates whether to store QoS 0 message for a offline session.
-	QueueQos0Msg bool `yaml:"queue_qos0_messages"`
+	QueueQos0Msg bool `yaml:"queueQos0Messages"`
 	// DeliveryMode is the delivery mode. The possible value can be "overlap" or "onlyonce".
 	// It is possible for a client’s subscriptions to overlap so that a published message might match multiple filters.
 	// When set to "overlap" , the server will deliver one message for each matching subscription and respecting the subscription’s QoS in each case.
 	// When set to "onlyOnce",the server will deliver the message to the client respecting the maximum QoS of all the matching subscriptions.
-	DeliveryMode string `yaml:"delivery_mode"`
+	DeliveryMode string `yaml:"deliveryMode"`
 	// AllowZeroLenClientId indicates whether to allow a client to connect with empty client id.
-	AllowZeroLenClientId bool `yaml:"allow_zero_len_client_id"`
+	AllowZeroLenClientId bool `yaml:"allowZeroLenClientId"`
 }

config/config.yaml

@@ -0,0 +1,42 @@
+package config
+
+import "time"
+
+type (
+	Persistence struct {
+		Session StoreType `yaml:"session"`
+		Queue   StoreType `yaml:"queue"`
+	}
+
+	StoreType struct {
+		Type  string         `yaml:"type"` // memory|redis
+		Redis RedisStoreType `yaml:"redis"`
+	}
+
+	RedisStoreType struct {
+		Type string `yaml:"nodeType"`
+		// Addr is the redis server address.
+		// If empty, use "127.0.0.1:6379" as default.
+		Addr string `yaml:"addr"`
+		// Password is the redis password.
+		Password string `yaml:"password"`
+		// Database is the number of the redis database to be connected.
+		Database uint `yaml:"database"`
+		// MaxIdle is the maximum number of idle connections in the pool.
+		// If nil, use 1000 as default.
+		// This value will pass to redis.Pool.MaxIde.
+		MaxIdle *uint `yaml:"maxIdle"`
+		// MaxActive is the maximum number of connections allocated by the pool at a given time.
+		// If nil, use 0 as default.
+		// If zero, there is no limit on the number of connections in the pool.
+		// This value will pass to redis.Pool.MaxActive.
+		MaxActive *uint `yaml:"maxActive"`
+		// Close connections after remaining idle for this duration. If the value
+		// is zero, then idle connections are not closed. Applications should set
+		// the timeout to a value less than the server's timeout.
+		// Ff zero, use 240 * time.Second as default.
+		// This value will pass to redis.Pool.IdleTimeout.
+		IdleTimeout time.Duration `yaml:"idleTimeout"`
+		Timeout     time.Duration `yaml:"timeout"`
+	}
+)

config/config_test.go

@@ -3,30 +3,32 @@ module github.com/yunqi/lighthouse
 go 1.17
 
 require (
-	github.com/chenquan/go-pkg v0.1.9
+	github.com/bytedance/gopkg v0.0.0-20220118075514-1372042b2bbc
+	github.com/chenquan/go-pkg v0.1.18
+	github.com/go-playground/validator/v10 v10.10.0
 	github.com/go-redis/redis/v8 v8.11.4
 	github.com/golang/mock v1.6.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/panjf2000/ants/v2 v2.4.7
 	github.com/stretchr/testify v1.7.0
 	go.uber.org/zap v1.19.0
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
+	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 )
 
 require (
+	github.com/BurntSushi/toml v1.0.0 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/go-playground/locales v0.14.0 // indirect
 	github.com/go-playground/universal-translator v0.18.0 // indirect
-	github.com/go-playground/validator/v10 v10.10.0 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97 // indirect
-	golang.org/x/sys v0.0.0-20210806184541-e5e7981a1069 // indirect
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
+	golang.org/x/sys v0.0.0-20220110181412-a018aaa089fe // indirect
 	golang.org/x/text v0.3.6 // indirect
-	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 )