Repository: elastic-rs/elastic (language: rust)

Issue:
Add support for Bulk
It looks like the main way `elastic` is used is for bulk operations. We should make this more convenient so users don't have to manually construct the json payload themselves if they don't want.

I think it would be good to offer something like:

```rust
let bulk_ops = BulkOps::new();

for doc in docs {
    let id = doc.id;
    let op = BulkOp::insert(doc).index("my_index").ty("my_ty").id(id);
    bulk_ops.add(op);
}

client.bulk()
      .ops(bulk_ops);
```

It'd be nice to figure out how to configure the op without having to define a local for it, we could do:

```rust
let bulk_ops = BulkOps::new();

for doc in docs {
    let id = doc.id;
    bulk_ops.insert(doc).index("my_index").ty("my_ty").id(id);
}

client.bulk()
      .ops(bulk_ops);
```

Where `bulk_ops.insert` returns an `IndexBulkOpBuilder<'a>` that calls `add` on drop. I think we need to be careful with that though.

Or if you want to supply your own payload:

```rust
client.bulk()
      .body(bulk_body);
```

The `BulkOps` object should be able to efficiently construct a payload as more operations are added. I think just formatting into a buffer is the easiest way to support multiple serialisable types.


Discussion / hints:

Fix patch:
diff --git a/.travis.yml b/.travis.yml
index 0325cee50e..c2d1b2f818 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -13,10 +13,6 @@ matrix:
       env:
         - CHANNEL="nightly"
         - KIND="build"
-    - rust: nightly
-      env:
-        - CHANNEL="nightly"
-        - KIND="bench"
     - rust: nightly
       env:
         - CHANNEL="nightly"
diff --git a/src/elastic/Cargo.toml b/src/elastic/Cargo.toml
index 1cdcf6b3b8..39fb45df41 100644
--- a/src/elastic/Cargo.toml
+++ b/src/elastic/Cargo.toml
@@ -30,6 +30,8 @@ tokio-core = "~0.1.9"
 tokio-io = "~0.1.4"
 futures-cpupool = "~0.1.6"
 fluent_builder = "~0.5"
+tokio-timer = "~0.1.2"
+crossbeam-channel = "~0.1.2"
 
 elastic_requests = { version = "~0.20.2", path = "../requests" }
 elastic_responses = { version = "~0.20.2", path = "../responses" }
diff --git a/src/elastic/examples/bulk.rs b/src/elastic/examples/bulk.rs
index c7091f33ed..1de4e86b67 100644
--- a/src/elastic/examples/bulk.rs
+++ b/src/elastic/examples/bulk.rs
@@ -9,6 +9,8 @@
 
 extern crate elastic;
 extern crate env_logger;
+#[macro_use]
+extern crate serde_json;
 
 use std::error::Error;
 use elastic::prelude::*;
@@ -17,11 +19,20 @@ fn run() -> Result<(), Box<Error>> {
     // A HTTP client and request parameters
     let client = SyncClientBuilder::new().build()?;
 
+    let ops = (0..1000)
+        .into_iter()
+        .map(|i| bulk_index(json!({
+                "id": i,
+                "title": "some string value"
+            }))
+            .id(i));
+
     // Execute a bulk request
-    let bulk = client
-        .request(BulkRequest::new(bulk_body()))
-        .send()?
-        .into_response::<BulkResponse>()?;
+    let bulk = client.bulk()
+        .index("bulk_idx")
+        .ty("bulk_ty")
+        .extend(ops)
+        .send()?;
 
     for op in bulk {
         match op {
@@ -33,24 +44,6 @@ fn run() -> Result<(), Box<Error>> {
     Ok(())
 }
 
-fn bulk_body() -> String {
-    let mut bulk = String::new();
-    for i in 1..1000 {
-        let header = format!(
-            "{{ \"index\" : {{ \"_index\" : \"test\", \"_type\" : \"ty\", \"_id\" : \"{}\" }} }}",
-            i
-        );
-        let body = format!("{{ \"title\" : \"string value {}\" }}", i);
-
-        bulk.push_str(&header);
-        bulk.push('\n');
-        bulk.push_str(&body);
-        bulk.push('\n');
-    }
-
-    bulk
-}
-
 fn main() {
     env_logger::init();
     run().unwrap()
diff --git a/src/elastic/examples/bulk_async.rs b/src/elastic/examples/bulk_async.rs
index f28f89327a..ad48ccc1c7 100644
--- a/src/elastic/examples/bulk_async.rs
+++ b/src/elastic/examples/bulk_async.rs
@@ -11,6 +11,8 @@ extern crate elastic;
 extern crate env_logger;
 extern crate futures;
 extern crate tokio_core;
+#[macro_use]
+extern crate serde_json;
 
 use std::error::Error;
 use futures::Future;
@@ -23,13 +25,22 @@ fn run() -> Result<(), Box<Error>> {
     // A HTTP client and request parameters
     let client = AsyncClientBuilder::new().build(&core.handle())?;
 
+    let ops = (0..1000)
+        .into_iter()
+        .map(|i| bulk_index(json!({
+                "id": i,
+                "title": "some string value"
+            }))
+            .id(i));
+
     // Execute a bulk request
-    let res_future = client
-        .request(BulkRequest::new(bulk_body()))
-        .send()
-        .and_then(|res| res.into_response::<BulkResponse>());
+    let res_future = client.bulk()
+        .index("bulk_idx")
+        .ty("bulk_ty")
+        .extend(ops)
+        .send();
 
-    let bulk_future = res_future.and_then(|bulk| {
+    let res_future = res_future.and_then(|bulk| {
         for op in bulk {
             match op {
                 Ok(op) => println!("ok: {:?}", op),
@@ -40,29 +51,11 @@ fn run() -> Result<(), Box<Error>> {
         Ok(())
     });
 
-    core.run(bulk_future)?;
+    core.run(res_future)?;
 
     Ok(())
 }
 
-fn bulk_body() -> String {
-    let mut bulk = String::new();
-    for i in 1..1000 {
-        let header = format!(
-            "{{ \"index\" : {{ \"_index\" : \"test\", \"_type\" : \"ty\", \"_id\" : \"{}\" }} }}",
-            i
-        );
-        let body = format!("{{ \"title\" : \"string value {}\" }}", i);
-
-        bulk.push_str(&header);
-        bulk.push('\n');
-        bulk.push_str(&body);
-        bulk.push('\n');
-    }
-
-    bulk
-}
-
 fn main() {
     env_logger::init();
     run().unwrap()
diff --git a/src/elastic/examples/bulk_async_stream.rs b/src/elastic/examples/bulk_async_stream.rs
new file mode 100644
index 0000000000..e8e74930c0
--- /dev/null
+++ b/src/elastic/examples/bulk_async_stream.rs
@@ -0,0 +1,68 @@
+//! Send a bulk request.
+//!
+//! NOTE: This sample expects you have a node running on `localhost:9200`.
+//!
+//! This sample demonstrates a bulk request.
+//! There are a few extra dials you can tweak on bulk requests to get more
+//! performance out of them.
+//! See the docs for `BulkResponse` for more details.
+
+extern crate elastic;
+extern crate env_logger;
+extern crate futures;
+extern crate tokio_core;
+#[macro_use]
+extern crate serde_json;
+
+use std::error::Error;
+use std::time::Duration;
+use futures::{stream, Future, Stream, Sink};
+use tokio_core::reactor::Core;
+use elastic::prelude::*;
+
+fn run() -> Result<(), Box<Error>> {
+    let mut core = Core::new()?;
+
+    // A HTTP client and request parameters
+    let client = AsyncClientBuilder::new().build(&core.handle())?;
+
+    // Get a stream for bulk operations
+    // Individual operations can be sent to the stream and will be buffered to Elasticsearch
+    let (bulk_stream, bulk_responses) = client.bulk_stream()
+        .index("bulk_idx")
+        .ty("bulk_ty")
+        .timeout(Duration::from_secs(5))
+        .body_size_bytes(1024)
+        .build();
+
+    let ops = (0..1000)
+        .into_iter()
+        .map(|i| bulk_index(json!({
+                "id": i,
+                "title": "some string value"
+            }))
+            .id(i));
+
+    let req_future = bulk_stream.send_all(stream::iter_ok(ops));
+
+    let res_future = bulk_responses.for_each(|bulk| {
+        println!("response:");
+        for op in bulk {
+            match op {
+                Ok(op) => println!("  ok: {:?}", op),
+                Err(op) => println!("  err: {:?}", op),
+            }
+        }
+
+        Ok(())
+    });
+
+    core.run(req_future.join(res_future))?;
+
+    Ok(())
+}
+
+fn main() {
+    env_logger::init().unwrap();
+    run().unwrap()
+}
diff --git a/src/elastic/src/client/mod.rs b/src/elastic/src/client/mod.rs
index aa549396db..25c8e7a1c6 100644
--- a/src/elastic/src/client/mod.rs
+++ b/src/elastic/src/client/mod.rs
@@ -102,6 +102,7 @@ They're exposed as methods on the `Client`:
 Client method                                                 | Elasticsearch API                  | Raw request type                                        | Response type
 ------------------------------------------------------------- | ---------------------------------- | ------------------------------------------------------- | ------------------------------------
 [`search`][Client.search]                                     | [Search][docs-search]              | [`SearchRequest`][SearchRequest]                        | [`SearchResponse`][SearchResponse]
+[`bulk`][Client.bulk]                                         | [Bulk][docs-bulk]                  | [`BulkRequest`][BulkRequest]                            | [`BulkResponse`][BulkResponse]
 [`document_get`][Client.document_get]                         | [Get Document][docs-get]           | [`GetRequest`][GetRequest]                              | [`GetResponse`][GetResponse]
 [`document_index`][Client.document_index]                     | [Index Document][docs-index]       | [`IndexRequest`][IndexRequest]                          | [`IndexResponse`][IndexResponse]
 [`document_update`][Client.document_update]                   | [Update Document][docs-update]     | [`UpdateRequest`][UpdateRequest]                        | [`UpdateResponse`][UpdateResponse]
@@ -437,6 +438,7 @@ future.and_then(|body| {
 `AsyncHttpResponse` implements the async `Stream` trait so you can buffer out the raw response data.
 For more details see the [`responses`][responses-mod] module.
 
+[docs-bulk]: http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
 [docs-search]: http://www.elastic.co/guide/en/elasticsearch/reference/current/search-search.html
 [docs-get]: http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-get.html
 [docs-update]: http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update.html
@@ -458,6 +460,7 @@ For more details see the [`responses`][responses-mod] module.
 [AsyncClient]: type.AsyncClient.html
 [AsyncClientBuilder]: struct.AsyncClientBuilder.html
 [Client.request]: struct.Client.html#method.request
+[Client.bulk]: struct.Client.html#bulk-request
 [Client.search]: struct.Client.html#search-request
 [Client.document_get]: struct.Client.html#get-document-request
 [Client.document_update]: struct.Client.html#update-document-request
@@ -475,6 +478,7 @@ For more details see the [`responses`][responses-mod] module.
 [RequestBuilder.params]: requests/struct.RequestBuilder.html#method.params
 [RawRequestBuilder]: requests/type.RawRequestBuilder.html
 [SearchRequest]: requests/endpoints/struct.SearchRequest.html
+[BulkRequest]: requests/endpoints/struct.BulkRequest.html
 [GetRequest]: requests/endpoints/struct.GetRequest.html
 [UpdateRequest]: requests/endpoints/struct.UpdateRequest.html
 [DeleteRequest]: requests/endpoints/struct.DeleteRequest.html
@@ -494,10 +498,11 @@ For more details see the [`responses`][responses-mod] module.
 [AsyncResponseBuilder]: responses/struct.AsyncResponseBuilder.html
 [AsyncResponseBuilder.into_response]: responses/struct.AsyncResponseBuilder.html#method.into_response
 [AsyncResponseBuilder.into_raw]: responses/struct.AsyncResponseBuilder.html#method.into_raw
-[SearchResponse]: responses/type.SearchResponse.html
-[GetResponse]: responses/type.GetResponse.html
-[UpdateResponse]: responses/type.UpdateResponse.html
-[DeleteResponse]: responses/type.DeleteResponse.html
+[SearchResponse]: responses/struct.SearchResponse.html
+[BulkResponse]: responses/struct.BulkResponse.html
+[GetResponse]: responses/struct.GetResponse.html
+[UpdateResponse]: responses/struct.UpdateResponse.html
+[DeleteResponse]: responses/struct.DeleteResponse.html
 [IndexResponse]: responses/struct.IndexResponse.html
 [IndicesExistsResponse]: responses/struct.IndicesExistsResponse.html
 [PingResponse]: responses/struct.PingResponse.html
diff --git a/src/elastic/src/client/requests/bulk/mod.rs b/src/elastic/src/client/requests/bulk/mod.rs
new file mode 100644
index 0000000000..6ab0b1288f
--- /dev/null
+++ b/src/elastic/src/client/requests/bulk/mod.rs
@@ -0,0 +1,896 @@
+/*!
+Builders for [bulk requests][docs-bulk].
+
+[docs-bulk]: https://www.elastic.co/guide/en/elasticsearch/reference/current/bulk.html
+*/
+
+use std::fmt;
+use std::error::Error as StdError;
+use std::time::Duration;
+use std::marker::PhantomData;
+
+use futures::{Future, Poll};
+use serde::ser::Serialize;
+use serde::de::DeserializeOwned;
+
+use error::{self, Error};
+use client::{Client, RequestParams};
+use client::sender::{AsyncSender, Sender, SyncSender};
+use client::requests::RequestBuilder;
+use http::{SyncBody, AsyncBody};
+use client::requests::params::{Index, Type};
+use client::requests::endpoints::BulkRequest;
+use client::requests::raw::RawRequestInner;
+use client::responses::{BulkResponse, BulkErrorsResponse};
+use client::responses::parse::IsOk;
+
+/**
+A [bulk request][docs-bulk] builder that can be configured before sending. 
+
+Call [`Client.bulk`][Client.bulk] to get a `BulkRequestBuilder`.
+The `send` method will either send the request [synchronously][send-sync] or [asynchronously][send-async], depending on the `Client` it was created from.
+
+Call [`Client.bulk_stream`][Client.bulk_stream] to get a `BulkRequestBuilder` that can be used to stream bulk operations asynchronously.
+
+[docs-bulk]: https://www.elastic.co/guide/en/elasticsearch/reference/current/bulk.html
+[send-sync]: #send-synchronously
+[send-async]: #send-asynchronously
+[Client.bulk]: ../../struct.Client.html#bulk-request
+[Client.bulk_stream]: ../../struct.Client.html#bulk-stream-request
+*/
+pub type BulkRequestBuilder<TSender, TBody, TResponse> = RequestBuilder<TSender, BulkRequestInner<TBody, TResponse>>;
+
+mod stream;
+mod operation;
+
+pub use self::stream::*;
+pub use self::operation::*;
+
+#[doc(hidden)]
+pub struct BulkRequestInner<TBody, TResponse> {
+    index: Option<Index<'static>>,
+    ty: Option<Type<'static>>,
+    body: WrappedBody<TBody>,
+    _marker: PhantomData<TResponse>,
+}
+
+/**
+# Bulk request
+*/
+impl<TSender> Client<TSender>
+where
+    TSender: Sender,
+{
+    /** 
+    Create a [`BulkRequestBuilder`][BulkRequestBuilder] with this `Client` that can be configured before sending.
+
+    For more details, see:
+
+    - [builder methods][builder-methods]
+    - [send synchronously][send-sync]
+    - [send asynchronously][send-async]
+
+    # Examples
+
+    Send a bulk request to index some documents:
+
+    ```no_run
+    # extern crate serde;
+    # #[macro_use]
+    # extern crate serde_derive;
+    # #[macro_use]
+    # extern crate elastic_derive;
+    # extern crate elastic;
+    # use elastic::prelude::*;
+    # fn main() { run().unwrap() }
+    # fn run() -> Result<(), Box<::std::error::Error>> {
+    # #[derive(Serialize, Deserialize, ElasticType)]
+    # struct MyType {
+    #     pub id: i32,
+    #     pub title: String,
+    # }
+    # let client = SyncClientBuilder::new().build()?;
+    let ops = (0..1000)
+        .into_iter()
+        .map(|i| bulk_index(MyType {
+                id: i,
+                title: "some string value".into()
+            })
+            .id(i));
+
+    let response = client.bulk()
+                         .index("myindex")
+                         .ty(MyType::name())
+                         .extend(ops)
+                         .send()?;
+
+    for op in response {
+        match op {
+            Ok(op) => println!("ok: {:?}", op),
+            Err(op) => println!("err: {:?}", op),
+        }
+    }
+    # Ok(())
+    # }
+    ```
+
+    [BulkRequestBuilder]: requests/bulk/type.BulkRequestBuilder.html
+    [builder-methods]: requests/bulk/type.BulkRequestBuilder.html#builder-methods
+    [send-sync]: requests/bulk/type.BulkRequestBuilder.html#send-synchronously
+    [send-async]: requests/bulk/type.BulkRequestBuilder.html#send-asynchronously
+    [types-mod]: ../../types/index.html
+    [documents-mod]: ../../types/document/index.html
+    [docs-querystring]: https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-query-string-query.html
+    */
+    pub fn bulk(&self) -> BulkRequestBuilder<TSender, Vec<u8>, BulkResponse> {
+        RequestBuilder::initial(
+            self.clone(),
+            BulkRequestInner {
+                index: None,
+                ty: None,
+                body: WrappedBody::new(Vec::new()),
+                _marker: PhantomData,
+            },
+        )
+    }
+}
+
+/**
+# Bulk stream request
+*/
+impl Client<AsyncSender> {
+    /** 
+    Create a [`BulkRequestBuilder`][BulkRequestBuilder] with this `Client` that can be configured before sending.
+
+    This method can configure a channel that individual bulk operations can be sent to.
+    The operations will be batched and debounced in the backgroun rather than being sent immediately.
+
+    For more details, see:
+
+    - [builder methods][builder-methods]
+    - [stream builder methods][stream-builder-methods]
+    - [send synchronously][send-sync]
+    - [send asynchronously][send-async]
+
+    # Examples
+
+    Stream a bulk request to index some documents:
+
+    ```no_run
+    # extern crate serde;
+    # extern crate futures;
+    # extern crate tokio_core;
+    # #[macro_use] extern crate serde_derive;
+    # #[macro_use] extern crate elastic_derive;
+    # extern crate elastic;
+    # use std::time::Duration;
+    # use futures::{Future, Stream, Sink};
+    # use elastic::prelude::*;
+    # fn main() { run().unwrap() }
+    # fn run() -> Result<(), Box<::std::error::Error>> {
+    # #[derive(Serialize, Deserialize, ElasticType)]
+    # struct MyType {
+    #     pub id: i32,
+    #     pub title: String,
+    # }
+    # let core = tokio_core::reactor::Core::new()?;
+    # let client = AsyncClientBuilder::new().build(&core.handle())?;
+    let (bulk_stream, bulk_responses) = client.bulk_stream()
+        .index("bulk_idx")
+        .ty(MyType::name())
+        .timeout(Duration::from_secs(5))
+        .body_size_bytes(1024)
+        .build();
+
+    let ops = (0..1000)
+        .into_iter()
+        .map(|i| bulk_index(MyType {
+                id: i,
+                title: "some string value".into()
+            })
+            .id(i));
+
+    let req_future = bulk_stream.send_all(futures::stream::iter_ok(ops));
+
+    let res_future = bulk_responses.for_each(|bulk| {
+        println!("response:");
+        for op in bulk {
+            match op {
+                Ok(op) => println!("  ok: {:?}", op),
+                Err(op) => println!("  err: {:?}", op),
+            }
+        }
+
+        Ok(())
+    });
+    # Ok(())
+    # }
+    ```
+
+    [BulkRequestBuilder]: requests/bulk/type.BulkRequestBuilder.html
+    [builder-methods]: requests/bulk/type.BulkRequestBuilder.html#stream-builder-methods\
+    [send-sync]: requests/bulk/type.BulkRequestBuilder.html#send-synchronously
+    [send-async]: requests/bulk/type.BulkRequestBuilder.html#send-asynchronously
+    [types-mod]: ../../types/index.html
+    [documents-mod]: ../../types/document/index.html
+    [docs-querystring]: https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-query-string-query.html
+    */
+    pub fn bulk_stream<TDocument>(&self) -> BulkRequestBuilder<AsyncSender, Streamed<TDocument>, BulkResponse> {
+        RequestBuilder::initial(
+            self.clone(),
+            BulkRequestInner {
+                index: None,
+                ty: None,
+                body: WrappedBody::new(Streamed::new()),
+                _marker: PhantomData,
+            },
+        )
+    }
+}
+
+/**
+# Builder methods
+
+Configure a `BulkRequestBuilder` before sending it.
+*/
+impl<TSender, TBody, TResponse> BulkRequestBuilder<TSender, TBody, TResponse>
+where
+    TSender: Sender,
+{
+    /**
+    Set the default type for the bulk request.
+    
+    If an operation doesn't specify a type, then it will default to the supplied value here.
+    
+    # Deferred errors
+    
+    Calling `ty` without also calling `index` will result in an error when sending the request.
+    */
+    pub fn ty<I>(mut self, ty: I) -> Self
+    where
+        I: Into<Type<'static>>,
+    {
+        self.inner.ty = Some(ty.into());
+        self
+    }
+
+    /**
+    Set the default index for the bulk request.
+    
+    If an operation doesn't specify an index, then it will default to the supplied value here.
+    */
+    pub fn index<I>(mut self, index: I) -> Self
+    where
+        I: Into<Index<'static>>,
+    {
+        self.inner.index = Some(index.into());
+        self
+    }
+
+    /**
+    Set the type used to deserialize the index field on the response.
+    
+    Sometimes a bulk response will use the same index value many times.
+    To avoid allocating a lot of individual strings, the type used to deserialize the field can be changed.
+    `string_cache::DefaultAtom` or a custom `enum` can be effective ways to reduce allocations in large bulk responses.
+    */
+    pub fn response_index<I>(self) -> BulkRequestBuilder<TSender, TBody, TResponse::WithNewIndex>
+    where
+        TResponse: ChangeIndex<I>,
+    {
+        RequestBuilder::new(
+            self.client,
+            self.params_builder,
+            BulkRequestInner {
+                index: self.inner.index,
+                ty: self.inner.ty,
+                body: self.inner.body,
+                _marker: PhantomData,
+            },
+        )
+    }
+
+    /**
+    Set the type used to deserialize the type field on the response.
+    
+    Sometimes a bulk response will use the same type value many times.
+    To avoid allocating a lot of individual strings, the type used to deserialize the field can be changed.
+    `string_cache::DefaultAtom` or a custom `enum` can be effective ways to reduce allocations in large bulk responses.
+    */
+    pub fn response_ty<I>(self) -> BulkRequestBuilder<TSender, TBody, TResponse::WithNewType>
+    where
+        TResponse: ChangeType<I>,
+    {
+        RequestBuilder::new(
+            self.client,
+            self.params_builder,
+            BulkRequestInner {
+                index: self.inner.index,
+                ty: self.inner.ty,
+                body: self.inner.body,
+                _marker: PhantomData,
+            },
+        )
+    }
+
+    /**
+    Set the type used to deserialize the id field on the response.
+    
+    It's less likely that id fields in bulk responses will be repeated, but they're probably short.
+    To avoid allocating a lot of individual strings, the type used to deserialize the field can be changed.
+    `inlinable_string::InlinableString` can be an effective way to recude allocation in large bulk responses.
+    */
+    pub fn response_id<I>(self) -> BulkRequestBuilder<TSender, TBody, TResponse::WithNewId>
+    where
+        TResponse: ChangeId<I>,
+    {
+        RequestBuilder::new(
+            self.client,
+            self.params_builder,
+            BulkRequestInner {
+                index: self.inner.index,
+                ty: self.inner.ty,
+                body: self.inner.body,
+                _marker: PhantomData,
+            },
+        )
+    }
+}
+
+impl<TSender, TBody, TIndex, TType, TId> BulkRequestBuilder<TSender, TBody, BulkResponse<TIndex, TType, TId>>
+where
+    TSender: Sender,
+{
+    /**
+    Only deserialize failed bulk operations in the response.
+
+    Elasticsearch returns a response that's proportional in size to the number of operations in the request.
+    If you only care about failures then it can be more efficient to ignore the common case where operations succeed.
+    */
+    pub fn errors_only(self) -> BulkRequestBuilder<TSender, TBody, BulkErrorsResponse<TIndex, TType, TId>> {
+        RequestBuilder::new(
+            self.client,
+            self.params_builder,
+            BulkRequestInner {
+                index: self.inner.index,
+                ty: self.inner.ty,
+                body: self.inner.body,
+                _marker: PhantomData,
+            },
+        )
+    }
+}
+
+impl<TSender, TBody, TResponse> BulkRequestBuilder<TSender, TBody, TResponse>
+where
+    TSender: Sender,
+    TBody: BulkBody,
+{
+    fn push_internal<TDocument, TOperation>(&mut self, op: TOperation)
+    where
+        TOperation: Into<BulkOperation<TDocument>>,
+        TDocument: Serialize,
+    {
+        self.inner.body.with_inner_mut(|b| b.push(op.into()));
+    }
+
+    /**
+    Push an operation onto the bulk request.
+
+    # Deferred errors
+
+    If the document can't be serialized then sending the request will return an error.
+    */
+    pub fn push<TDocument, TOperation>(mut self, op: TOperation) -> Self
+    where
+        TOperation: Into<BulkOperation<TDocument>>,
+        TDocument: Serialize,
+    {
+        self.push_internal(op);
+        self
+    }
+
+    /**
+    Push a collection of operations onto the bulk request.
+
+    # Deferred errors
+
+    If any documents can't be serialized then sending the request will return an error.
+    */
+    pub fn extend<TIter, TDocument>(mut self, iter: TIter) -> Self
+    where
+        TIter: IntoIterator<Item = BulkOperation<TDocument>>,
+        TDocument: Serialize,
+    {
+        for op in iter.into_iter() {
+            self.push_internal(op);
+        }
+        self
+    }
+}
+
+impl<TSender, TBody, TDocument, TResponse> Extend<BulkOperation<TDocument>> for BulkRequestBuilder<TSender, TBody, TResponse>
+where
+    TSender: Sender,
+    TBody: BulkBody,
+    TDocument: Serialize,
+{
+    fn extend<T>(&mut self, iter: T) where
+    T: IntoIterator<Item = BulkOperation<TDocument>>,
+    {
+        for op in iter.into_iter() {
+            self.push_internal(op);
+        }
+    }
+}
+
+/**
+# Stream builder methods
+
+Configure a `SearchRequestBuilder` before sending it.
+*/
+impl<TDocument, TResponse> BulkRequestBuilder<AsyncSender, Streamed<TDocument>, TResponse> {
+    /**
+    Specify a timeout for filling up the request buffer.
+
+    This parameter can be used to control the miminum frequency of bulk requests.
+    If the timeout expires before the buffer is full then a bulk request will be sent with whatever data was written.
+    The timeout isn't restarted when operations are pushed.
+    */
+    pub fn timeout(mut self, timeout: Duration) -> Self {
+        self.inner.body.with_inner_mut(|s| {
+            s.timeout = timeout;
+            Ok(())
+        });
+
+        self
+    }
+
+    /**
+    Specify a maximum request size in bytes.
+
+    This parameter can be used to control the maximum size of a single bulk request emitted.
+    Operations will be appended to the request until the `body_size` is reached.
+    */
+    pub fn body_size_bytes(mut self, body_size: usize) -> Self {
+        self.inner.body.with_inner_mut(|s| {
+            s.body_size = body_size;
+            Ok(())
+        });
+
+        self
+    }
+
+    /**
+    Create a channel for streaming bulk operations.
+
+    This will return a channel with a [`BulkSender`] and [`BulkReceiver`] pair.
+    Push operations into the sender.
+    Once an internal buffer is full, or a timeout expires then the bulk request will be sent.
+    Responses can be pulled by the receiver.
+
+    # Examples
+
+    > TODO
+    */
+    pub fn build(self) -> (BulkSender<TDocument, TResponse>, BulkReceiver<TResponse>) {
+        let body = self.inner.body.try_into_inner().expect("building a stream should be infallible");
+
+        let body_size = body.body_size;
+        let duration = body.timeout;
+
+        let params = self.params_builder.into_value(RequestParams::default);
+        let body = SenderBody::new(body_size);
+        let timeout = Timeout::new(duration);
+        let req_template = SenderRequestTemplate::new(self.client, params, self.inner.index, self.inner.ty);
+
+        BulkSender::new(req_template, timeout, body)
+    }
+}
+
+impl<TBody, TResponse> BulkRequestInner<TBody, TResponse>
+where
+    TBody: BulkBody,
+{
+    fn into_request(self) -> Result<BulkRequest<'static, TBody>, Error> {
+        let body = self.body.try_into_inner()?;
+
+        match (self.index, self.ty) {
+            (Some(index), None) => Ok(BulkRequest::for_index(
+                index,
+                body,
+            )),
+            (Some(index), Some(ty)) => Ok(BulkRequest::for_index_ty(
+                index,
+                ty,
+                body,
+            )),
+            (None, None) => Ok(BulkRequest::new(
+                body,
+            )),
+            (None, Some(_)) => Err(error::request(BulkRequestError("missing `index` parameter".to_owned())))
+        }
+    }
+}
+
+/**
+# Send synchronously
+*/
+impl<TBody, TResponse> BulkRequestBuilder<SyncSender, TBody, TResponse>
+where
+    TBody: Into<SyncBody> + BulkBody + 'static,
+    TResponse: DeserializeOwned + IsOk + 'static,
+{
+    /**
+    Send a `BulkRequestBuilder` synchronously using a [`SyncClient`][SyncClient].
+
+    This will block the current thread until a response arrives and is deserialised.
+
+    # Examples
+
+    Send a bulk request to index some documents:
+
+    ```no_run
+    # extern crate serde;
+    # #[macro_use]
+    # extern crate serde_derive;
+    # #[macro_use]
+    # extern crate elastic_derive;
+    # extern crate elastic;
+    # use elastic::prelude::*;
+    # fn main() { run().unwrap() }
+    # fn run() -> Result<(), Box<::std::error::Error>> {
+    # #[derive(Serialize, Deserialize, ElasticType)]
+    # struct MyType {
+    #     pub id: i32,
+    #     pub title: String,
+    # }
+    # let client = SyncClientBuilder::new().build()?;
+    let ops = (0..1000)
+        .into_iter()
+        .map(|i| bulk_index(MyType {
+                id: i,
+                title: "some string value".into()
+            })
+            .id(i));
+
+    let response = client.bulk()
+                         .index("myindex")
+                         .ty(MyType::name())
+                         .extend(ops)
+                         .send()?;
+
+    for op in response {
+        match op {
+            Ok(op) => println!("ok: {:?}", op),
+            Err(op) => println!("err: {:?}", op),
+        }
+    }
+    # Ok(())
+    # }
+    ```
+
+    [SyncClient]: ../../type.SyncClient.html
+    */
+    pub fn send(self) -> Result<TResponse, Error> {
+        let req = self.inner.into_request()?;
+
+        RequestBuilder::new(self.client, self.params_builder, RawRequestInner::new(req))
+            .send()?
+            .into_response()
+    }
+}
+
+/**
+# Send asynchronously
+*/
+impl<TBody, TResponse> BulkRequestBuilder<AsyncSender, TBody, TResponse>
+where
+    TBody: Into<AsyncBody> + BulkBody + Send + 'static,
+    TResponse: DeserializeOwned + IsOk + Send + 'static,
+{
+    /**
+    Send a `BulkRequestBuilder` asynchronously using an [`AsyncClient`][AsyncClient].
+    
+    This will return a future that will resolve to the deserialised search response.
+
+    # Examples
+
+    Send a bulk request to index some documents:
+
+    ```no_run
+    # extern crate serde;
+    # extern crate futures;
+    # extern crate tokio_core;
+    # #[macro_use] extern crate serde_derive;
+    # #[macro_use] extern crate elastic_derive;
+    # extern crate elastic;
+    # use futures::Future;
+    # use elastic::prelude::*;
+    # fn main() { run().unwrap() }
+    # fn run() -> Result<(), Box<::std::error::Error>> {
+    # #[derive(Serialize, Deserialize, ElasticType)]
+    # struct MyType {
+    #     pub id: i32,
+    #     pub title: String,
+    # }
+    # let core = tokio_core::reactor::Core::new()?;
+    # let client = AsyncClientBuilder::new().build(&core.handle())?;
+    let ops = (0..1000)
+        .into_iter()
+        .map(|i| bulk_index(MyType {
+                id: i,
+                title: "some string value".into()
+            })
+            .id(i));
+
+    let future = client.bulk()
+                         .index("myindex")
+                         .ty(MyType::name())
+                         .extend(ops)
+                         .send();
+
+    future.and_then(|response| {
+        for op in response {
+            match op {
+                Ok(op) => println!("ok: {:?}", op),
+                Err(op) => println!("err: {:?}", op),
+            }
+        }
+
+        Ok(())
+    });
+    # Ok(())
+    # }
+    ```
+
+    [AsyncClient]: ../../type.AsyncClient.html
+    */
+    pub fn send(self) -> Pending<TResponse> {
+        let (client, params_builder, inner) = (self.client, self.params_builder, self.inner);
+
+        let req_future = client.sender.maybe_async(move || inner.into_request());
+
+        let res_future = req_future.and_then(move |req| {
+            RequestBuilder::new(client, params_builder, RawRequestInner::new(req))
+                .send()
+                .and_then(|res| res.into_response())
+        });
+
+        Pending::new(res_future)
+    }
+}
+
+const DEFAULT_BODY_SIZE: usize = 1024 * 1024 * 5;
+const DEFAULT_TIMEOUT_SECS: u64 = 30;
+
+/**
+A streaming bulk request body.
+*/
+pub struct Streamed<TDocument> {
+    body_size: usize,
+    timeout: Duration,
+    _marker: PhantomData<TDocument>,
+}
+
+impl<TDocument> Streamed<TDocument> {
+    fn new() -> Self {
+        Streamed {
+            body_size: DEFAULT_BODY_SIZE,
+            timeout: Duration::from_secs(DEFAULT_TIMEOUT_SECS),
+            _marker: PhantomData,
+        }
+    }
+}
+
+struct WrappedBody<T> {
+    inner: T,
+    errs: Vec<Error>,
+}
+
+impl<T> WrappedBody<T> {
+    fn new(inner: T) -> Self {
+        WrappedBody {
+            inner,
+            errs: Vec::new()
+        }
+    }
+
+    fn with_inner_mut<F>(&mut self, f: F)
+    where
+        F: FnOnce(&mut T) -> Result<(), Error>
+    {
+        if let Err(e) = f(&mut self.inner) {
+            self.errs.push(e);
+        }
+    }
+
+    fn try_into_inner(self) -> Result<T, Error> {
+        if self.errs.len() > 0 {
+            Err(error::request(BulkBodyError(self.errs)))
+        }
+        else {
+            Ok(self.inner)
+        }
+    }
+}
+
+#[derive(Debug)]
+struct BulkBodyError(Vec<Error>);
+
+impl fmt::Display for BulkBodyError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        writeln!(f, "errors ({}) writing bulk request body:", self.0.len())?;
+
+        for err in &self.0 {
+            writeln!(f, "{}", err)?;
+        }
+
+        Ok(())
+    }
+}
+
+impl StdError for BulkBodyError {
+    fn description(&self) -> &str {
+        "errors writing bulk request body"
+    }
+}
+
+#[derive(Debug)]
+struct BulkRequestError(String);
+
+impl fmt::Display for BulkRequestError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        writeln!(f, "{}", self.0)
+    }
+}
+
+impl StdError for BulkRequestError {
+    fn description(&self) -> &str {
+        "error building bulk request body"
+    }
+}
+
+/**
+A bulk request body.
+
+The body can receive a bulk operation for any type of document.
+*/
+pub trait BulkBody {
+    /**
+    Push a new document onto the end of the body.
+
+    # Errors
+
+    If the document can't be serialized then this method will return an error.
+    There's no guarantee that other operations can be pushed onto the body after an error has occurred.
+    */
+    fn push<TDocument>(&mut self, op: BulkOperation<TDocument>) -> Result<(), Error> where TDocument: Serialize;
+}
+
+impl BulkBody for Vec<u8> {
+    fn push<TDocument>(&mut self, op: BulkOperation<TDocument>) -> Result<(), Error> where TDocument: Serialize {
+        op.write(self).map_err(error::request)?;
+
+        Ok(())
+    }
+}
+
+/** A future returned by calling `send`. */
+pub struct Pending<TResponse> {
+    inner: Box<Future<Item = TResponse, Error = Error>>,
+}
+
+impl<TResponse> Pending<TResponse> {
+    fn new<F>(fut: F) -> Self
+    where
+        F: Future<Item = TResponse, Error = Error> + 'static,
+    {
+        Pending {
+            inner: Box::new(fut),
+        }
+    }
+}
+
+impl<TResponse> Future for Pending<TResponse> {
+    type Item = TResponse;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.inner.poll()
+    }
+}
+
+#[doc(hidden)]
+pub trait ChangeIndex<TIndex> { type WithNewIndex; }
+
+impl<TIndex, TType, TId, TNewIndex> ChangeIndex<TNewIndex> for BulkResponse<TIndex, TType, TId> {
+    type WithNewIndex = BulkResponse<TNewIndex, TType, TId>;
+}
+
+impl<TIndex, TType, TId, TNewIndex> ChangeIndex<TNewIndex> for BulkErrorsResponse<TIndex, TType, TId> {
+    type WithNewIndex = BulkErrorsResponse<TNewIndex, TType, TId>;
+}
+
+#[doc(hidden)]
+pub trait ChangeType<TType> { type WithNewType; }
+
+impl<TIndex, TType, TId, TNewType> ChangeType<TNewType> for BulkResponse<TIndex, TType, TId> {
+    type WithNewType = BulkResponse<TIndex, TNewType, TId>;
+}
+
+impl<TIndex, TType, TId, TNewType> ChangeType<TNewType> for BulkErrorsResponse<TIndex, TType, TId> {
+    type WithNewType = BulkErrorsResponse<TIndex, TNewType, TId>;
+}
+
+#[doc(hidden)]
+pub trait ChangeId<TId> { type WithNewId; }
+
+impl<TIndex, TType, TId, TNewId> ChangeId<TNewId> for BulkResponse<TIndex, TType, TId> {
+    type WithNewId = BulkResponse<TIndex, TType, TNewId>;
+}
+
+impl<TIndex, TType, TId, TNewId> ChangeId<TNewId> for BulkErrorsResponse<TIndex, TType, TId> {
+    type WithNewId = BulkErrorsResponse<TIndex, TType, TNewId>;
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::{self, Value};
+    use prelude::*;
+
+    #[test]
+    fn default_request() {
+        let client = SyncClientBuilder::new().build().unwrap();
+
+        let req = client
+            .bulk()
+            .inner
+            .into_request()
+            .unwrap();
+
+        assert_eq!("/_bulk", req.url.as_ref());
+    }
+
+    #[test]
+    fn specify_index() {
+        let client = SyncClientBuilder::new().build().unwrap();
+
+        let req = client
+            .bulk()
+            .index("test-idx")
+            .inner
+            .into_request()
+            .unwrap();
+
+        assert_eq!("/test-idx/_bulk", req.url.as_ref());
+    }
+
+    #[test]
+    fn specify_index_ty() {
+        let client = SyncClientBuilder::new().build().unwrap();
+
+        let req = client
+            .bulk()
+            .index("test-idx")
+            .ty("new-ty")
+            .inner
+            .into_request()
+            .unwrap();
+
+        assert_eq!("/test-idx/new-ty/_bulk", req.url.as_ref());
+    }
+
+    #[test]
+    fn specify_ty() {
+        let client = SyncClientBuilder::new().build().unwrap();
+
+        let req = client
+            .bulk()
+            .ty("new-ty")
+            .inner
+            .into_request();
+
+        assert!(req.is_err());
+    }
+}
diff --git a/src/elastic/src/client/requests/bulk/operation.rs b/src/elastic/src/client/requests/bulk/operation.rs
new file mode 100644
index 0000000000..b6d548a39d
--- /dev/null
+++ b/src/elastic/src/client/requests/bulk/operation.rs
@@ -0,0 +1,248 @@
+use std::io::{self, Write};
+use std::ops::Deref;
+
+use serde::ser::{Serialize, Serializer, SerializeMap};
+use serde_json;
+
+use client::requests::params::{Index, Type, Id};
+use client::requests::common::{Doc, Script, ScriptBuilder, DefaultParams};
+
+pub use client::responses::bulk::Action;
+
+/**
+A bulk operation.
+*/
+pub struct BulkOperation<TValue> {
+    action: Action,
+    header: BulkHeader,
+    inner: Option<TValue>,
+}
+
+#[derive(Serialize)]
+struct BulkHeader {
+    #[serde(rename = "_index", serialize_with = "serialize_param", skip_serializing_if = "Option::is_none")]
+    index: Option<Index<'static>>,
+    #[serde(rename = "_type", serialize_with = "serialize_param", skip_serializing_if = "Option::is_none")]
+    ty: Option<Type<'static>>,
+    #[serde(rename = "_id", serialize_with = "serialize_param", skip_serializing_if = "Option::is_none")]
+    id: Option<Id<'static>>,
+}
+
+fn serialize_param<S, T>(field: &Option<T>, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+    T: Deref<Target = str>,
+{
+    serializer.serialize_str(&*field.as_ref().expect("serialize `None` value"))
+}
+
+impl<TDocument> BulkOperation<Doc<TDocument>> {
+    /**
+    A convenient method for creating an index bulk operation.
+    */
+    pub fn new_index(doc: TDocument) -> Self {
+        BulkOperation::new(Action::Index, Some(Doc::value(doc)))
+    }
+
+    /**
+    A convenient method for creating an update bulk operation.
+    */
+    pub fn new_update(doc: TDocument) -> Self {
+        BulkOperation::new(Action::Update, Some(Doc::value(doc)))
+    }
+
+    /**
+    A convenient method for creating a create bulk operation.
+    */
+    pub fn new_create(doc: TDocument) -> Self {
+        BulkOperation::new(Action::Create, Some(Doc::value(doc)))
+    }
+}
+
+impl BulkOperation<()> {
+    /**
+    A convenient method for creating a delete bulk operation.
+    */
+    pub fn new_delete() -> Self {
+        BulkOperation::new(Action::Delete, None)
+    }
+}
+
+impl BulkOperation<Script<DefaultParams>> {
+    /**
+    A convenient method for creating an update bulk operation.
+    */
+    pub fn new_update_script<TScript>(script: TScript) -> Self
+    where
+        TScript: ToString,
+    {
+        BulkOperation::new(Action::Update, Some(Script::new(script)))
+    }
+}
+
+impl<TParams> BulkOperation<Script<TParams>> {
+    /**
+    Set the script for this bulk operation.
+    */
+    pub fn script_fluent<TBuilder, TNewParams>(self, builder: TBuilder) -> BulkOperation<Script<TNewParams>>
+    where
+        TBuilder: Fn(ScriptBuilder<TParams>) -> ScriptBuilder<TNewParams>,
+    {
+        let inner = self.inner.map(|script| builder(ScriptBuilder::from_script(script)).build());
+
+        BulkOperation {
+            action: self.action,
+            header: self.header,
+            inner,
+        }
+    }
+}
+
+impl<TValue> BulkOperation<TValue> {
+    /**
+    Create a new operation for the given action.
+    */
+    fn new(action: Action, value: Option<TValue>) -> Self {
+        BulkOperation {
+            action,
+            header: BulkHeader {
+                index: None,
+                ty: None,
+                id: None,
+            },
+            inner: value,
+        }
+    }
+    
+    /**
+    Set the index for this bulk operation.
+    */
+    pub fn index<I>(mut self, index: I) -> Self
+    where
+        I: Into<Index<'static>>,
+    {
+        self.header.index = Some(index.into());
+        self
+    }
+
+    /**
+    Set the type for this bulk operation.
+    */
+    pub fn ty<I>(mut self, ty: I) -> Self
+    where
+        I: Into<Type<'static>>,
+    {
+        self.header.ty = Some(ty.into());
+        self
+    }
+
+    /**
+    Set the id for this bulk operation.
+    */
+    pub fn id<I>(mut self, id: I) -> Self
+    where
+        I: Into<Id<'static>>,
+    {
+        self.header.id = Some(id.into());
+        self
+    }
+}
+
+impl<TDocument> BulkOperation<TDocument>
+where
+    TDocument: Serialize
+{
+    /**
+    Write the operation to the given writer.
+
+    Bulk operations have a particular line-delimited format.
+    This method will write a json header, then a newline, then the document body.
+    */
+    pub fn write<W>(&self, mut writer: W) -> io::Result<()>
+        where
+            W: Write,
+    {
+        struct Header<'a> {
+            action: Action,
+            inner: &'a BulkHeader,
+        }
+
+        impl<'a> Serialize for Header<'a> {
+            fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+                where S: Serializer
+            {
+                let mut map = serializer.serialize_map(Some(1))?;
+
+                let k = match self.action {
+                    Action::Create => "create",
+                    Action::Delete => "delete",
+                    Action::Index => "index",
+                    Action::Update => "update",
+                };
+
+                map.serialize_entry(k, &self.inner)?;
+
+                map.end()
+            }
+        }
+
+        serde_json::to_writer(&mut writer, &Header { action: self.action, inner: &self.header })?;
+        write!(&mut writer, "\n")?;
+
+        if let Some(ref inner) = self.inner {
+            serde_json::to_writer(&mut writer, inner)?;
+            write!(&mut writer, "\n")?;
+        }
+
+        Ok(())
+    }
+}
+
+/**
+A convenient method for creating an index bulk operation.
+*/
+pub fn bulk_index<TDocument>(doc: TDocument) -> BulkOperation<Doc<TDocument>> {
+    BulkOperation::new_index(doc)
+}
+
+/**
+A convenient method for creating an update bulk operation.
+*/
+pub fn bulk_update<TDocument>(doc: TDocument) -> BulkOperation<Doc<TDocument>> {
+    BulkOperation::new_update(doc)
+}
+
+/**
+A convenient method for creating a create bulk operation.
+*/
+pub fn bulk_create<TDocument>(doc: TDocument) -> BulkOperation<Doc<TDocument>> {
+    BulkOperation::new_create(doc)
+}
+
+/**
+A convenient method for creating a delete bulk operation.
+*/
+pub fn bulk_delete() -> BulkOperation<()> {
+    BulkOperation::new(Action::Delete, None)
+}
+
+/**
+A convenient method for creating an update bulk operation.
+*/
+pub fn bulk_update_script<TScript>(script: TScript) -> BulkOperation<Script<DefaultParams>>
+where
+    TScript: ToString,
+{
+    BulkOperation::new_update_script(script)
+}
+
+/**
+A convenient method for creating an update bulk operation.
+*/
+pub fn bulk_update_script_fluent<TScript, TBuilder, TParams>(script: TScript, builder: TBuilder) -> BulkOperation<Script<TParams>>
+    where
+        TScript: ToString,
+        TBuilder: Fn(ScriptBuilder<DefaultParams>) -> ScriptBuilder<TParams>,
+{
+    BulkOperation::new_update_script(script).script_fluent(builder)
+}
diff --git a/src/elastic/src/client/requests/bulk/stream.rs b/src/elastic/src/client/requests/bulk/stream.rs
new file mode 100644
index 0000000000..174ef14722
--- /dev/null
+++ b/src/elastic/src/client/requests/bulk/stream.rs
@@ -0,0 +1,380 @@
+use std::{mem, fmt};
+use std::io;
+use std::time::Duration;
+use std::error::Error as StdError;
+use std::marker::PhantomData;
+
+use bytes::{BufMut, BytesMut};
+use futures::{Future, Poll, Async, AsyncSink, Sink, Stream};
+use fluent_builder::FluentBuilder;
+use tokio_timer::{Timer, Sleep};
+use channel::{self, TrySendError, TryRecvError};
+use serde::ser::Serialize;
+use serde::de::DeserializeOwned;
+
+use error::{self, Error};
+use client::{Client, RequestParams};
+use client::sender::AsyncSender;
+use client::requests::RequestBuilder;
+use client::requests::params::{Index, Type};
+use client::responses::parse::IsOk;
+use super::{BulkRequestBuilder, BulkRequestInner, Pending, BulkOperation, WrappedBody};
+
+/**
+The sending half of a stream of bulk operations.
+
+The sender accepts individual operations and keeps them in a buffer until a timer has expired or the buffer fills up.
+*/
+pub struct BulkSender<TDocument, TResponse> {
+    tx: BulkSenderInner<TResponse>,
+    req_template: SenderRequestTemplate<TResponse>,
+    in_flight: BulkSenderInFlight<TResponse>,
+    timeout: Timeout,
+    body: SenderBody,
+    _marker: PhantomData<TDocument>,
+}
+
+impl<TDocument, TResponse> BulkSender<TDocument, TResponse> {
+    pub(super) fn new(req_template: SenderRequestTemplate<TResponse>, timeout: Timeout, body: SenderBody) -> (Self, BulkReceiver<TResponse>) {
+        let (tx, rx) = channel::bounded(1);
+
+        let sender = BulkSender {
+            tx: BulkSenderInner(Some(tx)),
+            req_template,
+            timeout,
+            body,
+            in_flight: BulkSenderInFlight::ReadyToSend,
+            _marker: PhantomData,
+        };
+
+        (sender, BulkReceiver { rx: BulkReceiverInner(rx) })
+    }
+}
+
+pub(super) struct SenderRequestTemplate<TResponse> {
+    client: Client<AsyncSender>,
+    params: RequestParams,
+    index: Option<Index<'static>>,
+    ty: Option<Type<'static>>,
+    _marker: PhantomData<TResponse>,
+}
+
+impl<TResponse> SenderRequestTemplate<TResponse> {
+    pub(super) fn new(client: Client<AsyncSender>, params: RequestParams, index: Option<Index<'static>>, ty: Option<Type<'static>>) -> Self {
+        SenderRequestTemplate {
+            client,
+            params,
+            index,
+            ty,
+            _marker: PhantomData,
+        }
+    }
+
+    fn to_request(&self, body: Vec<u8>) -> BulkRequestBuilder<AsyncSender, Vec<u8>, TResponse> {
+        RequestBuilder::new(
+            self.client.clone(),
+            FluentBuilder::new().value(self.params.clone()),
+            BulkRequestInner::<Vec<u8>, TResponse> {
+                index: self.index.clone(),
+                ty: self.ty.clone(),
+                body: WrappedBody::new(body),
+                _marker: PhantomData,
+            }
+        )
+    }
+}
+
+pub(super) struct Timeout {
+    timer: Timer,
+    duration: Duration,
+    sleep: Sleep,
+}
+
+impl Timeout {
+    pub(super) fn new(duration: Duration) -> Self {
+        let timer = Timer::default();
+        let sleep = timer.sleep(duration);
+
+        Timeout {
+            duration,
+            timer,
+            sleep
+        }
+    }
+
+    fn restart(&mut self) {
+        self.sleep = self.timer.sleep(self.duration);
+    }
+}
+
+impl Future for Timeout {
+    type Item = ();
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        self.sleep.poll().map_err(error::request)
+    }
+}
+
+/**
+The current state of the `BulkSender`.
+
+The `BulkSender` and `BulkBody` combination means operations can be pushed while a single request is in-flight.
+*/
+enum BulkSenderInFlight<TResponse> {
+    ReadyToSend,
+    Pending(Pending<TResponse>),
+    Transmitting(Option<TResponse>),
+    Transmitted,
+}
+
+struct BulkSenderInner<T>(Option<channel::Sender<T>>);
+struct BulkReceiverInner<T>(channel::Receiver<T>);
+
+/**
+The receiving half of a stream of bulk operations.
+
+The receiver emits complete bulk responses.
+*/
+pub struct BulkReceiver<TResponse> {
+    rx: BulkReceiverInner<TResponse>
+}
+
+pub(super) struct SenderBody {
+    scratch: Vec<u8>,
+    body: BytesMut,
+    size: usize,
+}
+
+impl SenderBody {
+    pub(super) fn new(size: usize) -> Self {
+        SenderBody {
+            scratch: Vec::new(),
+            size,
+            body: BytesMut::with_capacity(size),
+        }
+    }
+
+    fn take(&mut self) -> BytesMut {
+        // Make sure any oversize remaining scratch can be copied to the new buffer
+        let size = usize::max(self.scratch.len(), self.size);
+        let mut new_body = BytesMut::with_capacity(size);
+
+        // Copy out any scratch into the new buffer
+        // This would probably be a single operation that didn't fit
+        if self.scratch.len() > 0 {
+            new_body.put_slice(&self.scratch);
+            self.scratch.clear();
+        }
+
+        mem::replace(&mut self.body, new_body)
+    }
+
+    fn has_capacity(&self) -> bool {
+        self.scratch.len() == 0 && self.body.remaining_mut() > 0
+    }
+
+    fn is_empty(&self) -> bool {
+        self.body.len() == 0
+    }
+
+    fn is_full(&self) -> bool {
+        self.scratch.len() > 0
+    }
+
+    fn push<TDocument>(&mut self, op: BulkOperation<TDocument>) -> Result<(), io::Error>
+    where
+        TDocument: Serialize,
+    {
+        op.write(&mut self.scratch)?;
+
+        // Copy the scratch buffer into the request buffer if it fits
+        if self.scratch.len() < self.body.remaining_mut() {
+            self.body.put_slice(&self.scratch);
+            self.scratch.clear();
+
+            Ok(())
+        }
+        // If the body is empty and the buffer doesn't fit, replace the current body buffer
+        else if self.body.len() == 0 {
+            let scratch = mem::replace(&mut self.scratch, Vec::new());
+            self.body = BytesMut::from(scratch);
+
+            Ok(())
+        }
+        // If the buffer doesn't fit, then retain it for the next request
+        else {
+            Ok(())
+        }
+    }
+}
+
+impl<TDocument, TResponse> Sink for BulkSender<TDocument, TResponse>
+where
+    TDocument: Serialize + Send + 'static,
+    TResponse: DeserializeOwned + IsOk + Send + 'static,
+{
+    type SinkItem = BulkOperation<TDocument>;
+    type SinkError = Error;
+
+    fn start_send(&mut self, item: Self::SinkItem) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {
+        match self.timeout.poll() {
+            // Only respect the timeout if the body is not empty
+            Ok(Async::Ready(())) if !self.body.is_empty() => {
+                return match self.poll_complete() {
+                    Ok(_) => Ok(AsyncSink::NotReady(item)),
+                    Err(e) => Err(e)
+                }
+            },
+            // Continue
+            Ok(Async::Ready(_)) | Ok(Async::NotReady) => (),
+            Err(e) => return Err(error::request(e)),
+        }
+
+        if self.body.has_capacity() {
+            self.body.push(item).map_err(error::request)?;
+            Ok(AsyncSink::Ready)
+        }
+        else {
+            match self.poll_complete() {
+                Ok(_) => Ok(AsyncSink::NotReady(item)),
+                Err(e) => Err(e)
+            }
+        }
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        let in_flight = match self.in_flight {
+            // The `Sender` is ready to send another request
+            BulkSenderInFlight::ReadyToSend => {
+                match self.timeout.poll() {
+                    // If the timeout hasn't expired and the body isn't full then we're not ready
+                    Ok(Async::NotReady) if !self.body.is_full() && !self.body.is_empty() => return Ok(Async::NotReady),
+                    // Continue
+                    Ok(Async::NotReady) => (),
+                    // Restart the expired timer
+                    Ok(Async::Ready(())) => self.timeout.restart(),
+                    Err(e) => return Err(error::request(e)),
+                }
+
+                if self.body.is_empty() {
+                    return Ok(Async::Ready(()))
+                }
+
+                let body = self.body.take();
+
+                let req = self.req_template.to_request(body.to_vec());
+                let pending = req.send();
+
+                BulkSenderInFlight::Pending(pending)
+            }
+            // A request is pending
+            BulkSenderInFlight::Pending(ref mut pending) => {
+                let response = try_ready!(pending.poll());
+                BulkSenderInFlight::Transmitting(Some(response))
+            }
+            // A response is transmitting
+            BulkSenderInFlight::Transmitting(ref mut response) => {
+                if let Some(item) = response.take() {
+                    match self.tx.start_send(item) {
+                        Ok(AsyncSink::Ready) => BulkSenderInFlight::Transmitted,
+                        Ok(AsyncSink::NotReady(item)) => {
+                            *response = Some(item);
+                            return Ok(Async::NotReady)
+                        },
+                        Err(e) => return Err(e)
+                    }
+                }
+                else {
+                    BulkSenderInFlight::Transmitted
+                }
+            }
+            // The request has completed
+            BulkSenderInFlight::Transmitted => {
+                let _ = try_ready!(self.tx.poll_complete());
+                BulkSenderInFlight::ReadyToSend
+            }
+        };
+
+        self.in_flight = in_flight;
+        self.poll_complete()
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        let _ = try_ready!(self.poll_complete());
+        self.tx.close()
+    }
+}
+
+impl<T> Sink for BulkSenderInner<T>
+where
+    T: Send,
+{
+    type SinkItem = T;
+    type SinkError = Error;
+
+    fn start_send(&mut self, item: Self::SinkItem) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {
+        self.0.as_ref().map(|tx| match tx.try_send(item) {
+            Ok(()) => Ok(AsyncSink::Ready),
+            Err(TrySendError::Full(item)) => Ok(AsyncSink::NotReady(item)),
+            Err(TrySendError::Disconnected(_)) => Err(error::request(Disconnected)),
+        })
+        .unwrap_or(Err(error::request(Disconnected)))
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(Async::Ready(()))
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.0 = None;
+        Ok(Async::Ready(()))
+    }
+}
+
+impl<TResponse> Stream for BulkReceiver<TResponse>
+where
+    TResponse: Send
+{
+    type Item = TResponse;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.rx.poll()
+    }
+}
+
+impl<T> Stream for BulkReceiverInner<T>
+where
+    T: Send
+{
+    type Item = T;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        match self.0.try_recv() {
+            Ok(item) => Ok(Async::Ready(Some(item))),
+            Err(TryRecvError::Empty) => Ok(Async::NotReady),
+            // If the channel is disconnected, then we're finished processing
+            Err(TryRecvError::Disconnected) => Ok(Async::Ready(None)),
+        }
+    }
+}
+
+/**
+Alternative disconnected error because `TrySendError` and `TryReceiveError` don't implement `Error`.
+*/
+#[derive(Debug)]
+struct Disconnected;
+
+impl fmt::Display for Disconnected {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.write_str("disconnected")
+    }
+}
+
+impl StdError for Disconnected {
+    fn description(&self) -> &str {
+        "disconnected"
+    }
+}
diff --git a/src/elastic/src/client/requests/common.rs b/src/elastic/src/client/requests/common.rs
new file mode 100644
index 0000000000..4d9dfa5762
--- /dev/null
+++ b/src/elastic/src/client/requests/common.rs
@@ -0,0 +1,157 @@
+/*!
+Types that are common between requests.
+*/
+
+use serde::ser::{Serialize, Serializer};
+use serde_json::{Value, Map};
+
+/** Update an indexed document using a new document. */
+#[derive(Serialize)]
+pub struct Doc<TDocument> {
+    doc: DocInner<TDocument>,
+}
+
+impl<TDocument> Doc<TDocument> {
+    pub(crate) fn empty() -> Self {
+        Doc {
+            doc: DocInner { inner: None },
+        }
+    }
+
+    pub(crate) fn value(doc: TDocument) -> Self {
+        Doc {
+            doc: DocInner { inner: Some(doc) },
+        }
+    }
+}
+
+struct DocInner<TDocument> {
+    inner: Option<TDocument>,
+}
+
+impl<TDocument> Serialize for DocInner<TDocument>
+where
+    TDocument: Serialize,
+{
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match self.inner {
+            Some(ref doc) => doc.serialize(serializer),
+            None => Value::Object(Map::new()).serialize(serializer),
+        }
+    }
+}
+
+/** A default set of script parameters. */
+pub type DefaultParams = Map<String, Value>;
+
+/** Update an indexed document using a script. */
+#[derive(Serialize)]
+pub struct Script<TParams> {
+    script: ScriptInner<TParams>,
+}
+
+impl Script<DefaultParams> {
+    /** Create a new script builder using the given source. */
+    pub(crate) fn new<TScript>(source: TScript) -> Self
+        where
+            TScript: ToString,
+    {
+        ScriptBuilder::new(source).build()
+    }
+}
+
+#[derive(Serialize)]
+struct ScriptInner<TParams> {
+    #[serde(rename = "inline")] source: String,
+    #[serde(skip_serializing_if = "Option::is_none")] lang: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")] params: Option<TParams>,
+}
+
+/** A builder for an update script that can be configured before sending. */
+pub struct ScriptBuilder<TParams> {
+    source: String,
+    lang: Option<String>,
+    params: Option<TParams>,
+}
+
+impl ScriptBuilder<DefaultParams> {
+    /** Create a new script builder using the given source. */
+    pub fn new<TScript>(source: TScript) -> Self
+    where
+        TScript: ToString,
+    {
+        ScriptBuilder {
+            source: source.to_string(),
+            params: None,
+            lang: None,
+        }
+    }
+
+    /** Set a script parameter. */
+    pub fn param<TKey, TValue>(mut self, key: TKey, value: TValue) -> Self
+    where
+        TKey: ToString,
+        TValue: Into<Value>,
+    {
+        let mut params = self.params.unwrap_or_else(DefaultParams::new);
+        params.insert(key.to_string(), value.into());
+
+        self.params = Some(params);
+        self
+    }
+}
+
+impl<TParams> ScriptBuilder<TParams> {
+    pub(crate) fn from_script(script: Script<TParams>) -> Self {
+        let script = script.script;
+
+        ScriptBuilder {
+            source: script.source,
+            lang: script.lang,
+            params: script.params,
+        }
+    }
+
+    /** Set the language for the update script. */
+    pub fn lang<TLang>(mut self, lang: Option<TLang>) -> Self
+    where
+        TLang: ToString,
+    {
+        self.lang = lang.map(|lang| lang.to_string());
+        self
+    }
+
+    /** Specify a new set of parameters for the update script. */
+    pub fn params<TNewParams>(self, params: TNewParams) -> ScriptBuilder<TNewParams> {
+        ScriptBuilder {
+            source: self.source,
+            lang: self.lang,
+            params: Some(params),
+        }
+    }
+
+    pub(crate) fn build(self) -> Script<TParams> {
+        Script {
+            script: ScriptInner {
+                source: self.source,
+                params: self.params,
+                lang: self.lang,
+            },
+        }
+    }
+}
+
+impl From<String> for ScriptBuilder<DefaultParams> {
+    fn from(source: String) -> Self {
+        ScriptBuilder::new(source)
+    }
+}
+
+impl<'a> From<&'a str> for ScriptBuilder<DefaultParams> {
+    fn from(source: &'a str) -> Self {
+        ScriptBuilder::new(source)
+    }
+}
\ No newline at end of file
diff --git a/src/elastic/src/client/requests/document_update.rs b/src/elastic/src/client/requests/document_update.rs
index 51d9684373..07f6ac3871 100644
--- a/src/elastic/src/client/requests/document_update.rs
+++ b/src/elastic/src/client/requests/document_update.rs
@@ -6,8 +6,8 @@ Builders for [update document requests][docs-update].
 
 use std::marker::PhantomData;
 use futures::{Future, Poll};
-use serde_json::{self, Map, Value};
-use serde::ser::{Serialize, Serializer};
+use serde_json;
+use serde::ser::Serialize;
 
 use error::{self, Error};
 use client::Client;
@@ -19,6 +19,8 @@ use client::requests::raw::RawRequestInner;
 use client::responses::UpdateResponse;
 use types::document::DocumentType;
 
+pub use client::requests::common::{Doc, Script, ScriptBuilder, DefaultParams};
+
 /** 
 An [update document request][docs-update] builder that can be configured before sending.
 
@@ -641,136 +643,6 @@ impl Future for Pending {
     }
 }
 
-/** Update an indexed document using a new document. */
-#[derive(Serialize)]
-pub struct Doc<TDocument> {
-    doc: DocInner<TDocument>,
-}
-
-impl<TDocument> Doc<TDocument> {
-    fn empty() -> Self {
-        Doc {
-            doc: DocInner { inner: None },
-        }
-    }
-
-    fn value(doc: TDocument) -> Self {
-        Doc {
-            doc: DocInner { inner: Some(doc) },
-        }
-    }
-}
-
-struct DocInner<TDocument> {
-    inner: Option<TDocument>,
-}
-
-impl<TDocument> Serialize for DocInner<TDocument>
-where
-    TDocument: Serialize,
-{
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        match self.inner {
-            Some(ref doc) => doc.serialize(serializer),
-            None => Value::Object(Map::new()).serialize(serializer),
-        }
-    }
-}
-
-type DefaultParams = Map<String, Value>;
-
-/** Update an indexed document using a script. */
-#[derive(Serialize)]
-pub struct Script<TParams> {
-    script: ScriptInner<TParams>,
-}
-
-#[derive(Serialize)]
-struct ScriptInner<TParams> {
-    #[serde(rename = "inline")] source: String,
-    #[serde(skip_serializing_if = "Option::is_none")] lang: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")] params: Option<TParams>,
-}
-
-/** A builder for an update script that can be configured before sending. */
-pub struct ScriptBuilder<TParams> {
-    source: String,
-    lang: Option<String>,
-    params: Option<TParams>,
-}
-
-impl ScriptBuilder<DefaultParams> {
-    /** Create a new script builder using the given source. */
-    pub fn new<TScript>(source: TScript) -> Self
-    where
-        TScript: ToString,
-    {
-        ScriptBuilder {
-            source: source.to_string(),
-            params: None,
-            lang: None,
-        }
-    }
-
-    /** Set a script parameter. */
-    pub fn param<TKey, TValue>(mut self, key: TKey, value: TValue) -> Self
-    where
-        TKey: ToString,
-        TValue: Into<Value>,
-    {
-        let mut params = self.params.unwrap_or_else(DefaultParams::new);
-        params.insert(key.to_string(), value.into());
-
-        self.params = Some(params);
-        self
-    }
-}
-
-impl<TParams> ScriptBuilder<TParams> {
-    /** Set the language for the update script. */
-    pub fn lang<TLang>(mut self, lang: Option<TLang>) -> Self
-    where
-        TLang: ToString,
-    {
-        self.lang = lang.map(|lang| lang.to_string());
-        self
-    }
-
-    /** Specify a new set of parameters for the update script. */
-    pub fn params<TNewParams>(self, params: TNewParams) -> ScriptBuilder<TNewParams> {
-        ScriptBuilder {
-            source: self.source,
-            lang: self.lang,
-            params: Some(params),
-        }
-    }
-
-    fn build(self) -> Script<TParams> {
-        Script {
-            script: ScriptInner {
-                source: self.source,
-                params: self.params,
-                lang: self.lang,
-            },
-        }
-    }
-}
-
-impl From<String> for ScriptBuilder<DefaultParams> {
-    fn from(source: String) -> Self {
-        ScriptBuilder::new(source)
-    }
-}
-
-impl<'a> From<&'a str> for ScriptBuilder<DefaultParams> {
-    fn from(source: &'a str) -> Self {
-        ScriptBuilder::new(source)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use serde_json::{self, Value};
diff --git a/src/elastic/src/client/requests/mod.rs b/src/elastic/src/client/requests/mod.rs
index b9f8f7e51f..e79d8e5d49 100644
--- a/src/elastic/src/client/requests/mod.rs
+++ b/src/elastic/src/client/requests/mod.rs
@@ -50,7 +50,11 @@ pub use self::index_exists::IndexExistsRequestBuilder;
 
 // Misc requests
 pub mod ping;
+pub mod bulk;
 pub use self::ping::PingRequestBuilder;
+pub use self::bulk::BulkRequestBuilder;
+
+pub mod common;
 
 /**
 A builder for a request.
@@ -244,6 +248,13 @@ pub mod prelude {
     pub use super::params::*;
     pub use super::endpoints::*;
 
+    pub use super::bulk::{
+        bulk_index,
+        bulk_update,
+        bulk_create,
+        bulk_delete,
+    };
+
     pub use super::{
         empty_body,
         DefaultBody,
diff --git a/src/elastic/src/client/responses/mod.rs b/src/elastic/src/client/responses/mod.rs
index d9887f66fa..cc3c747002 100644
--- a/src/elastic/src/client/responses/mod.rs
+++ b/src/elastic/src/client/responses/mod.rs
@@ -24,5 +24,6 @@ pub use elastic_responses::bulk;
 pub mod prelude {
     /*! A glob import for convenience. */
 
+    pub use super::bulk::Action as BulkAction;
     pub use super::{AsyncResponseBuilder, SyncResponseBuilder, BulkErrorsResponse, BulkResponse, CommandResponse, DeleteResponse, GetResponse, IndexResponse, IndicesExistsResponse, PingResponse, SearchResponse, Shards, UpdateResponse};
 }
diff --git a/src/elastic/src/lib.rs b/src/elastic/src/lib.rs
index e23cf9fe22..90f2c44f81 100644
--- a/src/elastic/src/lib.rs
+++ b/src/elastic/src/lib.rs
@@ -286,6 +286,7 @@ extern crate elastic_types;
 #[macro_use]
 extern crate error_chain;
 extern crate fluent_builder;
+#[macro_use]
 extern crate futures;
 extern crate futures_cpupool;
 #[macro_use]
@@ -293,6 +294,7 @@ extern crate log;
 #[macro_use]
 extern crate quick_error;
 extern crate reqwest;
+extern crate crossbeam_channel as channel;
 extern crate serde;
 #[macro_use]
 extern crate serde_derive;
@@ -301,6 +303,7 @@ extern crate serde_json;
 extern crate tokio_core;
 extern crate tokio_io;
 extern crate url;
+extern crate tokio_timer;
 extern crate uuid;
 
 pub mod error;
diff --git a/src/responses/src/bulk.rs b/src/responses/src/bulk.rs
index 4056dd6c6f..c8aa5842e1 100644
--- a/src/responses/src/bulk.rs
+++ b/src/responses/src/bulk.rs
@@ -508,7 +508,7 @@ where
 }
 
 /** The bulk action being performed. */
-#[derive(Deserialize, Debug, Clone, Copy)]
+#[derive(Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
 pub enum Action {
     #[serde(rename = "index")] Index,
     #[serde(rename = "create")] Create,


Test patch:
diff --git a/tests/run/src/bulk/delete.rs b/tests/run/src/bulk/delete.rs
new file mode 100644
index 0000000000..dded190b3a
--- /dev/null
+++ b/tests/run/src/bulk/delete.rs
@@ -0,0 +1,62 @@
+use futures::Future;
+use elastic::prelude::*;
+use elastic::error::Error;
+use run_tests::IntegrationTest;
+
+#[derive(Debug, Clone, Copy)]
+pub struct Delete;
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, ElasticType)]
+pub struct Doc {
+    id: i32,
+}
+
+const INDEX: &'static str = "bulk_delete";
+const ID: i32 = 1;
+
+impl IntegrationTest for Delete {
+    type Response = BulkResponse;
+
+    fn kind() -> &'static str {
+        "bulk"
+    }
+    fn name() -> &'static str {
+        "delete"
+    }
+
+    // Ensure the index doesn't exist
+    fn prepare(&self, client: AsyncClient) -> Box<Future<Item = (), Error = Error>> {
+        let delete_res = client.index_delete(index(INDEX)).send().map(|_| ());
+
+        Box::new(delete_res)
+    }
+
+    // Index a document, get it, delete it, then try get it again
+    fn request(&self, client: AsyncClient) -> Box<Future<Item = Self::Response, Error = Error>> {
+        let index_res = client
+            .document_index(index(INDEX), Doc { id: ID })
+            .id(ID)
+            .params_fluent(|p| p.url_param("refresh", true))
+            .send();
+
+        let delete_res = client
+            .bulk()
+            .push(bulk_delete()
+                .index(INDEX)
+                .ty(Doc::name())
+                .id(ID))
+            .send();
+
+        Box::new(
+            index_res
+                .and_then(|_| delete_res)
+        )
+    }
+
+    // Ensure the document was found before deleting but not found after deleting
+    fn assert_ok(&self, res: &Self::Response) -> bool {
+        let deleted = res.iter().next().unwrap().unwrap();
+
+        deleted.action() == BulkAction::Delete && deleted.found()
+    }
+}
diff --git a/tests/run/src/bulk/index_create.rs b/tests/run/src/bulk/index_create.rs
new file mode 100644
index 0000000000..d0b341b354
--- /dev/null
+++ b/tests/run/src/bulk/index_create.rs
@@ -0,0 +1,63 @@
+use futures::Future;
+use elastic::prelude::*;
+use elastic::error::Error;
+use run_tests::IntegrationTest;
+
+#[derive(Debug, Clone, Copy)]
+pub struct IndexCreate;
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, ElasticType)]
+pub struct Doc {
+    id: i32,
+    title: String,
+    timestamp: Date<DefaultDateMapping>,
+}
+
+const INDEX: &'static str = "bulk_index_create";
+const ID: i32 = 1;
+
+fn doc() -> Doc {
+    Doc {
+        id: ID,
+        title: "A document title".to_owned(),
+        timestamp: Date::build(2017, 03, 24, 13, 44, 0, 0),
+    }
+}
+
+impl IntegrationTest for IndexCreate {
+    type Response = BulkResponse;
+
+    fn kind() -> &'static str {
+        "bulk"
+    }
+    fn name() -> &'static str {
+        "index_create"
+    }
+
+    // Ensure the index doesn't exist
+    fn prepare(&self, client: AsyncClient) -> Box<Future<Item = (), Error = Error>> {
+        let delete_res = client.index_delete(index(INDEX)).send().map(|_| ());
+
+        Box::new(delete_res)
+    }
+
+    // Index a document, then get it
+    fn request(&self, client: AsyncClient) -> Box<Future<Item = Self::Response, Error = Error>> {
+        let bulk_res = client
+            .bulk()
+            .push(bulk_create(doc())
+                .index(INDEX)
+                .ty(Doc::name())
+                .id(ID))
+            .send();
+
+        Box::new(bulk_res)
+    }
+
+    // Ensure the response contains the expected document
+    fn assert_ok(&self, res: &Self::Response) -> bool {
+        let created = res.iter().next().unwrap().unwrap();
+
+        created.action() == BulkAction::Create && created.created()
+    }
+}
diff --git a/tests/run/src/bulk/mod.rs b/tests/run/src/bulk/mod.rs
new file mode 100644
index 0000000000..c247d4b53d
--- /dev/null
+++ b/tests/run/src/bulk/mod.rs
@@ -0,0 +1,11 @@
+use run_tests::{test, Test};
+
+mod delete;
+mod index_create;
+
+pub fn tests() -> Vec<Test> {
+    vec![
+        Box::new(|client| test(client, delete::Delete)),
+        Box::new(|client| test(client, index_create::IndexCreate)),
+    ]
+}
diff --git a/tests/run/src/main.rs b/tests/run/src/main.rs
index d87167742f..0e16562bc2 100644
--- a/tests/run/src/main.rs
+++ b/tests/run/src/main.rs
@@ -27,6 +27,7 @@ use term_painter::ToStyle;
 use term_painter::Color::*;
 use clap::{App, Arg};
 
+mod bulk;
 mod document;
 mod search;
 mod index;
diff --git a/tests/run/src/run_tests.rs b/tests/run/src/run_tests.rs
index f02af98230..138a0439e0 100644
--- a/tests/run/src/run_tests.rs
+++ b/tests/run/src/run_tests.rs
@@ -87,14 +87,17 @@ pub fn call(client: AsyncClient, max_concurrent_tests: usize) -> Box<Future<Item
     use search;
     use document;
     use index;
+    use bulk;
 
     let document_tests = document::tests().into_iter();
     let search_tests = search::tests().into_iter();
     let index_tests = index::tests().into_iter();
+    let bulk_tests = bulk::tests().into_iter();
 
     let all_tests = document_tests
         .chain(search_tests)
         .chain(index_tests)
+        .chain(bulk_tests)
         .map(move |t| t(client.clone()));
 
     let test_stream = stream::futures_unordered(all_tests)