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

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
 }
 
@@ -233,6 +235,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

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
 }
 
@@ -328,6 +330,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

config/config.go

@@ -256,6 +256,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
@@ -265,6 +266,7 @@ func NewDefaultNatsRPCClientConfig() *NatsRPCClientConfig {
 		MaxReconnectionRetries: 15,
 		RequestTimeout:         time.Duration(5 * time.Second),
 		ConnectionTimeout:      time.Duration(2 * time.Second),
+		WebsocketCompression:   true,
 	}
 }
 
@@ -285,8 +287,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
@@ -301,8 +304,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,
 	}
 }
 

config/viper_config.go

@@ -21,11 +21,12 @@
 package config
 
 import (
-	"github.com/mitchellh/mapstructure"
 	"reflect"
 	"strings"
 	"time"
 
+	"github.com/mitchellh/mapstructure"
+
 	"github.com/spf13/viper"
 )
 
@@ -81,11 +82,13 @@ func (c *Config) fillDefaultValues() {
 		"pitaya.cluster.rpc.client.nats.connect":                natsRPCClientConfig.Connect,
 		"pitaya.cluster.rpc.client.nats.connectiontimeout":      natsRPCClientConfig.ConnectionTimeout,
 		"pitaya.cluster.rpc.client.nats.maxreconnectionretries": natsRPCClientConfig.MaxReconnectionRetries,
+		"pitaya.cluster.rpc.client.nats.websocketcompression":   natsRPCClientConfig.WebsocketCompression,
 		"pitaya.cluster.rpc.client.nats.requesttimeout":         natsRPCClientConfig.RequestTimeout,
 		"pitaya.cluster.rpc.server.grpc.port":                   grpcRPCServerConfig.Port,
 		"pitaya.cluster.rpc.server.nats.connect":                natsRPCServerConfig.Connect,
 		"pitaya.cluster.rpc.server.nats.connectiontimeout":      natsRPCServerConfig.ConnectionTimeout,
 		"pitaya.cluster.rpc.server.nats.maxreconnectionretries": natsRPCServerConfig.MaxReconnectionRetries,
+		"pitaya.cluster.rpc.server.nats.websocketcompression":   natsRPCServerConfig.WebsocketCompression,
 		"pitaya.cluster.rpc.server.nats.services":               natsRPCServerConfig.Services,
 		"pitaya.cluster.rpc.server.nats.buffer.messages":        natsRPCServerConfig.Buffer.Messages,
 		"pitaya.cluster.rpc.server.nats.buffer.push":            natsRPCServerConfig.Buffer.Push,

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