[chrismwendt]

This adds streaming bulk functionality via Ruby's Enumerators.

This does not to alter any of the existing methods in backwards incompatible ways so that a conversation about making the API more consistent could take place later. Because of that, some of the small operation methods had to be nearly duplicated.

This solves #92.

/cc @TwP @grantr

[chrismwendt]

Is there a simpler way to essentially map over an Enumerator than by constructing a new one and yielding each element of the original, as I did here?

[TwP]

One thought on this method is to not return an Enumerator. Instead, accept a block and perform the iteration directly with this method.

def bulk_stream_items(ops, params = {})
  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

  return stats
end

I've also added a hash of interesting numbers as the return value from this method. The is_ok? method does not exist yet, but it would check the HTTP status returned with each item.

def is_ok?(item)
  item["status"].between?(200, 299)
end

Thoughts on this approach?

[chrismwendt]

The client might want to retry failed documents, in which case they would need access to at least the failed response items. Maybe it's worth having a separate convenient method called bulk_stream_stats which does exactly what you suggest.

[TwP]

The client does have access to the failed items during the iteration. When yield item is called, the client can also check the status code of that particular operation.

Returning the stats hash is there just to be nice to the end user.

[chrismwendt]

My apologies, I read over your code too quickly and missed that yield item 🙇

I like the convenience of returning stats, and the block syntax would look almost the same as calling each on the Enumerator. I'll implement it like you showed.

[TwP]

I'm not a huge fan of the Bulk::index / Bulk::delete syntax here. My proposal is this:

ops = [
  [:index, document1, {:_type => "foo", :_id => 1}],
  [:create, document2],
  [:delete, {:_type => "bar", :_id => 42}]
]

The first item in the list is the bulk action to take. This corresponds to one of the action methods on the Bulk instance. The second item in the list is a document (or the params in the case of :delete). And thirdly comes the optional params.

Inside the Enumerator we have:

Enumerator.new do |yielder|
  ops.each do |action, *args|
    response = bulk_obj.send(action, *args)
    yielder.yield response unless response.nil?
  end
  response = bulk_obj.call
  yielder.yield response unless response.nil?
end

The Ruby send method is being used here to call the action method on the bulk_obj and passing the *args to the method.

[chrismwendt]

That's a pretty clever use of send to reuse the existing methods! I'll implement that now.

[TwP]

This can be reverted back to an instance method.

[grantr]

Can we document what the item typically looks like?

[chrismwendt]

@TwP: I reverted those class methods back to instance methods.

@grantr: I added a sample response item to the docs.

I also bumped the version and updated the changelog.

[TwP]

👍 ✨ and that's enough emoji for now.

[grantr]

This is great stuff, useful and easy to use. 👍