Optimise reading data from socket

README.rst

@@ -28,6 +28,7 @@ file `guide <https://github.com/grpc/grpc/blob/master/BUILDING.md>`_ and follow
     $ cd grpc-asyncio/vendor/grpc
     $ git submodule init
     $ git submodule update --recursive
+    $ git submodule update --init
     $ make
 
 For debugging purposes the library can be compiled in debug mode, as can be seen in the following command::

grpc_asyncio/socket.pyx

@@ -1,7 +1,11 @@
+import asyncio
 import socket
 
 from libc cimport string
 
+include "streams.pyx"
+
+
 cdef class Socket:
     def __cinit__(self):
         self.g_socket = NULL
@@ -80,19 +84,24 @@ cdef class Socket:
     cdef void connect(self, object host, object port, grpc_custom_connect_callback g_connect_cb):
         assert not self.task_connect
 
+        self.g_connect_cb = g_connect_cb
         self.task_connect = asyncio.create_task(
-            asyncio.open_connection(host, port)
+            open_connection(host, port)
         )
         self.task_connect.add_done_callback(self._connect_cb)
-        self.g_connect_cb = g_connect_cb
 
     cdef void read(self, char * buffer_, size_t length, grpc_custom_read_callback g_read_cb):
         assert not self.task_read
 
-        self.task_read = asyncio.create_task(
-            self.reader.read(n=length)
-        )
-        self.task_read.add_done_callback(self._read_cb)
+        if self.reader.is_data_available(length):
+          self._read_cb(self.reader.read(n=length))
+
+        else:
+          self.task_read = asyncio.create_task(
+              self.reader.read(n=length)
+          )
+          self.task_read.add_done_callback(self._read_cb)
+
         self.g_read_cb = g_read_cb
         self.read_buffer = buffer_
 

grpc_asyncio/streams.pyx

@@ -0,0 +1,25 @@
+import asyncio
+
+
+async def open_connection(
+        host=None, port=None, *, loop=None, limit=2 ** 16, **kwargs):
+
+    if loop is None:
+        loop = asyncio.get_event_loop()
+    reader = StreamReader(limit=limit, loop=loop)
+    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
+    transport, _ = await loop.create_connection(
+        lambda: protocol, host, port, **kwargs)
+    writer = asyncio.StreamWriter(transport, protocol, reader, loop)
+    return reader, writer
+
+
+class StreamReader(asyncio.StreamReader):
+    def is_data_available(self, n):
+        """
+        Check if there is n length data in the buffer
+        """
+        if len(self._buffer) >= n:
+            return True
+
+        return False