basho/riak-python-client — pull request #357 by seancribbs (2014-08-07 17:59:58+00:00)
Title: Fix connection leaks when streaming

Addresses #318 and #312.

This was broken previously because iterators were returned from inside a `with` statement that ensured the connection was returned to the pool. However, because the release logic was invoked *before* returning the iterator/generator to the caller, meaning that even though it was in-use, it was eligible for reuse by other threads (or even the same thread). In pseudo-python:

```python
# in the client
def stream_something():
with get_transport() as transport:
return transport.stream_something()

# 'with' statement expanded:
def stream_something():
try:
transport = get_connection()
return transport.stream_something()
finally:
release_connection(transport)

# in the user code:
for chunk in client.stream_something():
# Here the connection is already released and the same connection
# could be reused before it's done streaming!
client.get_the_key(chunk)
```

The general strategy is to manually take connections from the pool, request the stream, then attach the pool resource to the stream so that when it closes, it can be released back to the pool. This does not explicitly solve the case where a stream generator/iterator falls out of scope and is never closed, but conscientious callers will need to either:

1. Consume the entire stream, even if discarding parts. This will invoke close logic in the iterator when `StopIteration` is raised.
2. Explicitly call `close()` on the stream when finished with it.
3. Use `contextlib.closing()` which accomplishes (2) automatically.

These options are documented on every streaming operation defined in `riak/client/operations.py`.