Function TestBulkIndexerFlush

esutil/bulk_indexer_internal_test.go:1922–2281

nolint:gocyclo

(t *testing.T)

//nolint:gocyclo
func TestBulkIndexerFlush(t *testing.T) {
	t.Run("Basic", func(t *testing.T) {
		responseBody := `{"took":1,"errors":false,"items":[{"index":{"_index":"test","_id":"1","_version":1,"result":"created","status":201}}]}`

		es, err := elasticsearch.NewClient(elasticsearch.Config{Transport: &mockTransport{
			RoundTripFunc: func(_ *http.Request) (*http.Response, error) {
				return &http.Response{
					StatusCode: 200,
					Body:       io.NopCloser(strings.NewReader(responseBody)),
					Header:     http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
				}, nil
			},
		}})
		if err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}

		bi, err := NewBulkIndexer(BulkIndexerConfig{
			NumWorkers:    1,
			FlushInterval: time.Hour,
			Client:        es,
		})
		if err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}

		if err := bi.Add(context.Background(), BulkIndexerItem{
			Action:     "index",
			DocumentID: "1",
			Body:       strings.NewReader(`{"title":"foo"}`),
		}); err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}

		if err := bi.Flush(context.Background()); err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}

		stats := bi.Stats()
		if stats.NumFlushed != 1 {
			t.Errorf("Unexpected NumFlushed after Flush: want=1, got=%d", stats.NumFlushed)
		}

		// Indexer is still usable after Flush
		if err := bi.Add(context.Background(), BulkIndexerItem{
			Action:     "index",
			DocumentID: "2",
			Body:       strings.NewReader(`{"title":"bar"}`),
		}); err != nil {
			t.Fatalf("Unexpected error adding after Flush: %s", err)
		}

		if err := bi.Close(context.Background()); err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}

		stats = bi.Stats()
		if stats.NumFlushed != 2 {