Added config flags for the new MQTT options

couchbase · May 17, 2024 · ece4fca · ece4fca
1 parent bd38d3d
commit ece4fca
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 5 deletions.
diff --git a/mqtt/server.go b/mqtt/server.go
@@ -30,7 +30,7 @@ const kDefaultMaximumSessionExpiryInterval = 60 * 60 * 24
 type ServerConfig struct {
 	Enabled                      *bool          `json:"enabled,omitempty"`
 	PublicInterface              string         `json:"public_interface,omitempty"  help:"Network interface to bind MQTT server to"`
-	MetadataDB                   *string        `json:"metadata_db,omitempty" help:"Name of database to persist MQTT state to"`
+	MetadataDB                   string         `json:"metadata_db,omitempty" help:"Name of database to persist MQTT state to"`
 	Cluster                      *ClusterConfig `json:"cluster,omitempty" help:"Cluster configuration (omit for no clustering)"`
 	MaximumMessageExpiryInterval int64          `json:"maximum_message_expiry_interval,omitempty" help:"Maximum message lifetime, in seconds; 0 means default"`
 	MaximumSessionExpiryInterval uint32         `json:"maximum_session_expiry_interval,omitempty" help:"Maximum disconnected session lifetime, in seconds; 0 means default"`
@@ -50,8 +50,8 @@ func (config *ServerConfig) IsEnabled() bool {
 }
 
 func (config *ServerConfig) ParseMetadataStore() (db string, ds sgbucket.DataStoreName, err error) {
-	if config.MetadataDB != nil {
-		pieces := strings.Split(*config.MetadataDB, sgbucket.ScopeCollectionSeparator)
+	if config.MetadataDB != "" {
+		pieces := strings.Split(config.MetadataDB, sgbucket.ScopeCollectionSeparator)
 		db = pieces[0]
 		scope := sgbucket.DefaultScope
 		collection := sgbucket.DefaultCollection

diff --git a/rest/config_flags.go b/rest/config_flags.go
@@ -64,6 +64,15 @@ func registerConfigFlags(config *StartupConfig, fs *flag.FlagSet) map[string]con
 		"api.cors.headers":      {&config.API.CORS.Headers, fs.String("api.cors.headers", "", "List of comma separated allowed headers")},
 		"api.cors.max_age":      {&config.API.CORS.MaxAge, fs.Int("api.cors.max_age", 0, "Maximum age of the CORS Options request")},
 
+		"api.mqtt.enabled":                         {&config.API.MQTT.Enabled, fs.Bool("api.mqtt.enabled", true, "Set to false to disable MQTT")},
+		"api.mqtt.public_interface":                {&config.API.MQTT.PublicInterface, fs.String("api.mqtt.public_interface", "", "Network interface to listen for MQTT connections")},
+		"api.mqtt.metadata_db":                     {&config.API.MQTT.MetadataDB, fs.String("api.mqtt.metadata_db", "", "Name of database to persist MQTT state to")},
+		"api.mqtt.maximum_message_expiry_interval": {&config.API.MQTT.MaximumMessageExpiryInterval, fs.Int64("api.mqtt.maximum_message_expiry_interval", 0, "Maximum message lifetime, in seconds; 0 means default")},
+		"api.mqtt.maximum_session_expiry_interval": {&config.API.MQTT.MaximumSessionExpiryInterval, fs.Int64("api.mqtt.maximum_session_expiry_interval", 0, "Maximum disconnected session lifetime, in seconds; 0 means default")},
+
+		"api.mqtt.cluster.enabled":           {&config.API.MQTT.Cluster.Enabled, fs.Bool("api.mqtt.cluster.enabled", true, "Set to false to disable MQTT clustering")},
+		"api.mqtt.cluster.discovery_address": {&config.API.MQTT.Cluster.DiscoveryAddr, fs.String("api.mqtt.cluster.discovery_address", "", "Address+port for peer discovery and gossip")},
+
 		"logging.log_file_path":   {&config.Logging.LogFilePath, fs.String("logging.log_file_path", "", "Absolute or relative path on the filesystem to the log file directory. A relative path is from the directory that contains the Sync Gateway executable file")},
 		"logging.redaction_level": {&config.Logging.RedactionLevel, fs.String("logging.redaction_level", "", "Redaction level to apply to log output. Options: none, partial, full, unset")},
 
@@ -138,6 +147,7 @@ func registerConfigFlags(config *StartupConfig, fs *flag.FlagSet) map[string]con
 		"unsupported.allow_dbconfig_env_vars":              {&config.Unsupported.AllowDbConfigEnvVars, fs.Bool("unsupported.allow_dbconfig_env_vars", true, "Can be set to false to skip environment variable expansion in database configs")},
 
 		"unsupported.user_queries": {&config.Unsupported.UserQueries, fs.Bool("unsupported.user_queries", false, "Whether user-query APIs are enabled")},
+		"unsupported.mqtt":         {&config.Unsupported.MQTT, fs.Bool("unsupported.mqtt", false, "Whether MQTT is enabled")},
 
 		"database_credentials": {&config.DatabaseCredentials, fs.String("database_credentials", "null", "JSON-encoded per-database credentials, that can be used instead of the bootstrap ones. This will override bucket_credentials that target the bucket that the database is in.")},
 		"bucket_credentials":   {&config.BucketCredentials, fs.String("bucket_credentials", "null", "JSON-encoded per-bucket credentials, that can be used instead of the bootstrap ones.")},

diff --git a/rest/config_startup.go b/rest/config_startup.go
@@ -230,7 +230,9 @@ func NewEmptyStartupConfig() StartupConfig {
 	return StartupConfig{
 		API: APIConfig{
 			CORS: &auth.CORSConfig{},
-			MQTT: &mqtt.ServerConfig{},
+			MQTT: &mqtt.ServerConfig{
+				Cluster: &mqtt.ClusterConfig{Enabled: base.BoolPtr(false)},
+			},
 		},
 		Logging: base.LoggingConfig{
 			Console: &base.ConsoleLoggerConfig{},

diff --git a/rest/mqtt_server.go b/rest/mqtt_server.go
@@ -33,7 +33,7 @@ func (sc *ServerContext) StartMQTTServer(ctx context.Context, config *mqtt.Serve
 
 	// Metadata store:
 	var metadataStore sgbucket.DataStore
-	if config.MetadataDB != nil {
+	if config.MetadataDB != "" {
 		dbName, dsName, err := config.ParseMetadataStore()
 		if err != nil {
 			return err