Method Watch

jetstream/object.go:1278–1349

Watch for changes in the underlying store and receive meta information updates.

(ctx context.Context, opts ...WatchOpt)

// Watch for changes in the underlying store and receive meta information updates.
func (obs *obs) Watch(ctx context.Context, opts ...WatchOpt) (ObjectWatcher, error) {
	var o watchOpts
	for _, opt := range opts {
		if opt != nil {
			if err := opt.configureWatcher(&o); err != nil {
				return nil, err
			}
		}
	}

	var initDoneMarker bool

	w := &objWatcher{updates: make(chan *ObjectInfo, 32)}

	update := func(m *nats.Msg) {
		var info ObjectInfo
		if err := json.Unmarshal(m.Data, &info); err != nil {
			return // TODO(dlc) - Communicate this upwards?
		}
		meta, err := m.Metadata()
		if err != nil {
			return
		}

		if !o.ignoreDeletes || !info.Deleted {
			info.ModTime = meta.Timestamp
			w.updates <- &info
		}

		// if UpdatesOnly is set, no not send nil to the channel
		// as it would always be triggered after initializing the watcher
		if !initDoneMarker && meta.NumPending == 0 {
			initDoneMarker = true
			w.updates <- nil
		}
	}

	allMeta := fmt.Sprintf(objAllMetaPreTmpl, obs.name)
	_, err := obs.stream.GetLastMsgForSubject(ctx, allMeta)
	// if there are no messages on the stream and we are not watching
	// updates only, send nil to the channel to indicate that the initial
	// watch is done
	if !o.updatesOnly {
		if errors.Is(err, ErrMsgNotFound) {
			initDoneMarker = true
			w.updates <- nil
		}
	} else {
		// if UpdatesOnly was used, mark initialization as complete
		initDoneMarker = true
	}

	// Used ordered consumer to deliver results.
	streamName := fmt.Sprintf(objNameTmpl, obs.name)
	subOpts := []nats.SubOpt{nats.OrderedConsumer(), nats.BindStream(streamName)}
	if !o.includeHistory {
		subOpts = append(subOpts, nats.DeliverLastPerSubject())
	}