in docappendertest/docappendertest.go [145:194]
func decodeBulkRequest(r *http.Request) (
docs [][]byte,
meta []BulkRequestItemMeta,
result BulkIndexerResponse,
stats RequestStats,
) {
body := r.Body
switch r.Header.Get("Content-Encoding") {
case "gzip":
r, err := gzip.NewReader(body)
if err != nil {
panic(err)
}
defer r.Close()
body = r
}
cr := &countReader{
ReadCloser: body,
}
body = cr
defer cr.Close()
scanner := bufio.NewScanner(body)
for scanner.Scan() {
action := make(map[string]BulkRequestItemMeta)
if err := json.NewDecoder(strings.NewReader(scanner.Text())).Decode(&action); err != nil {
panic(err)
}
var actionType string
for actionType = range action {
}
if !scanner.Scan() {
panic("expected source")
}
doc := append([]byte{}, scanner.Bytes()...)
if !json.Valid(doc) {
panic(fmt.Errorf("invalid JSON: %s", doc))
}
docs = append(docs, doc)
item := BulkIndexerResponseItem{Status: http.StatusCreated, Index: action[actionType].Index}
result.Items = append(result.Items, map[string]BulkIndexerResponseItem{actionType: item})
itemMeta := action[actionType]
itemMeta.Action = actionType
meta = append(meta, itemMeta)
}
return docs, meta, result, RequestStats{int64(cr.bytesRead)}
}