adding delete to bulk indexer

lib/corebulk.go

@@ -327,6 +327,12 @@ func (b *BulkIndexer) Update(index string, _type string, id, ttl string, date *t
 	return nil
 }
 
+func (b *BulkIndexer) Delete(index, _type, id string, refresh bool) {
+	queryLine := fmt.Sprintf("{\"delete\":{\"_index\":%q,\"_type\":%q,\"_id\":%q,\"refresh\":%t}}\n", index, _type, id, refresh)
+	b.bulkChannel <- []byte(queryLine)
+	return
+}
+
 func (b *BulkIndexer) UpdateWithPartialDoc(index string, _type string, id, ttl string, date *time.Time, partialDoc interface{}, upsert bool, refresh bool) error {
 
 	var data map[string]interface{} = make(map[string]interface{})

lib/corebulk_test.go

@@ -190,6 +190,33 @@ func TestBulkSmallBatch(t *testing.T) {
 
 }
 
+func TestBulkDelete(t *testing.T) {
+	InitTests(true)
+
+	c := NewTestConn()
+	indexer := c.NewBulkIndexer(1)
+	sentBytes := []byte{}
+
+	indexer.Sender = func(buf *bytes.Buffer) error {
+		sentBytes = append(sentBytes, buf.Bytes()...)
+		return nil
+	}
+
+	indexer.Start()
+
+	indexer.Delete("fake", "fake_type", "1", true)
+
+	indexer.Flush()
+	indexer.Stop()
+
+	sent := string(sentBytes)
+
+	expected := `{"delete":{"_index":"fake","_type":"fake_type","_id":"1","refresh":true}}
+`
+	asExpected := sent == expected
+	assert.T(t, asExpected, fmt.Sprintf("Should have sent '%s' but actually sent '%s'", expected, sent))
+}
+
 func XXXTestBulkErrors(t *testing.T) {
 	// lets set a bad port, and hope we get a conn refused error?
 	c := NewTestConn()