Repository: DiceDB/dice
File: server/async_tcp.go (Go)

Code before:
		ctx:              ctx,
		cancel:           cancel,
		wg:               wg,
		maxClients:       20000,
		connectedClients: make(map[int]*core.Client),
		store:            core.NewStore(),
		lastCronExecTime: utils.GetCurrentTime(),
		cronFrequency:    1 * time.Second,
	}
}

// SetupUsers initializes the default user for the server
func (s *AsyncServer) SetupUsers() error {
	user, err := core.UserStore.Add(core.DefaultUserName)
	if err != nil {
		return err
	}
	if err := user.SetPassword(config.RequirePass); err != nil {
		return err
	}
	log.Info("default user set up", "password required", config.RequirePass != constants.EmptyStr)
	return nil
}

// WatchKeys watches for changes in keys and notifies clients
func (s *AsyncServer) WatchKeys() {
	defer s.wg.Done()
	for {
		select {
		case event := <-core.WatchChannel:
			s.store.WatchList.Range(func(key, value interface{}) bool {
				query := key.(core.DSQLQuery)
				clients := value.(*sync.Map)

				if core.WildCardMatch(query.KeyRegex, event.Key) {
					queryResult, err := core.ExecuteQuery(query, s.store)
					if err != nil {
						log.Error(err)
						return true
					}

					encodedResult := core.Encode(core.CreatePushResponse(&query, &queryResult), false)
					clients.Range(func(clientKey, _ interface{}) bool {
						clientFd := clientKey.(int)
						_, err := syscall.Write(clientFd, encodedResult)
						if err != nil {
							s.store.RemoveWatcher(query, clientFd)
						}
						return true
					})
				}

Review comment (suggestion):
```suggestion
func (s *AsyncServer) WatchKeys(ctx context.Context) {
```
Context can be passed here. And later we can think do we really want to wait on the context? I mean that case is required or not, for graceful shutdown we can have other ways. 

Code after:
		maxClients:             config.ServerMaxClients,
		connectedClients:       make(map[int]*core.Client),
		store:                  core.NewStore(),
		multiplexerPollTimeout: config.ServerMultiplexerPollTimeout,
		lastCronExecTime:       utils.GetCurrentTime(),
		cronFrequency:          config.ServerCronFrequency,
	}
}

// SetupUsers initializes the default user for the server
func (s *AsyncServer) SetupUsers() error {
	user, err := core.UserStore.Add(core.DefaultUserName)
	if err != nil {
		return err
	}
	if err := user.SetPassword(config.RequirePass); err != nil {
		return err
	}
	log.Info("default user set up", "password required", config.RequirePass != constants.EmptyStr)
	return nil
}

// WatchKeys watches for changes in keys and notifies clients
func (s *AsyncServer) WatchKeys(ctx context.Context) {
	for {
		select {
		case event := <-core.WatchChannel:
			s.store.WatchList.Range(func(key, value interface{}) bool {
				query := key.(core.DSQLQuery)
				clients := value.(*sync.Map)

				if core.WildCardMatch(query.KeyRegex, event.Key) {
					queryResult, err := core.ExecuteQuery(query, s.store)
					if err != nil {
						log.Error(err)
						return true
					}

					encodedResult := core.Encode(core.CreatePushResponse(&query, &queryResult), false)
					clients.Range(func(clientKey, _ interface{}) bool {
						clientFd := clientKey.(int)
						_, err := syscall.Write(clientFd, encodedResult)
						if err != nil {
							s.store.RemoveWatcher(query, clientFd)
						}
						return true
					})
				}
				return true
			})
		case <-ctx.Done():