remove dispatcherPoolSize

Author: antonbashir

.vscode/launch.json

@@ -33,7 +33,6 @@
         "MERCURE_EXTRA_DIRECTIVES": "anonymous\nwrite_timeout 10s",
         "MERCURE_REDIS_ADDRESS": "localhost:6379",
         "MERCURE_REDIS_SUBSCRIBERS_SIZE": "10000",
-        "MERCURE_REDIS_DISPATCHER_POOL_SIZE": "16",
         "MERCURE_REDIS_CHANNEL": "mercure",
         "GLOBAL_OPTIONS": "debug",
         "SERVER_NAME": ":1234",

caddy/redis.go

@@ -15,12 +15,11 @@ func init() {
 }
 
 type Redis struct {
-	Address            string `json:"address,omitempty"`
-	Username           string `json:"username,omitempty"`
-	Password           string `json:"password,omitempty"`
-	SubscribersSize    int    `json:"subscribers_size,omitempty"`
-	DispatcherPoolSize int    `json:"dispatcher_pool_size,omitempty"`
-	RedisChannel       string `json:"redis_channel,omitempty"`
+	Address         string `json:"address,omitempty"`
+	Username        string `json:"username,omitempty"`
+	Password        string `json:"password,omitempty"`
+	SubscribersSize int    `json:"subscribers_size,omitempty"`
+	RedisChannel    string `json:"redis_channel,omitempty"`
 
 	transport    *mercure.RedisTransport
 	transportKey string
@@ -47,7 +46,7 @@ func (r *Redis) Provision(ctx caddy.Context) error {
 	r.transportKey = key.String()
 
 	destructor, _, err := TransportUsagePool.LoadOrNew(r.transportKey, func() (caddy.Destructor, error) {
-		t, err := mercure.NewRedisTransport(ctx.Logger(), r.Address, r.Username, r.Password, r.SubscribersSize, r.DispatcherPoolSize, r.RedisChannel)
+		t, err := mercure.NewRedisTransport(ctx.Logger(), r.Address, r.Username, r.Password, r.SubscribersSize, r.RedisChannel)
 		if err != nil {
 			return nil, err
 		}
@@ -108,18 +107,6 @@ func (r *Redis) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
 
 				r.SubscribersSize = s
 
-			case "dispatcher_pool_size":
-				if !d.NextArg() {
-					return d.ArgErr()
-				}
-
-				s, e := strconv.Atoi(replacer.ReplaceKnown(d.Val(), ""))
-				if e != nil {
-					return e
-				}
-
-				r.DispatcherPoolSize = s
-
 			case "redis_channel":
 				if !d.NextArg() {
 					return d.ArgErr()

charts/mercure/templates/deployment.yaml

@@ -56,10 +56,6 @@ spec:
             - name: MERCURE_REDIS_SUBSCRIBERS_SIZE
               value: "{{ .Values.redis.subscribers_size }}"
             {{- end }}
-            {{- if .Values.redis.dispatcher_pool_size }}
-            - name: MERCURE_REDIS_DISPATCHER_POOL_SIZE
-              value: "{{ .Values.redis.dispatcher_pool_size }}"
-            {{- end }}
             {{- if .Values.redis.redis_channel }}
             - name: MERCURE_REDIS_CHANNEL
               value: "{{ .Values.redis.redis_channel }}"

charts/mercure/values.yaml

@@ -223,7 +223,6 @@ redis:
   password: ~
   dispatch_timer: ~
   subscribers_size: ~
-  dispatcher_pool_size: ~
   redis_channel: mercure
 
 customLabels: ~

redis.Caddyfile

@@ -31,7 +31,6 @@
       username {env.MERCURE_REDIS_USERNAME}
       password {env.MERCURE_REDIS_PASSWORD}
       subscribers_size {env.MERCURE_REDIS_SUBSCRIBERS_SIZE}
-      dispatcher_pool_size {env.MERCURE_REDIS_DISPATCHER_POOL_SIZE}
       redis_channel {env.MERCURE_REDIS_CHANNEL}
     }
 	}

redis.dev.Caddyfile

@@ -37,7 +37,6 @@
       username {env.MERCURE_REDIS_USERNAME}
       password {env.MERCURE_REDIS_PASSWORD}
       subscribers_size {env.MERCURE_REDIS_SUBSCRIBERS_SIZE}
-      dispatcher_pool_size {env.MERCURE_REDIS_DISPATCHER_POOL_SIZE}
       redis_channel {env.MERCURE_REDIS_CHANNEL}
     }
 	}

redis.go

@@ -24,7 +24,6 @@ type RedisTransport struct {
 	logger             Logger
 	client             *redis.Client
 	subscribers        *SubscriberList
-	dispatcherPoolSize int
 	closed             chan any
 	publishScript      *redis.Script
 	closedOnce         sync.Once
@@ -37,7 +36,6 @@ func NewRedisTransport(
 	username string,
 	password string,
 	subscribersSize int,
-	dispatcherPoolSize int,
 	redisChannel string,
 ) (*RedisTransport, error) {
 	client := redis.NewClient(&redis.Options{
@@ -50,14 +48,13 @@ func NewRedisTransport(
 		return nil, fmt.Errorf("failed to connect to Redis: %w", pong.Err())
 	}
 
-	return NewRedisTransportInstance(logger, client, subscribersSize, dispatcherPoolSize, redisChannel)
+	return NewRedisTransportInstance(logger, client, subscribersSize, redisChannel)
 }
 
 func NewRedisTransportInstance(
 	logger Logger,
 	client *redis.Client,
 	subscribersSize int,
-	dispatcherPoolSize int,
 	redisChannel string,
 ) (*RedisTransport, error) {
 	subscriber := client.PSubscribe(context.Background(), redisChannel)
@@ -68,7 +65,6 @@ func NewRedisTransportInstance(
 		logger:             logger,
 		client:             client,
 		subscribers:        NewSubscriberList(subscribersSize),
-		dispatcherPoolSize: dispatcherPoolSize,
 		publishScript:      redis.NewScript(publishScript),
 		closed:             make(chan any),
 		redisChannel:       redisChannel,