Method handleBulkDocs

rest/bulk_api.go:523–644

HTTP handler for a POST to _bulk_docs

()

// HTTP handler for a POST to _bulk_docs
func (h *handler) handleBulkDocs() error {

	if h.db.DatabaseContext.Options.UnsupportedOptions != nil && h.db.DatabaseContext.Options.UnsupportedOptions.RejectWritesWithSkippedSequences {
		// if we are in slow broadcast mode reject write with 503 and increment rejected writes stat
		if h.db.BroadcastSlowMode.Load() {
			h.db.DbStats.DatabaseStats.NumDocWritesRejected.Add(1)
			return base.HTTPErrorf(http.StatusServiceUnavailable, "Database cache is behind and cannot accept writes at this time. Please try again later.")
		}
	}

	startTime := time.Now()
	defer func() {
		h.db.DbStats.CBLReplicationPush().WriteProcessingTime.Add(time.Since(startTime).Nanoseconds())
	}()

	body, err := h.readJSON()
	if err != nil {
		return err
	}

	newEdits, ok := body["new_edits"].(bool)
	if !ok {
		newEdits = true
	}

	userDocs, ok := body["docs"].([]interface{})
	if !ok {
		err = base.HTTPErrorf(http.StatusBadRequest, "missing 'docs' property")
		return err
	}
	lenDocs := len(userDocs)

	// split out local docs, save them on their own
	localDocs := make([]interface{}, 0, lenDocs)
	docs := make([]interface{}, 0, lenDocs)
	for _, item := range userDocs {
		doc, ok := item.(map[string]interface{})
		if !ok {
			err = base.HTTPErrorf(http.StatusBadRequest, "Document body must be JSON")
			return err
		}

		// If ID is present, check whether local doc. (note: if _id is absent or non-string, docid will be
		// empty string and handled during normal doc processing)
		docid, _ := doc[db.BodyId].(string)

		if strings.HasPrefix(docid, db.LocalDocPrefix) {
			localDocs = append(localDocs, doc)
		} else {
			docs = append(docs, doc)
		}
	}

	result := make([]db.Body, 0, len(docs))
	for _, item := range docs {
		doc := item.(map[string]interface{})
		docid, _ := doc[db.BodyId].(string)
		var docErr error