complete full refactor, added a lot of features including binary messages through zeromq, fixed the rpc manager, major cleanup

Author: Jimmy

.gitmodules

@@ -1,6 +1,3 @@
-[submodule "generated/proto/ticker"]
-	path = generated/proto/ticker
-	url = https://github.com/RoseLabsMx/ftso-data-sources-proto-files.git
 [submodule "datasource/cryptocurrency/mexc/protobuf"]
 	path = datasource/cryptocurrency/mexc/protobuf
 	url = https://github.com/mexcdevelop/websocket-proto.git

README.md

@@ -14,7 +14,7 @@ Flow goes like this:
 # Supported data sources:
 ### Crypto:
     binance,binance.us,bitfinex,bitget,bitmart,bitrue,bitstamp,bybit,coinex,coinbase,cryptocom,digifinex,fmfw,gateio,
-    hitbtc,huobi,kraken,kucoin,lbank,mexc,okx,pionex,toobit,whitebit,xt
+    hitbtc,htx,kraken,kucoin,lbank,mexc,okx,pionex,toobit,whitebit,xt
 
 ### Stocks, Commodities, Forex:
     tiingo,metalsdev
@@ -28,11 +28,9 @@ Flow goes like this:
 `git clone https://github.com/LightFTSO/ftso-data-sources.git`
 2. cd into it
 `cd ftso-data-sources`
-3. (optionally clone submodules for protobuf codegen)
-`git submodule update --init --recursive`
-4. Create the config.yaml (see the section below)
+3. Create the config.yaml (see the section below)
 `touch config.yaml`
-5. Run it 
+4. Run it 
 Locally with `make run`:
 You need go 1.22+ installed, see https://go.dev/doc/install
 ```bash
@@ -43,11 +41,7 @@ make run
 Run in docker:
 `docker compose up -d ftso-data-sources`
 
-Supported go version: 1.22+
-
-# Protobuf Definitions and Codegen
-
-See https://github.com/RoseLabsMx/ftso-data-sources-proto-files
+Supported go version: 1.25.5+
 
 # Configuration 
 Modify the following sample configuration, by default, the program will look for a file called `config.yaml` in it's root folder or you can specify the file with the `-config <file>` flag
@@ -78,7 +72,7 @@ datasources:
   - source: fmfw
   - source: gateio
   - source: hitbtc
-  - source: huobi
+  - source: htx
   - source: kucoin
   - source: kraken
   - source: lbank
@@ -105,10 +99,10 @@ stats:
   interval: 60s
 
 # see https://mqtt.org/
-mqtt:
-  enabled: false
-  url: "tcp://localhost:1883"
-  qos_level: 1
+zeromq:
+    enabled: false
+    port: 9998
+    flushinterval: 5ms
 
 # See https://redis.io/docs/latest/develop/data-types/timeseries/
 redis_ts:
@@ -132,6 +126,8 @@ file_output:
 websocket_server:
   enabled: false
   ticker_endpoint: /tickers
+  flush_interval: 50ms
+  serialization_protocol: json
 
 assets:
   forex:
@@ -338,15 +334,6 @@ Response:
 }
 ```
 
-# Counting ticker rate
-You can count the number of tickers per second enabling the file-output consumer, using /dev/stdout as the output file
-or using the MQTT consumer, in another terminal connect to it using a client program and pipe the output to the program `pv`, e.g.:
-
-`./ftso-data-sources | pv --line-mode --timer --rate > /dev/null`
-Outputs:
-`03:22 [ 496 /s]`
-For more info on `pv`, visit [https://docs.oracle.com/cd/E86824_01/html/E54763/pv-1.html](Oracle's man pages)
-
 # Contributing
 
 Contributions are welcome! Please feel free to submit a PR.

config/config.go

@@ -1,7 +1,8 @@
 package config
 
 import (
-	"log/slog"
+	"fmt"
+	"os"
 	"path"
 
 	"github.com/spf13/viper"
@@ -35,7 +36,7 @@ type ConfigOptions struct {
 	RedisOptions             consumer.RedisOptions             `mapstructure:"redis_ts"`
 	WebsocketConsumerOptions consumer.WebsocketConsumerOptions `mapstructure:"websocket_server"`
 	FileConsumerOptions      consumer.FileConsumerOptions      `mapstructure:"file_output"`
-	MQTTConsumerOptions      consumer.MqttConsumerOptions      `mapstructure:"mqtt"`
+	ZMQConsumerOptions       consumer.ZMQConsumerOptions       `mapstructure:"zeromq"`
 
 	TickerTransformationOptions []tickertopic.TransformationOptions `mapstructure:"ticker_transformations"`
 }
@@ -64,15 +65,27 @@ func LoadConfig(configFile string) (config ConfigOptions, err error) {
 		return ConfigOptions{}, err
 	}
 
-	config.WebsocketConsumerOptions.Port = config.Port
-
 	Config = config
 	return config, nil
 }
 
-func SaveConfig() error {
-	slog.Info("Saving current configuraton to backup file")
-	err := viper.WriteConfigAs("config.original.yaml")
+// SafeWriteConfig writes the config to a temp file and atomically moves it
+// to the destination. This prevents file corruption on crash.
+func WriteConfig(filename string) error {
+	// 1. Define a temporary filename (e.g., config.yaml.tmp)
+	tempFile := "tmp." + filename
+
+	// 2. Write the viper config to the temporary file
+	if err := viper.WriteConfigAs(tempFile); err != nil {
+		return fmt.Errorf("failed to write temp config: %w", err)
+	}
+
+	// 3. Atomically rename temp file to original file
+	// This operation is atomic on POSIX systems (Linux/macOS) and safe on Windows
+	// (if the file exists, it is replaced).
+	if err := os.Rename(tempFile, filename); err != nil {
+		return fmt.Errorf("failed to move temp config to %s: %w", filename, err)
+	}
 
-	return err
+	return nil
 }

config/defaults.go

@@ -12,7 +12,7 @@ func setDefaults() {
 	viper.SetDefault("message_buffer_size", 0)
 	viper.SetDefault("port", 9999)
 
-	viper.SetDefault("assets.crypto", constants.BASES_CRYPTO)
+	viper.SetDefault("assets.crypto", constants.AssetList{})
 	viper.SetDefault("assets.commodities", constants.AssetList{})
 	viper.SetDefault("assets.forex", constants.AssetList{})
 	viper.SetDefault("assets.stocks", constants.AssetList{})
@@ -25,9 +25,6 @@ func setDefaults() {
 	viper.SetDefault("file_output.enabled", false)
 	viper.SetDefault("file_output.filename", "")
 
-	viper.SetDefault("mqtt.enabled", false)
-	viper.SetDefault("mqtt.qos_level", 0)
-
 	viper.SetDefault("redis_ts.enabled", false)
 	viper.SetDefault("redis_ts.client_options.initaddress", []string{"127.0.0.1:6379"})
 	viper.SetDefault("redis_ts.client_options.username", "")
@@ -38,14 +35,12 @@ func setDefaults() {
 
 	viper.SetDefault("websocket_server.enabled", false)
 	viper.SetDefault("websocket_server.ticker_endpoint", "/tickers")
-	viper.SetDefault("websocket_server.flush_interval", "500ms")
+	viper.SetDefault("websocket_server.flush_interval", "50ms")
 	viper.SetDefault("websocket_server.serialization_protocol", "json")
 
-	viper.SetDefault("questdb.enabled", false)
-	viper.SetDefault("questdb.flush_interval", "10s")
-	viper.SetDefault("questdb.client_options.address", "127.0.0.0.1:9000")
-	viper.SetDefault("questdb.client_options.schema", "http")
-	viper.SetDefault("questdb.individual_feed_table", true)
+	viper.SetDefault("zeromq.enabled", "")
+	viper.SetDefault("zeromq.port", 9998)
+	viper.SetDefault("zeromq.flush_interval", "5ms")
 
 	viper.SetDefault("ticker_transformations", tickertopic.TransformationOptions{})
 }

config/update.go

@@ -1,58 +1,56 @@
 package config
 
 import (
+	"fmt"
 	"log/slog"
 
 	"github.com/spf13/viper"
 )
 
-func UpdateConfig(newConfig ConfigOptions, saveCurrentConfig bool) {
-	// save the configuration object before changes are applied, for backup purposes
-	if saveCurrentConfig {
-		SaveConfig()
-	}
+func UpdateConfig(newConfig ConfigOptions) {
+	fmt.Printf("%+v\n", newConfig)
 
+	// Core
 	viper.Set("env", newConfig.Env)
 	viper.Set("log_level", newConfig.LogLevel)
 	viper.Set("message_buffer_size", newConfig.MessageBufferSize)
 	viper.Set("port", newConfig.Port)
 
-	viper.Set("assets.crypto", newConfig.Assets.Crypto)
-	viper.Set("assets.commodities", newConfig.Assets.Commodities)
-	viper.Set("assets.forex", newConfig.Assets.Forex)
-	viper.Set("assets.stocks", newConfig.Assets.Stocks)
+	// Assets
+	viper.Set("assets", newConfig.Assets)
 
-	viper.Set("stats.enabled", newConfig.Stats.Enabled)
-	viper.Set("stats.interval", newConfig.Stats.Interval)
+	// Stats
+	viper.Set("stats", newConfig.Stats)
 
+	// Datasources
 	viper.Set("datasources", newConfig.Datasources)
 
-	viper.Set("file_output.enabled", newConfig.FileConsumerOptions.Enabled)
-	viper.Set("file_output.filename", newConfig.FileConsumerOptions.OutputFilename)
-
-	viper.Set("mqtt.enabled", newConfig.MQTTConsumerOptions.Enabled)
-	viper.Set("mqtt.qos_level", newConfig.MQTTConsumerOptions.QOSLevel)
+	// File Output
+	viper.Set("file_output", newConfig.FileConsumerOptions)
 
+	// Redis
 	viper.Set("redis_ts.enabled", newConfig.RedisOptions.Enabled)
-	viper.Set("redis_ts.client_options.initaddress", newConfig.RedisOptions.ClientOptions.InitAddress)
-	viper.Set("redis_ts.client_options.username", newConfig.RedisOptions.ClientOptions.Username)
-	viper.Set("redis_ts.client_options.password", newConfig.RedisOptions.ClientOptions.Password)
-	viper.Set("redis_ts.ts.retention", newConfig.RedisOptions.TsOptions.Retention)
-	viper.Set("redis_ts.ts.chunksize", newConfig.RedisOptions.TsOptions.ChunkSize)
-	viper.Set("redis_ts.ts.maxmemory", newConfig.RedisOptions.TsOptions.MaxMemory)
-
-	viper.Set("websocket_server.enabled", newConfig.WebsocketConsumerOptions.Enabled)
-	viper.Set("websocket_server.ticker_endpoint", newConfig.WebsocketConsumerOptions.TickersEndpoint)
-	viper.Set("websocket_server.flush_interval", newConfig.WebsocketConsumerOptions.FlushInterval)
-	viper.Set("websocket_server.serialization_protocol", newConfig.WebsocketConsumerOptions.SerializationProtocol)
+	viper.Set("redis_ts.client_options", newConfig.RedisOptions.ClientOptions)
+	viper.Set("redis_ts.ts", newConfig.RedisOptions.TsOptions)
+
+	// Websocket
+	viper.Set("websocket_server", newConfig.WebsocketConsumerOptions)
+
+	// ZeroMQ
+	viper.Set("zeromq", newConfig.ZMQConsumerOptions)
 
+	// Ticker Transformations
 	viper.Set("ticker_transformations", newConfig.TickerTransformationOptions)
 
-	if saveCurrentConfig {
-		slog.Info("Saving new config")
-		err := viper.WriteConfig()
-		if err != nil {
-			slog.Error("error saving new config to file", "error", err)
-		}
+	slog.Info("Persisting new configuration to disk")
+
+	// Get the actual config file path from viper (defaulting if not set)
+	configFile := viper.ConfigFileUsed()
+	if configFile == "" {
+		configFile = "config.yaml"
+	}
+
+	if err := WriteConfig(configFile); err != nil {
+		slog.Error("CRITICAL: Failed to save config file", "error", err)
 	}
 }

constants/constants.go

@@ -7,11 +7,6 @@ import (
 type AssetName string
 type AssetList []string
 
-var BASES_CRYPTO = AssetList{"FLR", "SGB", "BTC", "XRP", "LTC", "XLM", "DOGE",
-	"ADA", "ALGO", "ETH", "FIL", "ARB", "AVAX", "BNB", "MATIC", "SOL", "USDC", "USDT", "XDC",
-	"TRX", "DOT", "SHIB", "UNI", "HBAR", "NEAR", "VET", "RNDR", "STRK", "AAVE", "QNT",
-	"XTZ", "GALA", "ATOM", "ETC", "BEAM", "IMX", "STX", "APT", "OP", "ICP", "INJ", "TIA", "GRT", "SUI", "LDO"}
-
 var BASES_FOREX = AssetList{
 	"EUR",
 	"JPY",