From ab7bfaf7432de2bc7112af98981a7a7124e3bf85 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Wed, 16 Sep 2015 12:04:24 -0700
Subject: [PATCH 01/12] Whitespace

---
 lib/elastomer/client/bulk.rb | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb
index c73a6e83..ab85c999 100644
--- a/lib/elastomer/client/bulk.rb
+++ b/lib/elastomer/client/bulk.rb
@@ -156,7 +156,7 @@ def create( document, params )
       end
 
       # Add an update action to the list of bulk actions to be performed when
-      # the bulk API call is made. Parameters can be provided in the parameters 
+      # the bulk API call is made. Parameters can be provided in the parameters
       # hash (underscore prefix optional) or in the document hash (underscore
       # prefix required).
       #
@@ -178,7 +178,7 @@ def update( document, params )
       # the bulk API call is made.
       #
       # params - Parameters for the delete action (as a Hash)
-      #      
+      #
       # Examples
       #   delete(:_id => 1, :_type => 'foo')
       #

From d8c7284efb29251c1fb58db14543e61acf5fe34d Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Wed, 16 Sep 2015 12:34:54 -0700
Subject: [PATCH 02/12] Add streaming bulk functionality

---
 lib/elastomer/client/bulk.rb | 131 ++++++++++++++++++++++++++++++++---
 test/client/bulk_test.rb     |  26 +++++++
 2 files changed, 148 insertions(+), 9 deletions(-)

diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb
index ab85c999..0f953114 100644
--- a/lib/elastomer/client/bulk.rb
+++ b/lib/elastomer/client/bulk.rb
@@ -43,6 +43,48 @@ def bulk( body = nil, params = nil )
       end
     end
 
+    # Stream bulk actions from an Enumerator.
+    #
+    # Examples
+    #
+    #   ops = [
+    #     Bulk::index(document1, :_type => 'default-type', :_id => 1),
+    #     Bulk::index(document2, params2),
+    #     Bulk::delete(params3),
+    #   ]
+    #   bulk(ops, :index => 'default-index').each do |response|
+    #     puts response
+    #   end
+    #
+    # Returns an Enumerator of responses.
+    def bulk_stream_responses(ops, params = {})
+      bulk_obj = Bulk.new(self, params)
+
+      Enumerator.new do |yielder|
+        ops.each do |op|
+          response = bulk_obj.add_to_actions *op
+          yielder.yield response unless response.nil?
+        end
+
+        response = bulk_obj.call
+        yielder.yield response unless response.nil?
+      end
+    end
+
+    # Stream bulk actions from an Enumerator.
+    #
+    # See Client#bulk_stream_responses for more information.
+    #
+    # Returns an Enumerator of items from the responses.
+    def bulk_stream_items(ops, params = {})
+      Enumerator.new do |yielder|
+        bulk_stream_responses(ops, params).each do |response|
+          response["items"].each do |item|
+            yielder.yield item
+          end
+        end
+      end
+    end
 
     # The Bulk class provides some abstractions and helper methods for working
     # with the ElasticSearch bulk API command. Instances of the Bulk class
@@ -132,7 +174,7 @@ def action_count=(value)
       #
       # Returns the response from the bulk call if one was made or nil.
       def index( document, params = {} )
-        params = prepare_params(document, params)
+        params = Bulk::prepare_params(document, params)
         add_to_actions({:index => params}, document)
       end
 
@@ -151,7 +193,7 @@ def index( document, params = {} )
       #
       # Returns the response from the bulk call if one was made or nil.
       def create( document, params )
-        params = prepare_params(document, params)
+        params = Bulk::prepare_params(document, params)
         add_to_actions({:create => params}, document)
       end
 
@@ -170,7 +212,7 @@ def create( document, params )
       #
       # Returns the response from the bulk call if one was made or nil.
       def update( document, params )
-        params = prepare_params(document, params)
+        params = Bulk::prepare_params(document, params)
         add_to_actions({:update => params}, document)
       end
 
@@ -184,10 +226,81 @@ def update( document, params )
       #
       # Returns the response from the bulk call if one was made or nil.
       def delete( params )
-        params = prepare_params(nil, params)
+        params = Bulk::prepare_params(nil, params)
         add_to_actions({:delete => params})
       end
 
+      # Construct a bulk action which indexes a document. Parameters can be
+      # provided in the parameters hash (underscore prefix optional) or in the
+      # document hash (underscore prefix required).
+      #
+      # document - The document to index as a Hash or JSON encoded String
+      # params   - Parameters for the index action (as a Hash) (optional)
+      #
+      # Examples
+      #   index({"foo" => "bar"}, {:_id => 1, :_type => "foo", :_index => "foo"})
+      #   index({"foo" => "bar"}, {:id => 1, :type => "foo", :_index => "foo"})
+      #   index("foo" => "bar", "_id" => 1, "_type" => "foo", "_index" => "foo")
+      #   # => [{:index => {"_id" => 1, "_type" => "foo", "_index" => "foo"}}, {"foo" => "bar"}]
+      #
+      # Returns an Array representing the rows of the action in a bulk call
+      def self.op_index( document, params = {} )
+        [{:index => prepare_params(document, params)}, document]
+      end
+
+      # Construct a bulk action which creates a document. Parameters can be
+      # provided in the parameters hash (underscore prefix optional) or in the
+      # document hash (underscore prefix required).
+      #
+      # document - The document to create as a Hash or JSON encoded String
+      # params   - Parameters for the create action (as a Hash) (optional)
+      #
+      # Examples
+      #   create({"foo" => "bar"}, {:_id => 1, :_type => "foo", :_index => "foo"})
+      #   create({"foo" => "bar"}, {:id => 1, :type => "foo", :_index => "foo"})
+      #   create("foo" => "bar", "_id" => 1, "_type" => "foo", "_index" => "foo")
+      #   # => [{:create => {"_id" => 1, "_type" => "foo", "_index" => "foo"}}, {"foo" => "bar"}]
+      #
+      # Returns an Array representing the rows of the action in a bulk call
+      def self.op_create( document, params )
+        [{:create => prepare_params(document, params)}, document]
+      end
+
+      # Construct a bulk action which updates a document. Parameters can be
+      # provided in the parameters hash (underscore prefix optional) or in the
+      # document hash (underscore prefix required).
+      #
+      # document - The document to update as a Hash or JSON encoded String
+      # params   - Parameters for the update action (as a Hash) (optional)
+      #
+      # Examples
+      #   update({"foo" => "bar"}, {:_id => 1, :_type => "foo", :_index => "foo"})
+      #   update({"foo" => "bar"}, {:id => 1, :type => "foo", :_index => "foo"})
+      #   update("foo" => "bar", "_id" => 1, "_type" => "foo", "_index" => "foo")
+      #   # => [{:update => {"_id" => 1, "_type" => "foo", "_index" => "foo"}}, {"foo" => "bar"}]
+      #
+      # Returns an Array representing the rows of the action in a bulk call
+      def self.op_update( document, params )
+        [{:update => prepare_params(document, params)}, document]
+      end
+
+      # Construct a bulk action which deletes a document. Parameters can be
+      # provided in the parameters hash (underscore prefix optional) or in the
+      # document hash (underscore prefix required).
+      #
+      # params - Parameters for the delete action (as a Hash)
+      #
+      # Examples
+      #   delete({"foo" => "bar"}, {:_id => 1, :_type => "foo", :_index => "foo"})
+      #   delete({"foo" => "bar"}, {:id => 1, :type => "foo", :_index => "foo"})
+      #   delete("foo" => "bar", "_id" => 1, "_type" => "foo", "_index" => "foo")
+      #   # => [{:delete => {"_id" => 1, "_type" => "foo", "_index" => "foo"}}, {"foo" => "bar"}]
+      #
+      # Returns an Array representing the rows of the action in a bulk call
+      def self.op_delete( params )
+        [{:delete => prepare_params(nil, params)}]
+      end
+
       # Immediately execute a bulk API call with the currently accumulated
       # actions. The accumulated actions list will be cleared after the call
       # has been made.
@@ -211,7 +324,7 @@ def call
 
       # Internal: convert special key parameters to their wire representation
       # and apply any override document parameters.
-      def prepare_params(document, params)
+      def self.prepare_params(document, params)
         params = convert_special_keys(params)
         if document.is_a? Hash
           params = from_document(document).merge(params)
@@ -227,7 +340,7 @@ def prepare_params(document, params)
       # document - The document Hash
       #
       # Returns extracted key/value pairs as a Hash.
-      def from_document( document )
+      def self.from_document( document )
         opts = {}
 
         SPECIAL_KEYS_HASH.values.each do |field|
@@ -249,7 +362,7 @@ def from_document( document )
       # params - Hash.
       #
       # Returns a new params Hash with the special keys replaced.
-      def convert_special_keys(params)
+      def self.convert_special_keys(params)
         new_params = params.dup
 
         SPECIAL_KEYS_HASH.each do |k1, k2|
@@ -268,8 +381,8 @@ def convert_special_keys(params)
       # size, then a bulk API call will be performed. After the call the
       # actions list is cleared and we'll start accumulating actions again.
       #
-      # action   - The bulk action (as a Hash) to perform
-      # document - Optional document for the action as a Hash or JSON encoded String
+      # action - The bulk action (as a Hash) to perform
+      # document  - Optional document for the action as a Hash or JSON encoded String
       #
       # Returns the response from the bulk call if one was made or nil.
       def add_to_actions( action, document = nil )
diff --git a/test/client/bulk_test.rb b/test/client/bulk_test.rb
index e783b73b..7e74a0c7 100644
--- a/test/client/bulk_test.rb
+++ b/test/client/bulk_test.rb
@@ -293,4 +293,30 @@
     assert_equal 'tweet', items[0]['index']['_type']
     assert_equal 1, items[0]['index']['_version']
   end
+
+  it 'streams bulk responses' do
+    ops = [
+      Elastomer::Client::Bulk::op_index({ :message => 'tweet 1' }, { :_id => 1, :_type => 'book', :_index => @index.name }),
+      Elastomer::Client::Bulk::op_index({ :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }),
+      Elastomer::Client::Bulk::op_index({ :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name })
+    ]
+    responses = $client.bulk_stream_responses(ops, { :action_count => 2 }).to_a
+    assert_equal(2, responses.length)
+    assert_bulk_index(responses[0]["items"][0])
+    assert_bulk_index(responses[0]["items"][1])
+    assert_bulk_index(responses[1]["items"][0])
+  end
+
+  it 'streams bulk items' do
+    ops = [
+      Elastomer::Client::Bulk::op_index({ :message => 'tweet 1' }, { :_id => 1, :_type => 'book', :_index => @index.name }),
+      Elastomer::Client::Bulk::op_index({ :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }),
+      Elastomer::Client::Bulk::op_index({ :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name })
+    ]
+    items = $client.bulk_stream_items(ops, { :action_count => 2 }).to_a
+    assert_equal(3, items.length)
+    assert_bulk_index(items[0])
+    assert_bulk_index(items[1])
+    assert_bulk_index(items[2])
+  end
 end

From 702ea16639c3f9dcb5c13771c6b2039c2705e3f4 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Wed, 16 Sep 2015 19:15:10 -0700
Subject: [PATCH 03/12] Whitespace

---
 lib/elastomer/client/bulk.rb | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb
index 0f953114..e43f1310 100644
--- a/lib/elastomer/client/bulk.rb
+++ b/lib/elastomer/client/bulk.rb
@@ -381,8 +381,8 @@ def self.convert_special_keys(params)
       # size, then a bulk API call will be performed. After the call the
       # actions list is cleared and we'll start accumulating actions again.
       #
-      # action - The bulk action (as a Hash) to perform
-      # document  - Optional document for the action as a Hash or JSON encoded String
+      # action   - The bulk action (as a Hash) to perform
+      # document - Optional document for the action as a Hash or JSON encoded String
       #
       # Returns the response from the bulk call if one was made or nil.
       def add_to_actions( action, document = nil )

From 5ba86c84a45e6c2e1f7983f5660b0f1537f5613a Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Wed, 16 Sep 2015 19:29:41 -0700
Subject: [PATCH 04/12] Use streaming bulk in delete_by_query

---
 lib/elastomer/client/delete_by_query.rb | 25 +++++++++++--------------
 1 file changed, 11 insertions(+), 14 deletions(-)

diff --git a/lib/elastomer/client/delete_by_query.rb b/lib/elastomer/client/delete_by_query.rb
index d6ee0efa..07551773 100644
--- a/lib/elastomer/client/delete_by_query.rb
+++ b/lib/elastomer/client/delete_by_query.rb
@@ -94,15 +94,13 @@ def categorize(item)
       #
       # response - A bulk response
       def accumulate(response)
-        unless response.nil?
-          @response_stats['took'] += response['took']
+        @response_stats['took'] += response['took']
 
-          response['items'].each do |item|
-            item = item['delete']
-            (@response_stats['_indices'][item['_index']] ||= {}).merge!(categorize(item)) { |_, n, m| n + m }
-            @response_stats['_indices']['_all'].merge!(categorize(item)) { |_, n, m| n + m }
-            @response_stats['failures'] << item unless is_ok? item['status']
-          end
+        response['items'].each do |item|
+          item = item['delete']
+          (@response_stats['_indices'][item['_index']] ||= {}).merge!(categorize(item)) { |_, n, m| n + m }
+          @response_stats['_indices']['_all'].merge!(categorize(item)) { |_, n, m| n + m }
+          @response_stats['failures'] << item unless is_ok? item['status']
         end
       end
 
@@ -110,14 +108,13 @@ def accumulate(response)
       #
       # Returns a Hash of statistics about the bulk operation
       def execute
-        # accumulate is called both inside and outside the bulk block in order
-        # to capture bulk responses returned from calls to `delete` and the call
-        # to `bulk`
-        accumulate(@client.bulk(@params) do |bulk|
+        ops = Enumerator.new do |yielder|
           @client.scan(@query, @params).each_document do |hit|
-            accumulate(bulk.delete(_id: hit["_id"], _type: hit["_type"], _index: hit["_index"]))
+            yielder.yield(Elastomer::Client::Bulk::op_delete(_id: hit["_id"], _type: hit["_type"], _index: hit["_index"]))
           end
-        end)
+        end
+
+        @client.bulk_stream_responses(ops, @params).each { |response| accumulate(response) }
 
         @response_stats
       end

From 1c977c097bc30062f97fdf9280dc51a3ee1b7212 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Wed, 16 Sep 2015 19:30:51 -0700
Subject: [PATCH 05/12] Pass the search hits directly from scan to bulk

---
 lib/elastomer/client/delete_by_query.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/elastomer/client/delete_by_query.rb b/lib/elastomer/client/delete_by_query.rb
index 07551773..80ec1ac7 100644
--- a/lib/elastomer/client/delete_by_query.rb
+++ b/lib/elastomer/client/delete_by_query.rb
@@ -110,7 +110,7 @@ def accumulate(response)
       def execute
         ops = Enumerator.new do |yielder|
           @client.scan(@query, @params).each_document do |hit|
-            yielder.yield(Elastomer::Client::Bulk::op_delete(_id: hit["_id"], _type: hit["_type"], _index: hit["_index"]))
+            yielder.yield(Elastomer::Client::Bulk::op_delete(hit))
           end
         end
 

From db5f4c0de1bd4d48a580db03f62036afed611620 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Wed, 16 Sep 2015 19:32:01 -0700
Subject: [PATCH 06/12] Revert "Pass the search hits directly from scan to
 bulk"

This reverts commit 1c977c097bc30062f97fdf9280dc51a3ee1b7212.

Elasticsearch throws an error when `_score` is present:

```
Elastomer::Client::ServerError: IllegalArgumentException[Action/metadata line [1] contains an unknown parameter [_score]]
```
---
 lib/elastomer/client/delete_by_query.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/elastomer/client/delete_by_query.rb b/lib/elastomer/client/delete_by_query.rb
index 80ec1ac7..07551773 100644
--- a/lib/elastomer/client/delete_by_query.rb
+++ b/lib/elastomer/client/delete_by_query.rb
@@ -110,7 +110,7 @@ def accumulate(response)
       def execute
         ops = Enumerator.new do |yielder|
           @client.scan(@query, @params).each_document do |hit|
-            yielder.yield(Elastomer::Client::Bulk::op_delete(hit))
+            yielder.yield(Elastomer::Client::Bulk::op_delete(_id: hit["_id"], _type: hit["_type"], _index: hit["_index"]))
           end
         end
 

From 7cf006285d670d46c195d223280f1ed04efab087 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Thu, 17 Sep 2015 16:34:11 -0700
Subject: [PATCH 07/12] Return stats from bulk_stream_items and yield each item

---
 lib/elastomer/client/bulk.rb            | 56 +++++++++++++++++++++----
 lib/elastomer/client/delete_by_query.rb | 22 ++++------
 test/client/bulk_test.rb                |  3 +-
 3 files changed, 60 insertions(+), 21 deletions(-)

diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb
index e43f1310..579147fc 100644
--- a/lib/elastomer/client/bulk.rb
+++ b/lib/elastomer/client/bulk.rb
@@ -71,19 +71,61 @@ def bulk_stream_responses(ops, params = {})
       end
     end
 
-    # Stream bulk actions from an Enumerator.
+    # Internal: Determine whether or not a response item has an HTTP status code
+    # in the range 200 to 299.
+    #
+    # item - The bulk response item
+    #
+    # Returns a boolean
+    def is_ok?(item)
+      item.values.first["status"].between?(200, 299)
+    end
+
+    # Stream bulk actions from an Enumerator and passes them to the given block.
+    #
+    # Examples
+    #
+    #   ops = [
+    #     Bulk::index(document1, :_type => 'default-type', :_id => 1),
+    #     Bulk::index(document2, params2),
+    #     Bulk::delete(params3),
+    #   ]
+    #   bulk_stream_items(ops, :index => 'default-index') do |item|
+    #     puts item
+    #   end
     #
-    # See Client#bulk_stream_responses for more information.
+    #   # return value:
+    #   # {
+    #   #   "took" => 256,
+    #   #   "errors" => false,
+    #   #   "success" => 3,
+    #   #   "failure" => 0
+    #   # }
     #
-    # Returns an Enumerator of items from the responses.
+    # Returns a Hash of stats about items from the responses.
     def bulk_stream_items(ops, params = {})
-      Enumerator.new do |yielder|
-        bulk_stream_responses(ops, params).each do |response|
-          response["items"].each do |item|
-            yielder.yield item
+      stats = {
+        "took" => 0,
+        "errors" => false,
+        "success" => 0,
+        "failure" => 0
+      }
+
+      bulk_stream_responses(ops, params).each do |response|
+        stats["took"] += response["took"]
+        stats["errors"] |= response["errors"]
+
+        response["items"].each do |item|
+          if is_ok?(item)
+            stats["success"] += 1
+          else
+            stats["failure"] += 1
           end
+          yield item
         end
       end
+
+      stats
     end
 
     # The Bulk class provides some abstractions and helper methods for working
diff --git a/lib/elastomer/client/delete_by_query.rb b/lib/elastomer/client/delete_by_query.rb
index 07551773..a57bbcda 100644
--- a/lib/elastomer/client/delete_by_query.rb
+++ b/lib/elastomer/client/delete_by_query.rb
@@ -90,18 +90,14 @@ def categorize(item)
         }
       end
 
-      # Internal: Combine the items in a response with the existing statistics
+      # Internal: Combine a response item with the existing statistics
       #
-      # response - A bulk response
-      def accumulate(response)
-        @response_stats['took'] += response['took']
-
-        response['items'].each do |item|
-          item = item['delete']
-          (@response_stats['_indices'][item['_index']] ||= {}).merge!(categorize(item)) { |_, n, m| n + m }
-          @response_stats['_indices']['_all'].merge!(categorize(item)) { |_, n, m| n + m }
-          @response_stats['failures'] << item unless is_ok? item['status']
-        end
+      # item - A bulk response item
+      def accumulate(item)
+        item = item["delete"]
+        (@response_stats['_indices'][item['_index']] ||= {}).merge!(categorize(item)) { |_, n, m| n + m }
+        @response_stats['_indices']['_all'].merge!(categorize(item)) { |_, n, m| n + m }
+        @response_stats['failures'] << item unless is_ok? item['status']
       end
 
       # Perform the Delete by Query action
@@ -114,8 +110,8 @@ def execute
           end
         end
 
-        @client.bulk_stream_responses(ops, @params).each { |response| accumulate(response) }
-
+        stats = @client.bulk_stream_items(ops, @params) { |item| accumulate(item) }
+        @response_stats['took'] = stats['took']
         @response_stats
       end
 
diff --git a/test/client/bulk_test.rb b/test/client/bulk_test.rb
index 7e74a0c7..4ea92385 100644
--- a/test/client/bulk_test.rb
+++ b/test/client/bulk_test.rb
@@ -313,7 +313,8 @@
       Elastomer::Client::Bulk::op_index({ :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }),
       Elastomer::Client::Bulk::op_index({ :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name })
     ]
-    items = $client.bulk_stream_items(ops, { :action_count => 2 }).to_a
+    items = []
+    stats = $client.bulk_stream_items(ops, { :action_count => 2 }) { |item| items << item }
     assert_equal(3, items.length)
     assert_bulk_index(items[0])
     assert_bulk_index(items[1])

From 74682c2e872fb2534394c13f9a1d7c73516d31e1 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Thu, 17 Sep 2015 16:34:49 -0700
Subject: [PATCH 08/12] Fix documentation example for bulk_stream_responses

---
 lib/elastomer/client/bulk.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb
index 579147fc..94ce84be 100644
--- a/lib/elastomer/client/bulk.rb
+++ b/lib/elastomer/client/bulk.rb
@@ -52,7 +52,7 @@ def bulk( body = nil, params = nil )
     #     Bulk::index(document2, params2),
     #     Bulk::delete(params3),
     #   ]
-    #   bulk(ops, :index => 'default-index').each do |response|
+    #   bulk_stream_responses(ops, :index => 'default-index').each do |response|
     #     puts response
     #   end
     #

From c90963b93189bd4fa0c5ed8e6fe604ca293e2664 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Thu, 17 Sep 2015 16:55:44 -0700
Subject: [PATCH 09/12] Adjust the bulk streaming interface in order to reuse
 existing methods

---
 lib/elastomer/client/bulk.rb            | 87 +++----------------------
 lib/elastomer/client/delete_by_query.rb |  2 +-
 test/client/bulk_test.rb                | 12 ++--
 3 files changed, 15 insertions(+), 86 deletions(-)

diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb
index 94ce84be..3ea4fb1b 100644
--- a/lib/elastomer/client/bulk.rb
+++ b/lib/elastomer/client/bulk.rb
@@ -48,9 +48,9 @@ def bulk( body = nil, params = nil )
     # Examples
     #
     #   ops = [
-    #     Bulk::index(document1, :_type => 'default-type', :_id => 1),
-    #     Bulk::index(document2, params2),
-    #     Bulk::delete(params3),
+    #     [:index, document1, {:_type => "foo", :_id => 1}],
+    #     [:create, document2],
+    #     [:delete, {:_type => "bar", :_id => 42}]
     #   ]
     #   bulk_stream_responses(ops, :index => 'default-index').each do |response|
     #     puts response
@@ -61,8 +61,8 @@ def bulk_stream_responses(ops, params = {})
       bulk_obj = Bulk.new(self, params)
 
       Enumerator.new do |yielder|
-        ops.each do |op|
-          response = bulk_obj.add_to_actions *op
+        ops.each do |action, *args|
+          response = bulk_obj.send(action, *args)
           yielder.yield response unless response.nil?
         end
 
@@ -86,9 +86,9 @@ def is_ok?(item)
     # Examples
     #
     #   ops = [
-    #     Bulk::index(document1, :_type => 'default-type', :_id => 1),
-    #     Bulk::index(document2, params2),
-    #     Bulk::delete(params3),
+    #     [:index, document1, {:_type => "foo", :_id => 1}],
+    #     [:create, document2],
+    #     [:delete, {:_type => "bar", :_id => 42}]
     #   ]
     #   bulk_stream_items(ops, :index => 'default-index') do |item|
     #     puts item
@@ -272,77 +272,6 @@ def delete( params )
         add_to_actions({:delete => params})
       end
 
-      # Construct a bulk action which indexes a document. Parameters can be
-      # provided in the parameters hash (underscore prefix optional) or in the
-      # document hash (underscore prefix required).
-      #
-      # document - The document to index as a Hash or JSON encoded String
-      # params   - Parameters for the index action (as a Hash) (optional)
-      #
-      # Examples
-      #   index({"foo" => "bar"}, {:_id => 1, :_type => "foo", :_index => "foo"})
-      #   index({"foo" => "bar"}, {:id => 1, :type => "foo", :_index => "foo"})
-      #   index("foo" => "bar", "_id" => 1, "_type" => "foo", "_index" => "foo")
-      #   # => [{:index => {"_id" => 1, "_type" => "foo", "_index" => "foo"}}, {"foo" => "bar"}]
-      #
-      # Returns an Array representing the rows of the action in a bulk call
-      def self.op_index( document, params = {} )
-        [{:index => prepare_params(document, params)}, document]
-      end
-
-      # Construct a bulk action which creates a document. Parameters can be
-      # provided in the parameters hash (underscore prefix optional) or in the
-      # document hash (underscore prefix required).
-      #
-      # document - The document to create as a Hash or JSON encoded String
-      # params   - Parameters for the create action (as a Hash) (optional)
-      #
-      # Examples
-      #   create({"foo" => "bar"}, {:_id => 1, :_type => "foo", :_index => "foo"})
-      #   create({"foo" => "bar"}, {:id => 1, :type => "foo", :_index => "foo"})
-      #   create("foo" => "bar", "_id" => 1, "_type" => "foo", "_index" => "foo")
-      #   # => [{:create => {"_id" => 1, "_type" => "foo", "_index" => "foo"}}, {"foo" => "bar"}]
-      #
-      # Returns an Array representing the rows of the action in a bulk call
-      def self.op_create( document, params )
-        [{:create => prepare_params(document, params)}, document]
-      end
-
-      # Construct a bulk action which updates a document. Parameters can be
-      # provided in the parameters hash (underscore prefix optional) or in the
-      # document hash (underscore prefix required).
-      #
-      # document - The document to update as a Hash or JSON encoded String
-      # params   - Parameters for the update action (as a Hash) (optional)
-      #
-      # Examples
-      #   update({"foo" => "bar"}, {:_id => 1, :_type => "foo", :_index => "foo"})
-      #   update({"foo" => "bar"}, {:id => 1, :type => "foo", :_index => "foo"})
-      #   update("foo" => "bar", "_id" => 1, "_type" => "foo", "_index" => "foo")
-      #   # => [{:update => {"_id" => 1, "_type" => "foo", "_index" => "foo"}}, {"foo" => "bar"}]
-      #
-      # Returns an Array representing the rows of the action in a bulk call
-      def self.op_update( document, params )
-        [{:update => prepare_params(document, params)}, document]
-      end
-
-      # Construct a bulk action which deletes a document. Parameters can be
-      # provided in the parameters hash (underscore prefix optional) or in the
-      # document hash (underscore prefix required).
-      #
-      # params - Parameters for the delete action (as a Hash)
-      #
-      # Examples
-      #   delete({"foo" => "bar"}, {:_id => 1, :_type => "foo", :_index => "foo"})
-      #   delete({"foo" => "bar"}, {:id => 1, :type => "foo", :_index => "foo"})
-      #   delete("foo" => "bar", "_id" => 1, "_type" => "foo", "_index" => "foo")
-      #   # => [{:delete => {"_id" => 1, "_type" => "foo", "_index" => "foo"}}, {"foo" => "bar"}]
-      #
-      # Returns an Array representing the rows of the action in a bulk call
-      def self.op_delete( params )
-        [{:delete => prepare_params(nil, params)}]
-      end
-
       # Immediately execute a bulk API call with the currently accumulated
       # actions. The accumulated actions list will be cleared after the call
       # has been made.
diff --git a/lib/elastomer/client/delete_by_query.rb b/lib/elastomer/client/delete_by_query.rb
index a57bbcda..52eb955a 100644
--- a/lib/elastomer/client/delete_by_query.rb
+++ b/lib/elastomer/client/delete_by_query.rb
@@ -106,7 +106,7 @@ def accumulate(item)
       def execute
         ops = Enumerator.new do |yielder|
           @client.scan(@query, @params).each_document do |hit|
-            yielder.yield(Elastomer::Client::Bulk::op_delete(_id: hit["_id"], _type: hit["_type"], _index: hit["_index"]))
+            yielder.yield([:delete, { _id: hit["_id"], _type: hit["_type"], _index: hit["_index"] }])
           end
         end
 
diff --git a/test/client/bulk_test.rb b/test/client/bulk_test.rb
index 4ea92385..f56fb109 100644
--- a/test/client/bulk_test.rb
+++ b/test/client/bulk_test.rb
@@ -296,9 +296,9 @@
 
   it 'streams bulk responses' do
     ops = [
-      Elastomer::Client::Bulk::op_index({ :message => 'tweet 1' }, { :_id => 1, :_type => 'book', :_index => @index.name }),
-      Elastomer::Client::Bulk::op_index({ :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }),
-      Elastomer::Client::Bulk::op_index({ :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name })
+      [:index, { :message => 'tweet 1' }, { :_id => 1, :_type => 'book', :_index => @index.name }],
+      [:index, { :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }],
+      [:index, { :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name }]
     ]
     responses = $client.bulk_stream_responses(ops, { :action_count => 2 }).to_a
     assert_equal(2, responses.length)
@@ -309,9 +309,9 @@
 
   it 'streams bulk items' do
     ops = [
-      Elastomer::Client::Bulk::op_index({ :message => 'tweet 1' }, { :_id => 1, :_type => 'book', :_index => @index.name }),
-      Elastomer::Client::Bulk::op_index({ :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }),
-      Elastomer::Client::Bulk::op_index({ :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name })
+      [:index, { :message => 'tweet 1' }, { :_id => 1, :_type => 'book', :_index => @index.name }],
+      [:index, { :message => 'tweet 2' }, { :_id => 2, :_type => 'book', :_index => @index.name }],
+      [:index, { :message => 'tweet 3' }, { :_id => 3, :_type => 'book', :_index => @index.name }]
     ]
     items = []
     stats = $client.bulk_stream_items(ops, { :action_count => 2 }) { |item| items << item }

From 151d9320246decb42aa4ceaad91e1a3989ee9444 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Fri, 18 Sep 2015 09:41:01 -0700
Subject: [PATCH 10/12] Revert back to instance methods

This is a continuation of c90963b.
---
 lib/elastomer/client/bulk.rb | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb
index 3ea4fb1b..0a5a15be 100644
--- a/lib/elastomer/client/bulk.rb
+++ b/lib/elastomer/client/bulk.rb
@@ -216,7 +216,7 @@ def action_count=(value)
       #
       # Returns the response from the bulk call if one was made or nil.
       def index( document, params = {} )
-        params = Bulk::prepare_params(document, params)
+        params = prepare_params(document, params)
         add_to_actions({:index => params}, document)
       end
 
@@ -235,7 +235,7 @@ def index( document, params = {} )
       #
       # Returns the response from the bulk call if one was made or nil.
       def create( document, params )
-        params = Bulk::prepare_params(document, params)
+        params = prepare_params(document, params)
         add_to_actions({:create => params}, document)
       end
 
@@ -254,7 +254,7 @@ def create( document, params )
       #
       # Returns the response from the bulk call if one was made or nil.
       def update( document, params )
-        params = Bulk::prepare_params(document, params)
+        params = prepare_params(document, params)
         add_to_actions({:update => params}, document)
       end
 
@@ -268,7 +268,7 @@ def update( document, params )
       #
       # Returns the response from the bulk call if one was made or nil.
       def delete( params )
-        params = Bulk::prepare_params(nil, params)
+        params = prepare_params(nil, params)
         add_to_actions({:delete => params})
       end
 
@@ -295,7 +295,7 @@ def call
 
       # Internal: convert special key parameters to their wire representation
       # and apply any override document parameters.
-      def self.prepare_params(document, params)
+      def prepare_params(document, params)
         params = convert_special_keys(params)
         if document.is_a? Hash
           params = from_document(document).merge(params)
@@ -311,7 +311,7 @@ def self.prepare_params(document, params)
       # document - The document Hash
       #
       # Returns extracted key/value pairs as a Hash.
-      def self.from_document( document )
+      def from_document( document )
         opts = {}
 
         SPECIAL_KEYS_HASH.values.each do |field|
@@ -333,7 +333,7 @@ def self.from_document( document )
       # params - Hash.
       #
       # Returns a new params Hash with the special keys replaced.
-      def self.convert_special_keys(params)
+      def convert_special_keys(params)
         new_params = params.dup
 
         SPECIAL_KEYS_HASH.each do |k1, k2|

From f25925868b8ea76ecedba98f7c49c2bf77c37290 Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Fri, 18 Sep 2015 10:28:42 -0700
Subject: [PATCH 11/12] Add a sample response item to the `bulk_stream_items`
 docs

---
 lib/elastomer/client/bulk.rb | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/lib/elastomer/client/bulk.rb b/lib/elastomer/client/bulk.rb
index 0a5a15be..a9777090 100644
--- a/lib/elastomer/client/bulk.rb
+++ b/lib/elastomer/client/bulk.rb
@@ -81,7 +81,8 @@ def is_ok?(item)
       item.values.first["status"].between?(200, 299)
     end
 
-    # Stream bulk actions from an Enumerator and passes them to the given block.
+    # Stream bulk actions from an Enumerator and passes the response items to
+    # the given block.
     #
     # Examples
     #
@@ -102,6 +103,18 @@ def is_ok?(item)
     #   #   "failure" => 0
     #   # }
     #
+    #   # sample response item:
+    #   # {
+    #   #   "delete": {
+    #   #     "_index": "foo",
+    #   #     "_type": "bar",
+    #   #     "_id": "42",
+    #   #     "_version": 3,
+    #   #     "status": 200,
+    #   #     "found": true
+    #   #   }
+    #   # }
+    #
     # Returns a Hash of stats about items from the responses.
     def bulk_stream_items(ops, params = {})
       stats = {

From 129c372b81b9247b87bd937c8bbd06bd09b005fd Mon Sep 17 00:00:00 2001
From: Chris Wendt <chrismwendt@gmail.com>
Date: Fri, 18 Sep 2015 10:52:22 -0700
Subject: [PATCH 12/12] Bump version

---
 CHANGELOG.md             | 4 ++++
 lib/elastomer/version.rb | 2 +-
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b12dea55..9f3d184e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.7.0 (2015-09-18)
+- Add streaming bulk functionality via `bulk_stream_items`
+- Make Delete by Query compatible with Elasticsearch 2.0
+
 ## 0.6.0 (2015-09-11)
 - Support all URL parameters when using `Client.#scroll`
 - BREAKING: Moved some `Scroller` reader methods into `Scroller.opts`
diff --git a/lib/elastomer/version.rb b/lib/elastomer/version.rb
index 4a4028c4..b3c0e9e9 100644
--- a/lib/elastomer/version.rb
+++ b/lib/elastomer/version.rb
@@ -1,5 +1,5 @@
 module Elastomer
-  VERSION = '0.6.0'
+  VERSION = '0.7.0'
 
   def self.version
     VERSION