Method Run

rest/database_init_manager.go:248–303

()

}

func (w *DatabaseInitWorker) Run() {
	// Ensure cancelFunc resources are released on normal completion
	defer func() {
		if w.cancelFunc != nil {
			w.cancelFunc(errors.New("database initialization finished normally"))
		}
	}()

	if w.collectionStatusCallback != nil {
		for scName := range w.collections {
			w.collectionStatusCallback(w.dbName, scName, db.CollectionIndexStatusQueued)
		}
	}

	var indexErr error
	for scName, indexSet := range w.collections {
		if w.collectionStatusCallback != nil {
			w.collectionStatusCallback(w.dbName, scName, db.CollectionIndexStatusInProgress)
		}

		// Add the index set to the common indexOptions
		collectionIndexOptions := w.options.indexOptions
		collectionIndexOptions.MetadataIndexes = indexSet

		// Set the scope and collection name on the cluster n1ql store for use by initializeIndexes
		w.n1qlStore.SetScopeAndCollection(scName)
		keyspaceCtx := base.KeyspaceLogCtx(w.ctx, w.n1qlStore.BucketName(), scName.ScopeName(), scName.CollectionName())
		indexErr = db.InitializeIndexes(keyspaceCtx, w.n1qlStore, collectionIndexOptions)
		if w.collectionStatusCallback != nil {
			if indexErr != nil {
				w.collectionStatusCallback(w.dbName, scName, db.CollectionIndexStatusError)
				break
			}
			w.collectionStatusCallback(w.dbName, scName, db.CollectionIndexStatusReady)
		}

		// Check for context cancellation after each collection is processed - if cancelled, return cancellation error
		// to all watchers end exit
		if err := w.ctx.Err(); err != nil {
			indexErr = fmt.Errorf("Database initialization cancelled: %w", err)
			break
		}
	}

	// On completion (success or error), notify watchers
	w.watcherLock.Lock()
	defer w.watcherLock.Unlock()
	w.lastError = indexErr
	for _, doneChan := range w.watchers {
		if indexErr != nil {
			doneChan <- indexErr
		}
		close(doneChan)
	}
	w.completed = true
}

// Adds a watcher for the current worker.  Creates a new notification channel for completion and adds