feat(nats): add compression config (#432)

In NATS client and server, if the connection uses websocket we can enable compression by passing the Compression config - it needs to be enbaled on both client and server conn setup. Thus creating a new config to enbale on both sides - if not websocket, this config has no effect
topfreegames · Jan 14, 2025 · 3eb2a53 · 3eb2a53
1 parent c37f8a0
commit 3eb2a53
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 4 deletions.
diff --git a/cluster/nats_rpc_client.go b/cluster/nats_rpc_client.go
@@ -53,6 +53,7 @@ type NatsRPCClient struct {
 	server                 *Server
 	metricsReporters       []metrics.Reporter
 	appDieChan             chan bool
+	websocketCompression   bool
 }
 
 // NewNatsRPCClient ctor
@@ -86,6 +87,7 @@ func (ns *NatsRPCClient) configure(config config.NatsRPCClientConfig) error {
 	if ns.reqTimeout == 0 {
 		return constants.ErrNatsNoRequestTimeout
 	}
+	ns.websocketCompression = config.WebsocketCompression
 	return nil
 }
 
@@ -240,6 +242,7 @@ func (ns *NatsRPCClient) Init() error {
 		ns.appDieChan,
 		nats.MaxReconnects(ns.maxReconnectionRetries),
 		nats.Timeout(ns.connectionTimeout),
+		nats.Compression(ns.websocketCompression),
 	)
 	if err != nil {
 		return err

diff --git a/cluster/nats_rpc_server.go b/cluster/nats_rpc_server.go
@@ -63,6 +63,7 @@ type NatsRPCServer struct {
 	metricsReporters       []metrics.Reporter
 	sessionPool            session.SessionPool
 	appDieChan             chan bool
+	websocketCompression   bool
 }
 
 // NewNatsRPCServer ctor
@@ -114,6 +115,7 @@ func (ns *NatsRPCServer) configure(config config.NatsRPCServerConfig) error {
 	ns.userKickCh = make(chan *protos.KickMsg, ns.messagesBufferSize)
 	ns.responses = make([]*protos.Response, ns.service)
 	ns.requests = make([]*protos.Request, ns.service)
+	ns.websocketCompression = config.WebsocketCompression
 	return nil
 }
 
@@ -332,6 +334,7 @@ func (ns *NatsRPCServer) Init() error {
 		ns.appDieChan,
 		nats.MaxReconnects(ns.maxReconnectionRetries),
 		nats.Timeout(ns.connectionTimeout),
+		nats.Compression(ns.websocketCompression),
 	)
 	if err != nil {
 		return err

diff --git a/config/config.go b/config/config.go
@@ -202,6 +202,7 @@ type NatsRPCClientConfig struct {
 	MaxReconnectionRetries int           `mapstructure:"maxreconnectionretries"`
 	RequestTimeout         time.Duration `mapstructure:"requesttimeout"`
 	ConnectionTimeout      time.Duration `mapstructure:"connectiontimeout"`
+	WebsocketCompression   bool          `mapstructure:"websocketcompression"`
 }
 
 // newDefaultNatsRPCClientConfig provides default nats client configuration
@@ -211,6 +212,7 @@ func newDefaultNatsRPCClientConfig() *NatsRPCClientConfig {
 		MaxReconnectionRetries: 15,
 		RequestTimeout:         time.Duration(5 * time.Second),
 		ConnectionTimeout:      time.Duration(2 * time.Second),
+		WebsocketCompression:   true,
 	}
 }
 
@@ -222,8 +224,9 @@ type NatsRPCServerConfig struct {
 		Messages int `mapstructure:"messages"`
 		Push     int `mapstructure:"push"`
 	} `mapstructure:"buffer"`
-	Services          int           `mapstructure:"services"`
-	ConnectionTimeout time.Duration `mapstructure:"connectiontimeout"`
+	Services             int           `mapstructure:"services"`
+	ConnectionTimeout    time.Duration `mapstructure:"connectiontimeout"`
+	WebsocketCompression bool          `mapstructure:"websocketcompression"`
 }
 
 // newDefaultNatsRPCServerConfig provides default nats server configuration
@@ -238,8 +241,9 @@ func newDefaultNatsRPCServerConfig() *NatsRPCServerConfig {
 			Messages: 75,
 			Push:     100,
 		},
-		Services:          30,
-		ConnectionTimeout: time.Duration(2 * time.Second),
+		Services:             30,
+		ConnectionTimeout:    time.Duration(2 * time.Second),
+		WebsocketCompression: true,
 	}
 }
 

diff --git a/config/viper_config.go b/config/viper_config.go
@@ -66,11 +66,13 @@ func (c *Config) fillDefaultValues() {
 		"pitaya.cluster.rpc.client.nats.connect":                pitayaConfig.Cluster.RPC.Client.Nats.Connect,
 		"pitaya.cluster.rpc.client.nats.connectiontimeout":      pitayaConfig.Cluster.RPC.Client.Nats.ConnectionTimeout,
 		"pitaya.cluster.rpc.client.nats.maxreconnectionretries": pitayaConfig.Cluster.RPC.Client.Nats.MaxReconnectionRetries,
+		"pitaya.cluster.rpc.client.nats.websocketcompression":   pitayaConfig.Cluster.RPC.Client.Nats.WebsocketCompression,
 		"pitaya.cluster.rpc.client.nats.requesttimeout":         pitayaConfig.Cluster.RPC.Client.Nats.RequestTimeout,
 		"pitaya.cluster.rpc.server.grpc.port":                   pitayaConfig.Cluster.RPC.Server.Grpc.Port,
 		"pitaya.cluster.rpc.server.nats.connect":                pitayaConfig.Cluster.RPC.Server.Nats.Connect,
 		"pitaya.cluster.rpc.server.nats.connectiontimeout":      pitayaConfig.Cluster.RPC.Server.Nats.ConnectionTimeout,
 		"pitaya.cluster.rpc.server.nats.maxreconnectionretries": pitayaConfig.Cluster.RPC.Server.Nats.MaxReconnectionRetries,
+		"pitaya.cluster.rpc.server.nats.websocketcompression":   pitayaConfig.Cluster.RPC.Server.Nats.WebsocketCompression,
 		"pitaya.cluster.rpc.server.nats.services":               pitayaConfig.Cluster.RPC.Server.Nats.Services,
 		"pitaya.cluster.rpc.server.nats.buffer.messages":        pitayaConfig.Cluster.RPC.Server.Nats.Buffer.Messages,
 		"pitaya.cluster.rpc.server.nats.buffer.push":            pitayaConfig.Cluster.RPC.Server.Nats.Buffer.Push,

diff --git a/docs/configuration.rst b/docs/configuration.rst
@@ -130,6 +130,10 @@ The configurations only need to be set if the RPC Service is enabled with the gi
     - 15
     - int
     - Maximum number of retries to reconnect to nats for the client
+  * - pitaya.cluster.rpc.client.nats.websocketcompression
+    - true
+    - bool
+    - Enables compression in websocket connections to NATS. Needs both client and server to be enabled
   * - pitaya.cluster.rpc.server.nats.connect
     - nats://localhost:4222
     - string
@@ -142,6 +146,10 @@ The configurations only need to be set if the RPC Service is enabled with the gi
     - 15
     - int
     - Maximum number of retries to reconnect to nats for the server
+  * - pitaya.cluster.rpc.server.nats.websocketcompression
+    - true
+    - bool
+    - Enables compression in websocket connections to NATS. Needs both client and server to be enabled
   * - pitaya.cluster.rpc.server.grpc.port
     - 3434
     - int