Method Watch

db.go:1195–1302

** * Watch watches the key and bucket and calls the callback function for each message received. * The callback will be called once for each individual message in the batch. * * @param ctx - the context for the watch, we provide the context as the way of cancel manually watcher * @param bucket

(ctx context.Context, bucket string, key []byte, cb func(message *Message) error, opts ...WatchOptions)

 * @return error - the error if the watch is stopped
 */
func (db *DB) Watch(ctx context.Context, bucket string, key []byte, cb func(message *Message) error, opts ...WatchOptions) (*Watcher, error) {
	watchOpts := NewWatchOptions()

	if len(opts) > 0 {
		watchOpts = &opts[0]
	}

	if db.watchMgr == nil {
		return nil, ErrWatchFeatureDisabled
	}

	subscriber, err := db.watchMgr.subscribe(bucket, string(key))
	if err != nil {
		return nil, err
	}

	watchingFunc := func() error {
		maxBatchSize := 128
		batch := make([]*Message, 0, maxBatchSize)

		// Use a ticker to process the batch every 100 milliseconds
		// Avoid CPU busy spinning
		ticker := time.NewTicker(100 * time.Millisecond)
		keyWatch := string(key)

		processBatch := func(batch []*Message) error {
			if len(batch) == 0 {
				return nil
			}

			for _, msg := range batch {
				errChan := make(chan error, 1)
				go func(msg *Message) {
					errChan <- cb(msg)
				}(msg)

				select {
				case <-time.After(watchOpts.CallbackTimeout):
					return ErrWatchingCallbackTimeout
				case err := <-errChan:
					if err != nil {
						return err
					}
				}
			}
			return nil
		}

		defer func() {
			ticker.Stop()

			if subscriber != nil && subscriber.active.Load() {
				if err := db.watchMgr.unsubscribe(bucket, keyWatch, subscriber.id); err != nil {
					// ignore the error
				}
			}
		}()