// Helper function to open a Couchbase connection and return a specific bucket.
func ConnectToBucket(spec base.BucketSpec) (bucket base.Bucket, err error) {
	bucket, err = base.GetBucket(spec)
	if err != nil {
		err = base.HTTPErrorf(http.StatusBadGateway,
			"Unable to connect to server: %s", err)
	} else {
		err = installViews(bucket)
	}
	return
}

// Helper function to open a Couchbase connection and return a specific bucket.
func ConnectToBucket(spec base.BucketSpec, callback func(bucket string, err error)) (bucket base.Bucket, err error) {
	bucket, err = base.GetBucket(spec, callback)
	if err != nil {
		err = base.HTTPErrorf(http.StatusBadGateway,
			" Unable to connect to Couchbase Server (connection refused). Please ensure it is running and reachable at the configured host and port.  Detailed error: %s", err)
	} else {
		err = installViews(bucket)
	}
	return
}

func (k *kvChangeIndexWriter) Init(context *DatabaseContext, options *CacheOptions, indexOptions *ChangeIndexOptions, indexPartitionsCallback IndexPartitionsFunc) (err error) {

	k.context = context
	k.pending = make(chan *LogEntry, maxCacheUpdate)

	k.indexPartitionsCallback = indexPartitionsCallback

	// start process to work pending sequences
	go func() {
		err := k.indexPending()
		if err != nil {
			base.LogFatal("Indexer failed with unrecoverable error:%v", err)
		}

	}()

	k.channelIndexWriters = make(map[string]*kvChannelIndex)
	k.indexWriteBucket, err = base.GetBucket(indexOptions.Spec, nil)
	if err != nil {
		base.Logf("Error opening index bucket %q, pool %q, server <%s>",
			indexOptions.Spec.BucketName, indexOptions.Spec.PoolName, indexOptions.Spec.Server)
		// TODO: revert to local index?
		return err
	}
	cbBucket, ok := k.indexWriteBucket.(base.CouchbaseBucket)
	var maxVbNo uint16
	if ok {
		maxVbNo, _ = cbBucket.GetMaxVbno()
	} else {
		// walrus, for unit testing
		maxVbNo = 1024
	}

	// Set of worker goroutines used to process incoming entries
	k.unmarshalWorkQueue = make(chan *unmarshalEntry, 500)

	// Start fixed set of goroutines to work the unmarshal work queue
	for i := 0; i < maxUnmarshalProcesses; i++ {
		go func() {
			for {
				select {
				case unmarshalEntry := <-k.unmarshalWorkQueue:
					unmarshalEntry.process()
				case <-k.terminator:
					return
				}
			}
		}()
	}

	// Initialize unmarshalWorkers
	k.unmarshalWorkers = make([]*unmarshalWorker, maxVbNo)

	return nil
}

func (k *kvChangeIndexReader) Init(options *CacheOptions, indexOptions *ChangeIndexOptions, onChange func(base.Set), indexPartitionsCallback IndexPartitionsFunc) (err error) {

	k.channelIndexReaders = make(map[string]*kvChannelIndex)
	k.indexPartitionsCallback = indexPartitionsCallback

	// Initialize notification Callback
	k.onChange = onChange

	k.indexReadBucket, err = base.GetBucket(indexOptions.Spec, nil)
	if err != nil {
		base.Logf("Error opening index bucket %q, pool %q, server <%s>",
			indexOptions.Spec.BucketName, indexOptions.Spec.PoolName, indexOptions.Spec.Server)
		// TODO: revert to local index?
		return err
	}

	cbBucket, ok := k.indexReadBucket.(base.CouchbaseBucket)
	if ok {
		k.maxVbNo, _ = cbBucket.GetMaxVbno()
	} else {
		// walrus, for unit testing
		k.maxVbNo = 1024
	}

	// Start background task to poll for changes
	k.terminator = make(chan struct{})
	k.pollingActive = make(chan struct{})
	go func(k *kvChangeIndexReader) {
		defer close(k.pollingActive)
		pollStart := time.Now()
		for {
			timeSinceLastPoll := time.Since(pollStart)
			waitTime := (kPollFrequency * time.Millisecond) - timeSinceLastPoll
			if waitTime < 0 {
				waitTime = 0 * time.Millisecond
			}
			select {
			case <-k.terminator:
				return
			case <-time.After(waitTime):
				// TODO: Doesn't trigger the reader removal processing (in pollReaders) during long
				//       periods without changes to stableSequence.  In that scenario we'll continue
				//       stable sequence polling each poll interval, even if we *actually* don't have any
				//       active readers.
				pollStart = time.Now()
				if k.hasActiveReaders() && k.stableSequenceChanged() {
					k.pollReaders()
				}
			}

		}
	}(k)

	return nil
}

func init() {
	var err error
	gTestBucket, err = base.GetBucket(base.BucketSpec{
		Server:     kTestURL,
		BucketName: "sync_gateway_tests"}, nil)
	if err != nil {
		log.Fatalf("Couldn't connect to bucket: %v", err)
	}
	if err != nil {
		log.Fatalf("Couldn't install design doc: %v", err)
	}
}

func (sc *ServerContext) startShadowing(dbcontext *db.DatabaseContext, shadow *ShadowConfig) error {

	base.Warn("Bucket Shadowing feature comes with a number of limitations and caveats. See https://github.com/couchbase/sync_gateway/issues/1363 for more details.")

	var pattern *regexp.Regexp
	if shadow.Doc_id_regex != nil {
		var err error
		pattern, err = regexp.Compile(*shadow.Doc_id_regex)
		if err != nil {
			base.Warn("Invalid shadow doc_id_regex: %s", *shadow.Doc_id_regex)
			return err
		}
	}

	spec := base.BucketSpec{
		Server:     *shadow.Server,
		PoolName:   "default",
		BucketName: *shadow.Bucket,
		FeedType:   shadow.FeedType,
	}
	if shadow.Pool != nil {
		spec.PoolName = *shadow.Pool
	}
	if shadow.Username != "" {
		spec.Auth = shadow
	}

	bucket, err := base.GetBucket(spec, nil)

	if err != nil {
		err = base.HTTPErrorf(http.StatusBadGateway,
			"Unable to connect to shadow bucket: %s", err)
		return err
	}
	shadower, err := db.NewShadower(dbcontext, bucket, pattern)
	if err != nil {
		bucket.Close()
		return err
	}
	dbcontext.Shadower = shadower

	//Remove credentials from server URL before logging
	url, err := couchbase.ParseURL(spec.Server)
	if err == nil {
		base.Logf("Database %q shadowing remote bucket %q, pool %q, server <%s:%s/%s>", dbcontext.Name, spec.BucketName, spec.PoolName, url.Scheme, url.Host, url.Path)
	}
	return nil
}

// Helper function to open a Couchbase connection and return a specific bucket.
func ConnectToBucket(spec base.BucketSpec, callback func(bucket string, err error)) (bucket base.Bucket, err error) {
	//start a retry loop to connect to the bucket backing off double the delay each time
	worker := func() (shouldRetry bool, err error, value interface{}) {
		bucket, err = base.GetBucket(spec, callback)
		return err != nil, err, bucket
	}

	sleeper := base.CreateDoublingSleeperFunc(
		13, //MaxNumRetries approx 40 seconds total retry duration
		5,  //InitialRetrySleepTimeMS
	)

	description := fmt.Sprintf("Attempt to connect to bucket : %v", spec.BucketName)
	err, ibucket := base.RetryLoop(description, worker, sleeper)

	if err != nil {
		err = base.HTTPErrorf(http.StatusBadGateway,
			" Unable to connect to Couchbase Server (connection refused). Please ensure it is running and reachable at the configured host and port.  Detailed error: %s", err)
	} else {
		bucket, _ := ibucket.(base.Bucket)
		err = installViews(bucket)
	}
	return
}

//.........这里部分代码省略.........
		indexBucketName := ""

		if config.ChannelIndex.Server != nil {
			indexServer = *config.ChannelIndex.Server
		}
		if config.ChannelIndex.Pool != nil {
			indexPool = *config.ChannelIndex.Pool
		}
		if config.ChannelIndex.Bucket != nil {
			indexBucketName = *config.ChannelIndex.Bucket
		}
		indexSpec := base.BucketSpec{
			Server:          indexServer,
			PoolName:        indexPool,
			BucketName:      indexBucketName,
			CouchbaseDriver: base.GoCB,
		}
		if config.ChannelIndex.Username != "" {
			indexSpec.Auth = config.ChannelIndex
		}

		if config.ChannelIndex.NumShards != 0 {
			channelIndexOptions.NumShards = config.ChannelIndex.NumShards
		} else {
			channelIndexOptions.NumShards = 64
		}

		channelIndexOptions.ValidateOrPanic()

		channelIndexOptions.Spec = indexSpec
		channelIndexOptions.Writer = config.ChannelIndex.IndexWriter

		// TODO: separate config of hash bucket
		sequenceHashOptions.Bucket, err = base.GetBucket(indexSpec, nil)
		if err != nil {
			base.Logf("Error opening sequence hash bucket %q, pool %q, server <%s>",
				indexBucketName, indexPool, indexServer)
			// TODO: revert to local index?
			return nil, err
		}
		sequenceHashOptions.Size = 32
	} else {
		channelIndexOptions = nil
	}

	var revCacheSize uint32
	if config.RevCacheSize != nil && *config.RevCacheSize > 0 {
		revCacheSize = *config.RevCacheSize
	} else {
		revCacheSize = db.KDefaultRevisionCacheCapacity
	}

	contextOptions := db.DatabaseContextOptions{
		CacheOptions:          &cacheOptions,
		IndexOptions:          channelIndexOptions,
		SequenceHashOptions:   sequenceHashOptions,
		RevisionCacheCapacity: revCacheSize,
		AdminInterface:        sc.config.AdminInterface,
	}

	dbcontext, err := db.NewDatabaseContext(dbName, bucket, autoImport, contextOptions)
	if err != nil {
		return nil, err
	}
	dbcontext.BucketSpec = spec