code: func (p *Processor) initialize() (err error) {
  	// If we're already initialized, do nothing.
  	if p.annotationStream != nil {
  		return nil
  	}
  
  	annotationPath := p.o.AnnotationSubpath
  	if annotationPath == "" {
  		annotationPath = DefaultAnnotationSubpath
  	}
  	annotationPath = p.astate.RootStep().BaseStream(annotationPath)
  
  	// Create our annotation stream.
  	if p.annotationStream, err = p.createStream(annotationPath, &metadataStreamArchetype); err != nil {
  		log.WithError(err).Errorf(p.ctx, "Failed to create annotation stream.")
  		return
  	}
  
  	// Complete initialization and start our annotation meter.
  	p.annotationC = make(chan annotationSignal)
  	p.annotationFinishedC = make(chan struct{})
  	p.allEmittedStreams = map[*Stream]struct{}{}
  
  	// Run our annotation meter in a separate goroutine.
  	go p.runAnnotationMeter(p.annotationStream, p.o.MetadataUpdateInterval)
  	return nil
  }
docs: // initialize initializes p's annotation stream handling system. If it is called
  // more than once, it is a no-op.