From 71080fc3518e2d3555f555340c3e93f3b108a5b8 Mon Sep 17 00:00:00 2001
From: Victor Stinner <victor.stinner@gmail.com>
Date: Sat, 25 Jul 2015 02:23:21 +0200
Subject: [PATCH 1/2] asyncio: Add asyncio.compat module

Move compatibility helpers for the different Python versions to a new
asyncio.compat module.
---
 Lib/asyncio/compat.py     | 17 +++++++++++++++++
 Lib/asyncio/coroutines.py |  6 ++----
 Lib/asyncio/events.py     |  7 +++----
 Lib/asyncio/futures.py    | 10 ++++------
 Lib/asyncio/locks.py      |  6 ++----
 Lib/asyncio/streams.py    |  4 ++--
 Lib/asyncio/tasks.py      |  5 ++---
 Lib/asyncio/transports.py | 10 +++-------
 8 files changed, 35 insertions(+), 30 deletions(-)
 create mode 100644 Lib/asyncio/compat.py

diff --git a/Lib/asyncio/compat.py b/Lib/asyncio/compat.py
new file mode 100644
index 00000000000..660b7e7e6c9
--- /dev/null
+++ b/Lib/asyncio/compat.py
@@ -0,0 +1,17 @@
+"""Compatibility helpers for the different Python versions."""
+
+import sys
+
+PY34 = sys.version_info >= (3, 4)
+PY35 = sys.version_info >= (3, 5)
+
+
+def flatten_list_bytes(list_of_data):
+    """Concatenate a sequence of bytes-like objects."""
+    if not PY34:
+        # On Python 3.3 and older, bytes.join() doesn't handle
+        # memoryview.
+        list_of_data = (
+            bytes(data) if isinstance(data, memoryview) else data
+            for data in list_of_data)
+    return b''.join(list_of_data)
diff --git a/Lib/asyncio/coroutines.py b/Lib/asyncio/coroutines.py
index 15475f23b41..e11b21b0975 100644
--- a/Lib/asyncio/coroutines.py
+++ b/Lib/asyncio/coroutines.py
@@ -9,14 +9,12 @@ import sys
 import traceback
 import types
 
+from . import compat
 from . import events
 from . import futures
 from .log import logger
 
 
-_PY35 = sys.version_info >= (3, 5)
-
-
 # Opcode of "yield from" instruction
 _YIELD_FROM = opcode.opmap['YIELD_FROM']
 
@@ -140,7 +138,7 @@ class CoroWrapper:
     def gi_code(self):
         return self.gen.gi_code
 
-    if _PY35:
+    if compat.PY35:
 
         __await__ = __iter__ # make compatible with 'await' expression
 
diff --git a/Lib/asyncio/events.py b/Lib/asyncio/events.py
index 496075bacf3..d5f0d451951 100644
--- a/Lib/asyncio/events.py
+++ b/Lib/asyncio/events.py
@@ -17,12 +17,11 @@ import sys
 import threading
 import traceback
 
-
-_PY34 = sys.version_info >= (3, 4)
+from asyncio import compat
 
 
 def _get_function_source(func):
-    if _PY34:
+    if compat.PY34:
         func = inspect.unwrap(func)
     elif hasattr(func, '__wrapped__'):
         func = func.__wrapped__
@@ -31,7 +30,7 @@ def _get_function_source(func):
         return (code.co_filename, code.co_firstlineno)
     if isinstance(func, functools.partial):
         return _get_function_source(func.func)
-    if _PY34 and isinstance(func, functools.partialmethod):
+    if compat.PY34 and isinstance(func, functools.partialmethod):
         return _get_function_source(func.func)
     return None
 
diff --git a/Lib/asyncio/futures.py b/Lib/asyncio/futures.py
index d06828a6204..dbe06c4a98b 100644
--- a/Lib/asyncio/futures.py
+++ b/Lib/asyncio/futures.py
@@ -11,6 +11,7 @@ import reprlib
 import sys
 import traceback
 
+from . import compat
 from . import events
 
 # States for Future.
@@ -18,9 +19,6 @@ _PENDING = 'PENDING'
 _CANCELLED = 'CANCELLED'
 _FINISHED = 'FINISHED'
 
-_PY34 = sys.version_info >= (3, 4)
-_PY35 = sys.version_info >= (3, 5)
-
 Error = concurrent.futures._base.Error
 CancelledError = concurrent.futures.CancelledError
 TimeoutError = concurrent.futures.TimeoutError
@@ -199,7 +197,7 @@ class Future:
     # On Python 3.3 and older, objects with a destructor part of a reference
     # cycle are never destroyed. It's not more the case on Python 3.4 thanks
     # to the PEP 442.
-    if _PY34:
+    if compat.PY34:
         def __del__(self):
             if not self._log_traceback:
                 # set_exception() was not called, or result() or exception()
@@ -352,7 +350,7 @@ class Future:
         self._exception = exception
         self._state = _FINISHED
         self._schedule_callbacks()
-        if _PY34:
+        if compat.PY34:
             self._log_traceback = True
         else:
             self._tb_logger = _TracebackLogger(self, exception)
@@ -388,7 +386,7 @@ class Future:
         assert self.done(), "yield from wasn't used with future"
         return self.result()  # May raise too.
 
-    if _PY35:
+    if compat.PY35:
         __await__ = __iter__ # make compatible with 'await' expression
 
 
diff --git a/Lib/asyncio/locks.py b/Lib/asyncio/locks.py
index b2e516b5430..cc6f2bf76f1 100644
--- a/Lib/asyncio/locks.py
+++ b/Lib/asyncio/locks.py
@@ -5,14 +5,12 @@ __all__ = ['Lock', 'Event', 'Condition', 'Semaphore', 'BoundedSemaphore']
 import collections
 import sys
 
+from . import compat
 from . import events
 from . import futures
 from .coroutines import coroutine
 
 
-_PY35 = sys.version_info >= (3, 5)
-
-
 class _ContextManager:
     """Context manager.
 
@@ -70,7 +68,7 @@ class _ContextManagerMixin:
         yield from self.acquire()
         return _ContextManager(self)
 
-    if _PY35:
+    if compat.PY35:
 
         def __await__(self):
             # To make "with await lock" work.
diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py
index 176c65e3969..6cd60c42f64 100644
--- a/Lib/asyncio/streams.py
+++ b/Lib/asyncio/streams.py
@@ -12,6 +12,7 @@ if hasattr(socket, 'AF_UNIX'):
     __all__.extend(['open_unix_connection', 'start_unix_server'])
 
 from . import coroutines
+from . import compat
 from . import events
 from . import futures
 from . import protocols
@@ -20,7 +21,6 @@ from .log import logger
 
 
 _DEFAULT_LIMIT = 2**16
-_PY35 = sys.version_info >= (3, 5)
 
 
 class IncompleteReadError(EOFError):
@@ -488,7 +488,7 @@ class StreamReader:
 
         return b''.join(blocks)
 
-    if _PY35:
+    if compat.PY35:
         @coroutine
         def __aiter__(self):
             return self
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index d8193ba48e1..1d5f8654442 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -16,13 +16,12 @@ import traceback
 import warnings
 import weakref
 
+from . import compat
 from . import coroutines
 from . import events
 from . import futures
 from .coroutines import coroutine
 
-_PY34 = (sys.version_info >= (3, 4))
-
 
 class Task(futures.Future):
     """A coroutine wrapped in a Future."""
@@ -83,7 +82,7 @@ class Task(futures.Future):
     # On Python 3.3 or older, objects with a destructor that are part of a
     # reference cycle are never destroyed. That's not the case any more on
     # Python 3.4 thanks to the PEP 442.
-    if _PY34:
+    if compat.PY34:
         def __del__(self):
             if self._state == futures._PENDING and self._log_destroy_pending:
                 context = {
diff --git a/Lib/asyncio/transports.py b/Lib/asyncio/transports.py
index 22df3c7aede..7a28d908e43 100644
--- a/Lib/asyncio/transports.py
+++ b/Lib/asyncio/transports.py
@@ -2,7 +2,7 @@
 
 import sys
 
-_PY34 = sys.version_info >= (3, 4)
+from asyncio import compat
 
 __all__ = ['BaseTransport', 'ReadTransport', 'WriteTransport',
            'Transport', 'DatagramTransport', 'SubprocessTransport',
@@ -94,12 +94,8 @@ class WriteTransport(BaseTransport):
         The default implementation concatenates the arguments and
         calls write() on the result.
         """
-        if not _PY34:
-            # In Python 3.3, bytes.join() doesn't handle memoryview.
-            list_of_data = (
-                bytes(data) if isinstance(data, memoryview) else data
-                for data in list_of_data)
-        self.write(b''.join(list_of_data))
+        data = compat.flatten_list_bytes(list_of_data)
+        self.write(data)
 
     def write_eof(self):
         """Close the write end after flushing buffered data.

From eaf16abc68a09e2d976c37e34eb606f5b519f7ea Mon Sep 17 00:00:00 2001
From: Victor Stinner <victor.stinner@gmail.com>
Date: Sat, 25 Jul 2015 02:40:40 +0200
Subject: [PATCH 2/2] asyncio: sync with github

* Fix ResourceWarning warnings in test_streams
* Return True from StreamReader.eof_received() to fix
  http://bugs.python.org/issue24539 (but still needs a unittest).
  Add StreamReader.__repr__() for easy debugging.
* remove unused imports
* Issue #234: Drop JoinableQueue on Python 3.5+
---
 Lib/asyncio/locks.py                  |  1 -
 Lib/asyncio/queues.py                 | 10 ++++++----
 Lib/asyncio/streams.py                | 20 ++++++++++++++++++-
 Lib/asyncio/subprocess.py             |  2 --
 Lib/asyncio/tasks.py                  |  2 --
 Lib/asyncio/transports.py             |  2 --
 Lib/test/test_asyncio/test_streams.py | 28 ++++++++++++---------------
 7 files changed, 37 insertions(+), 28 deletions(-)

diff --git a/Lib/asyncio/locks.py b/Lib/asyncio/locks.py
index cc6f2bf76f1..7a132796b4f 100644
--- a/Lib/asyncio/locks.py
+++ b/Lib/asyncio/locks.py
@@ -3,7 +3,6 @@
 __all__ = ['Lock', 'Event', 'Condition', 'Semaphore', 'BoundedSemaphore']
 
 import collections
-import sys
 
 from . import compat
 from . import events
diff --git a/Lib/asyncio/queues.py b/Lib/asyncio/queues.py
index 3b4dc21ab86..c55dd8bbb0a 100644
--- a/Lib/asyncio/queues.py
+++ b/Lib/asyncio/queues.py
@@ -1,11 +1,11 @@
 """Queues"""
 
-__all__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty',
-           'JoinableQueue']
+__all__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty']
 
 import collections
 import heapq
 
+from . import compat
 from . import events
 from . import futures
 from . import locks
@@ -289,5 +289,7 @@ class LifoQueue(Queue):
         return self._queue.pop()
 
 
-JoinableQueue = Queue
-"""Deprecated alias for Queue."""
+if not compat.PY35:
+    JoinableQueue = Queue
+    """Deprecated alias for Queue."""
+    __all__.append('JoinableQueue')
diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py
index 6cd60c42f64..6484c435aac 100644
--- a/Lib/asyncio/streams.py
+++ b/Lib/asyncio/streams.py
@@ -6,7 +6,6 @@ __all__ = ['StreamReader', 'StreamWriter', 'StreamReaderProtocol',
            ]
 
 import socket
-import sys
 
 if hasattr(socket, 'AF_UNIX'):
     __all__.extend(['open_unix_connection', 'start_unix_server'])
@@ -240,6 +239,7 @@ class StreamReaderProtocol(FlowControlMixin, protocols.Protocol):
 
     def eof_received(self):
         self._stream_reader.feed_eof()
+        return True
 
 
 class StreamWriter:
@@ -321,6 +321,24 @@ class StreamReader:
         self._transport = None
         self._paused = False
 
+    def __repr__(self):
+        info = ['StreamReader']
+        if self._buffer:
+            info.append('%d bytes' % len(info))
+        if self._eof:
+            info.append('eof')
+        if self._limit != _DEFAULT_LIMIT:
+            info.append('l=%d' % self._limit)
+        if self._waiter:
+            info.append('w=%r' % self._waiter)
+        if self._exception:
+            info.append('e=%r' % self._exception)
+        if self._transport:
+            info.append('t=%r' % self._transport)
+        if self._paused:
+            info.append('paused')
+        return '<%s>' % ' '.join(info)
+
     def exception(self):
         return self._exception
 
diff --git a/Lib/asyncio/subprocess.py b/Lib/asyncio/subprocess.py
index 4600a9f417d..ead4039b2f7 100644
--- a/Lib/asyncio/subprocess.py
+++ b/Lib/asyncio/subprocess.py
@@ -1,10 +1,8 @@
 __all__ = ['create_subprocess_exec', 'create_subprocess_shell']
 
-import collections
 import subprocess
 
 from . import events
-from . import futures
 from . import protocols
 from . import streams
 from . import tasks
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index 1d5f8654442..9bfc1cf8147 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -10,8 +10,6 @@ import concurrent.futures
 import functools
 import inspect
 import linecache
-import sys
-import types
 import traceback
 import warnings
 import weakref
diff --git a/Lib/asyncio/transports.py b/Lib/asyncio/transports.py
index 7a28d908e43..70b323f2db9 100644
--- a/Lib/asyncio/transports.py
+++ b/Lib/asyncio/transports.py
@@ -1,7 +1,5 @@
 """Abstract Transport class."""
 
-import sys
-
 from asyncio import compat
 
 __all__ = ['BaseTransport', 'ReadTransport', 'WriteTransport',
diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py
index 242b377eb46..ef6f6030343 100644
--- a/Lib/test/test_asyncio/test_streams.py
+++ b/Lib/test/test_asyncio/test_streams.py
@@ -446,6 +446,8 @@ class StreamReaderTests(test_utils.TestCase):
             def handle_client(self, client_reader, client_writer):
                 data = yield from client_reader.readline()
                 client_writer.write(data)
+                yield from client_writer.drain()
+                client_writer.close()
 
             def start(self):
                 sock = socket.socket()
@@ -457,12 +459,8 @@ class StreamReaderTests(test_utils.TestCase):
                 return sock.getsockname()
 
             def handle_client_callback(self, client_reader, client_writer):
-                task = asyncio.Task(client_reader.readline(), loop=self.loop)
-
-                def done(task):
-                    client_writer.write(task.result())
-
-                task.add_done_callback(done)
+                self.loop.create_task(self.handle_client(client_reader,
+                                                         client_writer))
 
             def start_callback(self):
                 sock = socket.socket()
@@ -522,6 +520,8 @@ class StreamReaderTests(test_utils.TestCase):
             def handle_client(self, client_reader, client_writer):
                 data = yield from client_reader.readline()
                 client_writer.write(data)
+                yield from client_writer.drain()
+                client_writer.close()
 
             def start(self):
                 self.server = self.loop.run_until_complete(
@@ -530,18 +530,14 @@ class StreamReaderTests(test_utils.TestCase):
                                               loop=self.loop))
 
             def handle_client_callback(self, client_reader, client_writer):
-                task = asyncio.Task(client_reader.readline(), loop=self.loop)
-
-                def done(task):
-                    client_writer.write(task.result())
-
-                task.add_done_callback(done)
+                self.loop.create_task(self.handle_client(client_reader,
+                                                         client_writer))
 
             def start_callback(self):
-                self.server = self.loop.run_until_complete(
-                    asyncio.start_unix_server(self.handle_client_callback,
-                                              path=self.path,
-                                              loop=self.loop))
+                start = asyncio.start_unix_server(self.handle_client_callback,
+                                                  path=self.path,
+                                                  loop=self.loop)
+                self.server = self.loop.run_until_complete(start)
 
             def stop(self):
                 if self.server is not None: