Method openStreamRequest

base/dcp_client.go:500–546

openStreamRequest issues the OpenStream request, but doesn't perform any error handling. Callers should generally use openStream() for error and retry handling

(vbID uint16)

// openStreamRequest issues the OpenStream request, but doesn't perform any error handling.  Callers
// should generally use openStream() for error and retry handling
func (dc *DCPClient) openStreamRequest(vbID uint16) error {

	vbMeta := dc.metadata.GetMeta(vbID)

	options := gocbcore.OpenStreamOptions{}
	// Always use a collection-aware feed if supported
	if dc.supportsCollections {
		options.FilterOptions = &gocbcore.OpenStreamFilterOptions{CollectionIDs: dc.collectionIDs}
	}

	// This has to be buffered so that Cancel() below doesn't lead to blocking in the callback.
	// (If Cancel succeeds then it will lead to directly calling the callback).
	openStreamError := make(chan error, 1)
	openStreamCallback := func(f []gocbcore.FailoverEntry, err error) {
		if err == nil {
			err = dc.verifyFailoverLog(vbID, f)
			if err == nil {
				dc.metadata.SetFailoverEntries(vbID, f)
			}
		}
		openStreamError <- err
	}

	op, openErr := dc.agent.OpenStream(vbID,
		memd.DcpStreamAddFlagActiveOnly,
		vbMeta.VbUUID,
		vbMeta.StartSeqNo,
		vbMeta.EndSeqNo,
		vbMeta.SnapStartSeqNo,
		vbMeta.SnapEndSeqNo,
		dc,
		options,
		openStreamCallback)

	if openErr != nil {
		return openErr
	}

	select {
	case err := <-openStreamError:
		return err
	case <-time.After(openStreamTimeout):
		op.Cancel()
		<-openStreamError
		return ErrTimeout
	}
}

// verifyFailoverLog checks for VbUUID changes when failOnRollback is set, and
// writes the failover log to the client metadata store.  If previous VbUUID is zero, it's