Method onStreamEnd

base/dcp_client.go:583–617

(e endStreamEvent)

}

func (dc *DCPClient) onStreamEnd(e endStreamEvent) {
	if e.err == nil {
		DebugfCtx(dc.ctx, KeyDCP, "Stream (vb:%d) closed, all items streamed", e.vbID)
		dc.deactivateVbucket(e.vbID)
		return
	}

	if errors.Is(e.err, gocbcore.ErrDCPStreamClosed) {
		DebugfCtx(dc.ctx, KeyDCP, "Stream (vb:%d) closed by DCPClient", e.vbID)
		dc.fatalError(fmt.Errorf("Stream (vb:%d) closed by DCPClient", e.vbID))
		return
	}

	if errors.Is(e.err, gocbcore.ErrDCPStreamStateChanged) || errors.Is(e.err, gocbcore.ErrDCPStreamTooSlow) || errors.Is(e.err, gocbcore.ErrDCPStreamDisconnected) {
		DebugfCtx(dc.ctx, KeyDCP, "Stream (vb:%d) ended with a known error, will reconnect. Reason: %s", e.vbID, e.err)
	} else {
		InfofCtx(dc.ctx, KeyDCP, "Stream (vb:%d) ended with an unknown error, will reconnect. Reason: %s", e.vbID, e.err)
	}
	retries := infiniteOpenStreamRetries
	if dc.oneShot {
		retries = openRetryCount
	}

	// Re-opening the stream needs to be asynchronous, due to the way the DCPAgent performs locking while
	// reconfiguring memdclients - the old client can't be closed while it has pending StreamObserver.End calls,
	// and our openStream request won't succeed until the old client is closed.
	// Since we've got a relatively small event buffer for processing observer events (10 x 8 workers), a
	// synchronous openStream request will create a deadlock whenever more than 80 vbuckets need to be closed.
	go func(vb uint16, maxRetries uint32) {
		err := dc.openStream(vb, maxRetries)
		if err != nil {
			dc.fatalError(fmt.Errorf("Stream (vb:%d) failed to reopen: %w", vb, err))
		}
	}(e.vbID, retries)
}

func (dc *DCPClient) fatalError(err error) {
	dc.setCloseError(err)