Merge lp:~djfroofy/txaws/newagent-767205 into lp:txaws

[Drew Smathers]

[Robert Collins (lifeless)]

review: Abstain

[Duncan McGreggor (oubiwann)]

This change looks good to me at first glance (reading the diff on the merge proposal; I haven't branched the code yet). In particular, the amz_headers functionality is much-needed.

Drew, I don't see anything about the new web client usage, only headers and body producer. Is that coming in a future change?

At the risk of being a total pain in the ass, here's what I would recommend:

1) split out the headers changes into a new branch, and attach it to bug #972432.
2) create a new bug for the body producer/receiver factory feature, split that code out, and attach it to there in its own branch.
3) do the work on the new agent in this branch and this ticket (bug #767205).

review: Needs Fixing

[Daira Hopwood (daira)]

Thanks, will review tomorrow.

[Daira Hopwood (daira)]

Sorry for the delay. Looks good to me.

Just a minor style nit: some of the code indents parameters to after the '(', and some only by 4 spaces. E.g.
{{{
def __init__(self, creds=None, endpoint=None, query_factory=None,
             parser=None, receiver_factory=None):
}}}
vs
{{{
def __init__(self, creds=None, endpoint=None, query_factory=None,
    receiver_factory=None):
}}}
The former is more readable.

[Zooko Wilcox-O'Hearn (zooko)]

Okay, if david-sarah reviewed it and said "Looks good to me", then the branch ought to be merged. Who has authority do that?

[Duncan McGreggor (oubiwann)]

On Thu, Apr 12, 2012 at 5:02 PM, Zooko Wilcox-O'Hearn <email address hidden> wrote:
> Okay, if david-sarah reviewed it and said "Looks good to me", then the branch ought to be merged. Who has authority do that?

I'll take a look...

d

[Duncan McGreggor (oubiwann)]

On Fri, Apr 13, 2012 at 12:43 AM, Duncan McGreggor <email address hidden> wrote:
> On Thu, Apr 12, 2012 at 5:02 PM, Zooko Wilcox-O'Hearn <email address hidden> wrote:
>> Okay, if david-sarah reviewed it and said "Looks good to me", then the branch ought to be merged. Who has authority do that?
>
> I'll take a look...

Okay, I just took a look. Looks like a review comment I left earlier
(needs fixing) hasn't been addressed by the author (unless I'm missing
something?).

Once that's either done or discussed, we can move this puppy forward :-)

[Drew Smathers (djfroofy)]

> This change looks good to me at first glance (reading the diff on the merge
> proposal; I haven't branched the code yet). In particular, the amz_headers
> functionality is much-needed.
>
> Drew, I don't see anything about the new web client usage, only headers and
> body producer. Is that coming in a future change?
>

Ok, so I had to remember myself while staring at the diff. The code for agent is in the prerequisite branch: lp:~djfroofy/txaws/modernize-924459

> At the risk of being a total pain in the ass, here's what I would recommend:
>

No, no worries.

> 1) split out the headers changes into a new branch, and attach it to bug
> #972432.

#972432 must be a duplicate of #640098; the amz_headers argument has been merged already. To summarize what this diff is adding: Support for plugging in Body Producers and Receivers in s3.client (this is made possible by: lp:~djfroofy/txaws/modernize-924459).

There are actually no changes related to headers in this patch. Rather, I'm adding receiver_factory as __init__() argument to S3Client and body_producer as argument to regular S3 api calls.

Cutting to the point, I think #972432 needs to be reviewed first.

> 2) create a new bug for the body producer/receiver factory feature, split that
> code out, and attach it to there in its own branch.

That was in the scope of #972432

> 3) do the work on the new agent in this branch and this ticket (bug #767205).

Again: #972432 is where that functionality is.

[Drew Smathers (djfroofy)]

> Sorry for the delay. Looks good to me.
>
> Just a minor style nit: some of the code indents parameters to after the '(',
> and some only by 4 spaces. E.g.
> {{{
> def __init__(self, creds=None, endpoint=None, query_factory=None,
> parser=None, receiver_factory=None):
> }}}
> vs
> {{{
> def __init__(self, creds=None, endpoint=None, query_factory=None,
> receiver_factory=None):
> }}}
> The former is more readable.

Yes, I agree and will address this.

[lp:~djfroofy/txaws/newagent-767205]

144. By Drew Smathers on 2012-04-14

param indentation style fix [f=767205]

[Drew Smathers (djfroofy)]

Ok style fixes pushed to branch.

review: Needs Resubmitting

[Duncan McGreggor (oubiwann)]

On Sat, Apr 14, 2012 at 12:28 PM, Drew Smathers
<email address hidden> wrote:
>> This change looks good to me at first glance (reading the diff on the merge
>> proposal; I haven't branched the code yet). In particular, the amz_headers
>> functionality is much-needed.
>>
>> Drew, I don't see anything about the new web client usage, only headers and
>> body producer. Is that coming in a future change?
>>
>
> Ok, so I had to remember myself while staring at the diff. The code for agent is in the prerequisite branch: lp:~djfroofy/txaws/modernize-924459
>
>
>> At the risk of being a total pain in the ass, here's what I would recommend:
>>
>
> No, no worries.
>
>> 1) split out the headers changes into a new branch, and attach it to bug
>> #972432.
>
> #972432 must be a duplicate of #640098; the amz_headers argument has been merged already.  To summarize what this diff is adding: Support for plugging in Body Producers and Receivers in s3.client (this is made possible by: lp:~djfroofy/txaws/modernize-924459).
>
> There are actually no changes related to headers in this patch. Rather, I'm adding receiver_factory as __init__() argument to S3Client and body_producer as argument to regular S3 api calls.
>
> Cutting to the point, I think #972432 needs to be reviewed first.
>
>> 2) create a new bug for the body producer/receiver factory feature, split that
>> code out, and attach it to there in its own branch.
>
> That was in the scope of #972432
>
>> 3) do the work on the new agent in this branch and this ticket (bug #767205).
>
> Again: #972432 is where that functionality is.
> --
> https://code.launchpad.net/~djfroofy/txaws/newagent-767205/+merge/93312
> You are reviewing the proposed merge of lp:~djfroofy/txaws/newagent-767205 into lp:txaws.

Thanks for catching me up, Drew -- this is good stuff.

I'm going to be in the air tomorrow, on the way to the OpenStack
design summit and conference, but will branch all the code and check
out the associated tickets, to make sure I haven't missed anything.

Regardless, it looks like we've got a pending merge on our hands. This
code should land in trunk within a day or two...

Thanks again!

[Duncan McGreggor (oubiwann)]

On Sat, Apr 14, 2012 at 12:28 PM, Drew Smathers
<email address hidden> wrote:
>>
>> 1) split out the headers changes into a new branch, and attach it to bug
>> #972432.
>
> #972432 must be a duplicate of #640098; the amz_headers argument has been merged already.  To summarize what this diff is adding: Support for plugging in Body Producers and Receivers in s3.client (this is made possible by: lp:~djfroofy/txaws/modernize-924459).
>
> There are actually no changes related to headers in this patch. Rather, I'm adding receiver_factory as __init__() argument to S3Client and body_producer as argument to regular S3 api calls.
>
> Cutting to the point, I think #972432 needs to be reviewed first.

Do you mean bug #924459? (since bug #972432 is a duplicate of bug #640098 ...)

>> 2) create a new bug for the body producer/receiver factory feature, split that
>> code out, and attach it to there in its own branch.
>
> That was in the scope of #972432

It looks like you mean bug #640098?

More soon...

[Drew Smathers (djfroofy)]

> On Sat, Apr 14, 2012 at 12:28 PM, Drew Smathers
> <email address hidden> wrote:
> >>
> >> 1) split out the headers changes into a new branch, and attach it to bug
> >> #972432.
> >
> > #972432 must be a duplicate of #640098; the amz_headers argument has been
> merged already.  To summarize what this diff is adding: Support for plugging
> in Body Producers and Receivers in s3.client (this is made possible by:
> lp:~djfroofy/txaws/modernize-924459).
> >
> > There are actually no changes related to headers in this patch. Rather, I'm
> adding receiver_factory as __init__() argument to S3Client and body_producer
> as argument to regular S3 api calls.
> >
> > Cutting to the point, I think #972432 needs to be reviewed first.
>
> Do you mean bug #924459? (since bug #972432 is a duplicate of bug #640098 ...)
>
> >> 2) create a new bug for the body producer/receiver factory feature, split
> that
> >> code out, and attach it to there in its own branch.
> >
> > That was in the scope of #972432
>
> It looks like you mean bug #640098?

No, another typo, but I meant bug #924459. This has underlying support for the producer/receiver factory supports. This current ticket then just exposes that underlying support in s3 APIs.

[Drew Smathers (djfroofy)]

> On Sat, Apr 14, 2012 at 12:28 PM, Drew Smathers
> <email address hidden> wrote:
> >>
> >> 1) split out the headers changes into a new branch, and attach it to bug
> >> #972432.
> >
> > #972432 must be a duplicate of #640098; the amz_headers argument has been
> merged already.  To summarize what this diff is adding: Support for plugging
> in Body Producers and Receivers in s3.client (this is made possible by:
> lp:~djfroofy/txaws/modernize-924459).
> >
> > There are actually no changes related to headers in this patch. Rather, I'm
> adding receiver_factory as __init__() argument to S3Client and body_producer
> as argument to regular S3 api calls.
> >
> > Cutting to the point, I think #972432 needs to be reviewed first.
>
> Do you mean bug #924459? (since bug #972432 is a duplicate of bug #640098 ...)
>

Yes, this was a typo.

[Duncan McGreggor (oubiwann)]

Merge away!

review: Approve

[Drew Smathers (djfroofy)]

> Merge away!

Hi Duncan, am I the one who is supposed to do the merge, or does this need to be done by another developer?

[Duncan McGreggor (oubiwann)]

On Fri, May 11, 2012 at 1:19 PM, Drew Smathers
<email address hidden> wrote:
>> Merge away!
>
> Hi Duncan, am I the one who is supposed to do the merge, or does this need to be done by another developer?

Yes! I've just added you to the team -- given your contributions (and
god-like patience!), I think that's more than reasonable :-)

You'll want to do the obvious (merge trunk into your branches, check
for conflicts, push, etc.), but you should be all set!

Let us know if you run into any problems.

d

=== modified file 'txaws/client/base.py'
--- txaws/client/base.py	2012-04-14 16:36:18 +0000
+++ txaws/client/base.py	2012-04-14 16:36:18 +0000
@@ -74,9 +74,10 @@
     @param query_factory: The class or function that produces a query
         object for making requests to the EC2 service.
     @param parser: A parser object for parsing responses from the EC2 service.
+    @param receiver_factory: Factory for receiving responses from EC2 service.
     """
     def __init__(self, creds=None, endpoint=None, query_factory=None,
-                 parser=None):
+                 parser=None, receiver_factory=None):
         if creds is None:
             creds = AWSCredentials()
         if endpoint is None:
@@ -84,6 +85,7 @@
         self.creds = creds
         self.endpoint = endpoint
         self.query_factory = query_factory
+        self.receiver_factory = receiver_factory
         self.parser = parser
 
 class StreamingError(Exception):

=== modified file 'txaws/s3/client.py'
--- txaws/s3/client.py	2012-01-28 00:39:00 +0000
+++ txaws/s3/client.py	2012-04-14 16:36:18 +0000
@@ -74,10 +74,12 @@
 class S3Client(BaseClient):
     """A client for S3."""
 
-    def __init__(self, creds=None, endpoint=None, query_factory=None):
+    def __init__(self, creds=None, endpoint=None, query_factory=None,
+                 receiver_factory=None):
         if query_factory is None:
             query_factory = Query
-        super(S3Client, self).__init__(creds, endpoint, query_factory)
+        super(S3Client, self).__init__(creds, endpoint, query_factory,
+                                       receiver_factory=receiver_factory)
 
     def list_buckets(self):
         """
@@ -87,7 +89,8 @@
         the request.
         """
         query = self.query_factory(
-            action="GET", creds=self.creds, endpoint=self.endpoint)
+            action="GET", creds=self.creds, endpoint=self.endpoint,
+            receiver_factory=self.receiver_factory)
         d = query.submit()
         return d.addCallback(self._parse_list_buckets)
 
@@ -131,7 +134,7 @@
         """
         query = self.query_factory(
             action="GET", creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket)
+            bucket=bucket, receiver_factory=self.receiver_factory)
         d = query.submit()
         return d.addCallback(self._parse_get_bucket)
 
@@ -174,7 +177,8 @@
         """
         query = self.query_factory(action="GET", creds=self.creds,
                                    endpoint=self.endpoint, bucket=bucket,
-                                   object_name="?location")
+                                   object_name="?location",
+                                   receiver_factory=self.receiver_factory)
         d = query.submit()
         return d.addCallback(self._parse_bucket_location)
 
@@ -193,7 +197,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?lifecycle')
+            bucket=bucket, object_name='?lifecycle',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_lifecycle_config)
 
     def _parse_lifecycle_config(self, xml_bytes):
@@ -221,7 +226,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?website')
+            bucket=bucket, object_name='?website',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_website_config)
 
     def _parse_website_config(self, xml_bytes):
@@ -242,7 +248,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?notification')
+            bucket=bucket, object_name='?notification',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_notification_config)
 
     def _parse_notification_config(self, xml_bytes):
@@ -262,7 +269,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?versioning')
+            bucket=bucket, object_name='?versioning',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_versioning_config)
 
     def _parse_versioning_config(self, xml_bytes):
@@ -279,7 +287,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?acl')
+            bucket=bucket, object_name='?acl',
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_acl)
 
     def put_bucket_acl(self, bucket, access_control_policy):
@@ -289,7 +298,8 @@
         data = access_control_policy.to_xml()
         query = self.query_factory(
             action='PUT', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='?acl', data=data)
+            bucket=bucket, object_name='?acl', data=data,
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_acl)
 
     def _parse_acl(self, xml_bytes):
@@ -299,8 +309,8 @@
         """
         return AccessControlPolicy.from_xml(xml_bytes)
 
-    def put_object(self, bucket, object_name, data, content_type=None,
-                   metadata={}, amz_headers={}):
+    def put_object(self, bucket, object_name, data=None, content_type=None,
+                   metadata={}, amz_headers={}, body_producer=None):
         """
         Put an object in a bucket.
 
@@ -318,7 +328,8 @@
             action="PUT", creds=self.creds, endpoint=self.endpoint,
             bucket=bucket, object_name=object_name, data=data,
             content_type=content_type, metadata=metadata,
-            amz_headers=amz_headers)
+            amz_headers=amz_headers, body_producer=body_producer,
+            receiver_factory=self.receiver_factory)
         return query.submit()
 
     def copy_object(self, source_bucket, source_object_name, dest_bucket=None,
@@ -344,7 +355,8 @@
         query = self.query_factory(
             action="PUT", creds=self.creds, endpoint=self.endpoint,
             bucket=dest_bucket, object_name=dest_object_name,
-            metadata=metadata, amz_headers=amz_headers)
+            metadata=metadata, amz_headers=amz_headers,
+            receiver_factory=self.receiver_factory)
         return query.submit()
 
     def get_object(self, bucket, object_name):
@@ -353,7 +365,8 @@
         """
         query = self.query_factory(
             action="GET", creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name=object_name)
+            bucket=bucket, object_name=object_name,
+            receiver_factory=self.receiver_factory)
         return query.submit()
 
     def head_object(self, bucket, object_name):
@@ -384,7 +397,8 @@
         data = access_control_policy.to_xml()
         query = self.query_factory(
             action='PUT', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='%s?acl' % object_name, data=data)
+            bucket=bucket, object_name='%s?acl' % object_name, data=data,
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_acl)
 
     def get_object_acl(self, bucket, object_name):
@@ -393,7 +407,8 @@
         """
         query = self.query_factory(
             action='GET', creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name='%s?acl' % object_name)
+            bucket=bucket, object_name='%s?acl' % object_name,
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_acl)
 
     def put_request_payment(self, bucket, payer):
@@ -407,7 +422,8 @@
         data = RequestPayment(payer).to_xml()
         query = self.query_factory(
             action="PUT", creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name="?requestPayment", data=data)
+            bucket=bucket, object_name="?requestPayment", data=data,
+            receiver_factory=self.receiver_factory)
         return query.submit()
 
     def get_request_payment(self, bucket):
@@ -419,7 +435,8 @@
         """
         query = self.query_factory(
             action="GET", creds=self.creds, endpoint=self.endpoint,
-            bucket=bucket, object_name="?requestPayment")
+            bucket=bucket, object_name="?requestPayment",
+            receiver_factory=self.receiver_factory)
         return query.submit().addCallback(self._parse_get_request_payment)
 
     def _parse_get_request_payment(self, xml_bytes):
@@ -434,12 +451,13 @@
     """A query for submission to the S3 service."""
 
     def __init__(self, bucket=None, object_name=None, data="",
-                 content_type=None, metadata={}, amz_headers={}, *args,
-                 **kwargs):
+                 content_type=None, metadata={}, amz_headers={},
+                 body_producer=None, *args, **kwargs):
         super(Query, self).__init__(*args, **kwargs)
         self.bucket = bucket
         self.object_name = object_name
         self.data = data
+        self.body_producer = body_producer
         self.content_type = content_type
         self.metadata = metadata
         self.amz_headers = amz_headers
@@ -463,9 +481,14 @@
         """
         Build the list of headers needed in order to perform S3 operations.
         """
-        headers = {"Content-Length": len(self.data),
-                   "Content-MD5": calculate_md5(self.data),
+        if self.body_producer:
+            content_length = self.body_producer.length
+        else:
+            content_length = len(self.data)
+        headers = {"Content-Length": content_length,
                    "Date": self.date}
+        if self.body_producer is None:
+            headers["Content-MD5"] = calculate_md5(self.data)
         for key, value in self.metadata.iteritems():
             headers["x-amz-meta-" + key] = value
         for key, value in self.amz_headers.iteritems():
@@ -529,5 +552,6 @@
                 self.endpoint, self.bucket, self.object_name)
         d = self.get_page(
             url_context.get_url(), method=self.action, postdata=self.data,
-            headers=self.get_headers())
+            headers=self.get_headers(), body_producer=self.body_producer,
+            receiver_factory=self.receiver_factory)
         return d.addErrback(s3_error_wrapper)

=== modified file 'txaws/s3/tests/test_client.py'
--- txaws/s3/tests/test_client.py	2012-01-28 00:44:53 +0000
+++ txaws/s3/tests/test_client.py	2012-04-14 16:36:18 +0000
@@ -10,6 +10,7 @@
     s3clientSkip = None
 from txaws.s3.acls import AccessControlPolicy
 from txaws.s3.model import RequestPayment
+from txaws.testing.producers import StringBodyProducer
 from txaws.service import AWSServiceEndpoint
 from txaws.testing import payload
 from txaws.testing.base import TXAWSTestCase
@@ -100,7 +101,8 @@
 
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint):
+            def __init__(query, action, creds, endpoint,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds)
                 self.assertEquals(action, "GET")
@@ -134,7 +136,8 @@
 
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint, bucket=None):
+            def __init__(query, action, creds, endpoint, bucket=None,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket)
                 self.assertEquals(action, "PUT")
@@ -156,7 +159,8 @@
 
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint, bucket=None):
+            def __init__(query, action, creds, endpoint, bucket=None,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket)
                 self.assertEquals(action, "GET")
@@ -208,7 +212,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -243,7 +248,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -284,7 +290,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -323,7 +330,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -360,7 +368,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -396,7 +405,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -433,7 +443,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -473,7 +484,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -509,7 +521,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -546,7 +559,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None):
+                         object_name=None, body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name)
@@ -576,7 +590,8 @@
 
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint, bucket=None):
+            def __init__(query, action, creds, endpoint, bucket=None,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket)
                 self.assertEquals(action, "DELETE")
@@ -599,7 +614,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None, data=""):
+                         object_name=None, data="", body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name,
@@ -630,7 +646,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None, data=""):
+                         object_name=None, data="", receiver_factory=None,
+                         body_producer=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name,
@@ -665,7 +682,7 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None):
+                metadata=None, body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -701,7 +718,7 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None):
+                metadata=None, body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -730,7 +747,8 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None, amz_headers=None):
+                metadata=None, amz_headers=None, body_producer=None,
+                receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -756,6 +774,42 @@
                              metadata={"key": "some meta data"},
                              amz_headers={"acl": "public-read"})
 
+    def test_put_object_with_custom_body_producer(self):
+
+        class StubQuery(client.Query):
+
+            def __init__(query, action, creds, endpoint, bucket=None,
+                object_name=None, data=None, content_type=None,
+                metadata=None, amz_headers=None, body_producer=None,
+                receiver_factory=None):
+                super(StubQuery, query).__init__(
+                    action=action, creds=creds, bucket=bucket,
+                    object_name=object_name, data=data,
+                    content_type=content_type, metadata=metadata,
+                    amz_headers=amz_headers, body_producer=body_producer)
+                self.assertEqual(action, "PUT")
+                self.assertEqual(creds.access_key, "foo")
+                self.assertEqual(creds.secret_key, "bar")
+                self.assertEqual(query.bucket, "mybucket")
+                self.assertEqual(query.object_name, "objectname")
+                self.assertEqual(query.content_type, "text/plain")
+                self.assertEqual(query.metadata, {"key": "some meta data"})
+                self.assertEqual(query.amz_headers, {"acl": "public-read"})
+                self.assertIdentical(body_producer, string_producer)
+
+            def submit(query):
+                return succeed(None)
+
+
+        string_producer = StringBodyProducer("some data")
+        creds = AWSCredentials("foo", "bar")
+        s3 = client.S3Client(creds, query_factory=StubQuery)
+        return s3.put_object("mybucket", "objectname",
+                             content_type="text/plain",
+                             metadata={"key": "some meta data"},
+                             amz_headers={"acl": "public-read"},
+                             body_producer=string_producer)
+
     def test_copy_object(self):
         """
         L{S3Client.copy_object} creates a L{Query} to copy an object from one
@@ -766,7 +820,8 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None, amz_headers=None):
+                metadata=None, amz_headers=None, body_producer=None,
+                receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -798,7 +853,8 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None, amz_headers=None):
+                metadata=None, amz_headers=None, body_producer=None,
+                receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -822,7 +878,7 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None):
+                metadata=None, body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -846,7 +902,7 @@
 
             def __init__(query, action, creds, endpoint, bucket=None,
                 object_name=None, data=None, content_type=None,
-                metadata=None):
+                metadata=None, body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket,
                     object_name=object_name, data=data,
@@ -869,7 +925,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None, data=""):
+                         object_name=None, data="", body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name,
@@ -902,7 +959,8 @@
         class StubQuery(client.Query):
 
             def __init__(query, action, creds, endpoint, bucket=None,
-                         object_name=None, data=""):
+                         object_name=None, data="", body_producer=None,
+                         receiver_factory=None):
                 super(StubQuery, query).__init__(action=action, creds=creds,
                                                  bucket=bucket,
                                                  object_name=object_name,
@@ -1077,7 +1135,8 @@
         """
         class StubQuery(client.Query):
 
-            def __init__(query, action, creds, endpoint, bucket):
+            def __init__(query, action, creds, endpoint, bucket,
+                body_producer=None, receiver_factory=None):
                 super(StubQuery, query).__init__(
                     action=action, creds=creds, bucket=bucket)
                 self.assertEquals(action, "GET")