fist working code of a tailable cursor
This commit is contained in:
@@ -32,10 +32,11 @@ _QUERY_OPTIONS = {
|
||||
|
||||
|
||||
class _MongoQuery(object):
|
||||
def __init__(self, id, collection, limit):
|
||||
def __init__(self, id, collection, limit, tailable=False):
|
||||
self.id = id
|
||||
self.limit = limit
|
||||
self.collection = collection
|
||||
self.tailable = tailable
|
||||
self.documents = []
|
||||
self.future = asyncio.Future()
|
||||
|
||||
@@ -180,7 +181,7 @@ class MongoProtocol(asyncio.Protocol):
|
||||
query_opts |= _QUERY_OPTIONS["await_data"]
|
||||
query_opts = struct.pack("<i", query_opts)
|
||||
|
||||
query = _MongoQuery(self.__id, collection, limit)
|
||||
query = _MongoQuery(self.__id, collection, limit, tailable)
|
||||
self.__queries[self.__id] = query
|
||||
self.send_message(2004, collection, message, query_opts)
|
||||
return query.future
|
||||
@@ -196,6 +197,8 @@ class MongoProtocol(asyncio.Protocol):
|
||||
query = self.__queries.pop(request_id)
|
||||
except KeyError:
|
||||
return
|
||||
if query.tailable:
|
||||
query.documents = []
|
||||
if isinstance(documents, list):
|
||||
query.documents += documents
|
||||
else:
|
||||
@@ -216,5 +219,10 @@ class MongoProtocol(asyncio.Protocol):
|
||||
return
|
||||
self.__queries[self.__id] = query
|
||||
self.OP_GET_MORE(query.collection, next_batch, cursor_id)
|
||||
if query.tailable and query.documents:
|
||||
f = asyncio.Future()
|
||||
_f = query.future
|
||||
query.future = f
|
||||
_f.set_result((f, documents))
|
||||
else:
|
||||
query.future.set_result(query.documents)
|
||||
|
||||
Reference in New Issue
Block a user