'async for' support in streams in python 3.5+

prepare cancleable calls
python 2/3 str/unicode fixes
documentation of protocol messages

Author: Dobatymo

.gitignore

@@ -1,6 +1,8 @@
 # CeRPCerus
 testing/
 js/
+*.bak
+*.log
 
 # Twisted
 _trial_temp/
@@ -26,18 +28,6 @@ parts/
 sdist/
 var/
 *.egg-info/
-.installed.cfg
-*.egg
-
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
 
 # Unit test / coverage reports
 htmlcov/
@@ -47,17 +37,6 @@ htmlcov/
 nosetests.xml
 coverage.xml
 
-# Translations
-*.mo
-*.pot
-
-# Django stuff:
-*.log
-
 # Sphinx documentation
 docs/
 docs/_build/
-
-# PyBuilder
-target/
-install.bat

cerpcerus/iter_deferred.py

@@ -2,36 +2,48 @@
 
 import queue, logging
 
-from twisted.internet import defer, reactor, protocol
+from twisted.internet import defer
+
+from future.utils import PY2
+
+if PY2:
+	StopAsyncIteration = Exception
 
 logger = logging.getLogger(__name__)
 
 """
 call self.transport.pauseProducing() and self.transport.resumeProducing() if stream is sending too fast (queue gets too large
 """
 
-class MultiDeferredIterator(object):
+class MultiDeferredIterator(object): # inherit from RemoteObjectGeneric?
 
 	"""should this be made pause/resume-able?
 	should be stoppable for sure
 	maybe the stream should include information if it can be paused
 
 	if this stops the producer/transport, it will be stopped for all streams.
 	It would be better to have on collector with one queue for all streams, so it can be stopped with regard to all streams.
-	if every single iterator can pause/resume, there will be nterference among them
+	if every single iterator can pause/resume, there will be interference among them
 	"""
 
-	def __init__(self):
+	def __init__(self, conn, sequid, classname):
+		self._conn = conn
+		self._sequid = sequid
+		self._classname = classname
+
 		self.queue = queue.Queue()
 		self.deferred = None
 		self.transport = None # interfaces.IPushProducer
 
 	# called by client
 
+	def abort(self): # this should be called from the generators .close() method as well to stop the server from sending
+		self._conn._cancel_stream(self._sequid)
+
 	def __iter__(self):
 		return self
 
-	def next(self): # __next__ python3, this can hang if user is not careful
+	def __next__(self): # this can hang if user is not careful
 		#self.transport.resumeProducing() # if queue too empty
 		try:
 			deferred = self.queue.get_nowait()
@@ -40,6 +52,24 @@ def next(self): # __next__ python3, this can hang if user is not careful
 			self.deferred = defer.Deferred()
 			return self.deferred
 
+	next = __next__
+
+	def __aiter__(self):
+		return self
+
+	@defer.inlineCallbacks
+	def __anext__(self):
+		try:
+			try:
+				result = yield self.queue.get_nowait() # or yield from?
+				defer.returnValue(result) # return result
+			except queue.Empty:
+				self.deferred = defer.Deferred()
+				result = yield self.deferred # or yield from?
+				defer.returnValue(result) # return result
+		except StopIteration: # translate, as interface which is used by the user (iter/aiter) is unknown at time of raise
+			raise StopAsyncIteration()
+
 	def stop(self):
 		pass
 
@@ -71,68 +101,49 @@ def errback(self, val):
 	def completed(self):
 		return self.errback(StopIteration())
 
-""" py 3 only
-class AsyncMultiDeferredIterator:
+	def async_completed(self):
+		return self.errback(StopAsyncIteration())
 
-	def __init__(self):
-		self.queue = queue.Queue()
-		self.deferred = None
-
-	def __aiter__(self):
-		return self
+# tests
 
-	async def __anext__(self):
-		try:
-			result = await self.queue.get_nowait()
-			if result == "stop":
-				raise StopAsyncIteration
-			return result
-		except queue.Empty:
-			self.deferred = defer.Deferred()
-			result = await self.deferred
-			if result == "stop":
-				raise StopAsyncIteration
-			return result
-
-	def callback(self, val): # called multiple times
-		if self.deferred and not self.deferred.called:
-			print("call directly")
-			self.deferred.callback(val)
-		else:
-			print("add to queue")
-			self.queue.put_nowait(defer.succeed(val))
-
-	def errback(self, val):
-		if self.deferred and not self.deferred.called:
-			print("call directly")
-			self.deferred.errback(val)
-		else:
-			print("add to queue")
-			self.queue.put_nowait(defer.fail(val))
+from twisted.internet import reactor, protocol
 
+# python >= 3.5
+""" syntax error for older python
 async def async_recv_stream(async_iter):
+	from twisted.internet import error
+
 	print("start")
-	async for deferred in async_iter:
-		print(deferred)
+	try:
+		async for result in async_iter:
+			print(result)
+			await sleep(1)
+			print("slept for a second")
+
+	except error.ConnectionDone:
+		print("ConnectionDone")
+	except error.ConnectionLost:
+		print("ConnectionLost")
 	print("end")
-	return True
 """
 
-from .utils import sleep
-
+# python <= 3.4
 @defer.inlineCallbacks
 def recv_stream(async_iter):
+	print("start")
 	for deferred in async_iter:
 		try:
 			result = yield deferred
 			print(result)
 		except StopIteration:
-			print("stop")
+			print("end")
 			break
 		except GeneratorExit:
 			logger.exception("GeneratorExit in recv")
+			break
 		except Exception:
 			logger.exception("Exception in recv")
+			break
 		yield sleep(1)
 		print("slept for a second")
 	reactor.stop()
@@ -144,10 +155,14 @@ def __init__(self, mdit):
 
 	def dataReceived(self, data):
 		if data == b"\x1b": # ESCAPE in TELNET
-			self.mdit.stop()
+			#self.mdit.stop()
+			self.mdit.completed()
 		else:
 			self.mdit.callback(data)
 
+	def connectionLost(self, reason):
+		self.mdit.errback(reason)
+
 class RecvFactory(protocol.Factory):
 
 	def __init__(self, mdit):
@@ -156,18 +171,19 @@ def __init__(self, mdit):
 	def buildProtocol(self, addr):
 		return Recv(self.mdit)
 
-def main1():
+def main1(): # connect with telnet
 	mdit = MultiDeferredIterator()
 	reactor.listenTCP(8000, RecvFactory(mdit))
 	reactor.callWhenRunning(recv_stream, mdit)
 	reactor.run()
 
-def main2():
-	import asyncio
-	mdit = AsyncMultiDeferredIterator()
+def main2(): # connect with telnet
+	from twisted.internet import task
+	mdit = MultiDeferredIterator()
 	reactor.listenTCP(8000, RecvFactory(mdit))
-	reactor.callWhenRunning(defer.Deferred.fromFuture(asyncio.ensure_future(async_recv_stream)), mdit)
+	task.react(lambda reactor: defer.ensureDeferred(async_recv_stream(mdit)))
 	reactor.run()
 
 if __name__ == "__main__":
+	from utils import sleep
 	main1()

cerpcerus/rpc.py

@@ -1,4 +1,5 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
+from builtins import str
 
 from future.utils import iteritems
 
@@ -45,6 +46,7 @@ class CallAnyPublic(object):
 	"""Baseclass which delegates method calls based on leading "_"."""
 
 	def __getattr__(self, name): # only gets invoked for attributes that are not already defined. __getattribute__ would be called always
+		# name is byte string in python 2 and unicode string in python 3
 		if not name.startswith("_"):
 			return self._callpublic(name)
 		else:
@@ -84,11 +86,10 @@ def _callpublic(self, name):
 		return partial(self._call, self._alias.get(name, name))
 
 	def _callprivate(self, name):
-	
 		"""called from user, so return user friendly exception """
-	
+
 		# vs: __class__.__name__?
-		raise AttributeError("{} instance has no attribute '{}'".format(type(self).__name__, name))
+		raise AttributeError("{} instance has no attribute '{}'".format(type(self).__name__, name)) # str(name) needed for python2?
 
 	@property
 	def _connid(self):
@@ -134,27 +135,37 @@ def __init__(self, conn):
 		RemoteObjectGeneric.__init__(self, conn)
 
 	def _call(self, _name, *args, **kwargs):
-		# type: (unicode, *Any, **Any) -> RemoteResultDeferred
-		_name = unicode(_name)
+		# type: (str, *Any, **Any) -> RemoteResultDeferred
+		_name = str(_name) # this is needed for python2 is it?
 		return self._conn._call(_name, *args, **kwargs)
 
 	def _call_with_streams(self, _name, *args, **kwargs):
-		# type: (unicode, *Any, **Any) -> RemoteResultDeferred
-		_name = unicode(_name)
+		# type: (str, *Any, **Any) -> RemoteResultDeferred
+		_name = str(_name)
 		return self._conn._call_with_streams(_name, *args, **kwargs)
 
 	def _notify(self, _name, *args, **kwargs):
-		# type: (unicode, *Any, **Any) -> None
-		_name = unicode(_name)
+		# type: (str, *Any, **Any) -> None
+		_name = str(_name)
 		self._conn._notify(_name, *args, **kwargs)
 
+	def _notify_with_streams(self, _name, *args, **kwargs):
+		# type: (str, *Any, **Any) -> None
+		_name = str(_name)
+		self._conn._notify_with_streams(_name, *args, **kwargs)
+
 	def _stream(self, _name, *args, **kwargs): # could return a async iterator in the far future
-		# type: (unicode, *Any, **Any) -> MultiDeferredIterator
-		_name = unicode(_name)
+		# type: (str, *Any, **Any) -> MultiDeferredIterator
+		_name = str(_name)
 		return self._conn._stream(_name, *args, **kwargs)
 
+	def _stream_with_streams(self, _name, *args, **kwargs): # could return a async iterator in the far future
+		# type: (str, *Any, **Any) -> MultiDeferredIterator
+		_name = str(_name)
+		return self._conn._stream_with_streams(_name, *args, **kwargs)
+
 	def __repr__(self):
-		# type: () -> unicode
+		# type: () -> str
 		return "'<RemoteObject object to {} at {}>'".format(self._conn.name, self._conn.addr)
 
 	def __dir__(self):
@@ -169,29 +180,42 @@ class RemoteInstance(RemoteObjectGeneric):
 
 	def __init__(self, conn, objectid, classname):
 		RemoteObjectGeneric.__init__(self, conn)
-		assert isinstance(classname, unicode)
+		assert isinstance(classname, str)
 		self._objectid = objectid
 		self._classname = classname
 
 	def _call(self, _name, *args, **kwargs):
-		# type: (unicode, *Any, **Any) -> RemoteResultDeferred
+		# type: (str, *Any, **Any) -> RemoteResultDeferred
+		_name = str(_name) # this is needed for python2 is it?
 		return self._conn._callmethod(self._objectid, _name, *args, **kwargs)
 
 	def _call_with_streams(self, _name, *args, **kwargs):
-		# type: (unicode, *Any, **Any) -> RemoteResultDeferred
-		_name = unicode(_name)
+		# type: (str, *Any, **Any) -> RemoteResultDeferred
+		_name = str(_name)
 		return self._conn._callmethod_with_streams(_name, *args, **kwargs)
 
 	def _notify(self, _name, *args, **kwargs):
-		# type: (unicode, *Any, **Any) -> None
+		# type: (str, *Any, **Any) -> None
+		_name = str(_name)
 		self._conn._notifymethod(self._objectid, _name, *args, **kwargs)
 
+	def _notify_with_streams(self, _name, *args, **kwargs):
+		# type: (str, *Any, **Any) -> None
+		_name = str(_name)
+		self._conn._notifymethod_with_streams(self._objectid, _name, *args, **kwargs)
+
 	def _stream(self, _name, *args, **kwargs): # could return a async iterator in the far future
-		# type: (unicode, *Any, **Any) -> MultiDeferredIterator
+		# type: (str, *Any, **Any) -> MultiDeferredIterator
+		_name = str(_name)
 		return self._conn._streammethod(self._objectid, _name, *args, **kwargs)
 
+	def _stream_with_streams(self, _name, *args, **kwargs): # could return a async iterator in the far future
+		# type: (str, *Any, **Any) -> MultiDeferredIterator
+		_name = str(_name)
+		return self._conn._streammethod_with_streams(self._objectid, _name, *args, **kwargs)
+
 	def __repr__(self):
-		# type: () -> unicode
+		# type: () -> str
 		return "'<RemoteInstance object {} [{}] to {} at {}>'".format(self._classname, self._objectid, self._conn.name, self._conn.addr)
 
 	def __dir__(self):
@@ -219,17 +243,24 @@ class RemoteResult(RemoteObjectGeneric):
 
 	def __init__(self, conn, sequid, classname):
 		RemoteObjectGeneric.__init__(self, conn)
+		assert isinstance(classname, str)
 		self._sequid = sequid
 		self._classname = classname
 
 	def _call(self, _name, *args, **kwargs):
-		return self._conn._callmethod_onresult(self._sequid, _name, *args, **kwargs) # type: CallOnDeferred
+		# type: (str, *Any, **Any) -> RemoteResultDeferred
+		_name = str(_name) # this is needed for python2 is it?
+		return self._conn._callmethod_by_result(self._sequid, _name, *args, **kwargs)
 
 	def _notify(self, _name, *args, **kwargs):
-		self._conn._notifymethod_onresult(self._sequid, _name, *args, **kwargs) # type: None
+		# type: (str, *Any, **Any) -> None
+		_name = str(_name)
+		self._conn._notifymethod_by_result(self._sequid, _name, *args, **kwargs)
 
 	def _stream(self, _name, *args, **kwargs): # could return a async iterator in the far future
-		return self._conn._streammethod_onresult(self._sequid, _name, *args, **kwargs) # type: MultiDeferredIterator
+		# type: (str, *Any, **Any) -> MultiDeferredIterator
+		_name = str(_name)
+		return self._conn._streammethod_by_result(self._sequid, _name, *args, **kwargs)
 
 	def __repr__(self):
 		return "'<RemoteResult object {} [{}] to {} at {}>'".format(self._classname, self._sequid, self._conn.name, self._conn.addr)
@@ -241,8 +272,7 @@ def __dir__(self):
 	def __del__(self): # called when garbage collected
 		# RemoteObjectGeneric.__del__() # should be called, but doesn't exist
 		try:
-			#self._conn._delinstanceonresult(self._sequid) # this tries to delete results which are not actually objects
-			pass
+			self._conn._delinstance_by_result(self._sequid)
 		except NotAuthenticated:
 			pass
 
@@ -293,7 +323,7 @@ def call_and_catch_signature_error(_attr, *args, **kwargs):
 		return _attr(*args, **kwargs)
 	except TypeError as e:
 		try:
-			signature(_attr).bind(*args, **kwargs) #do sig test only in error case to preserve resources on successful calls
+			signature(_attr).bind(*args, **kwargs) # do sig test only in error case to preserve resources on successful calls
 			raise
 		except TypeError:
 			raise RPCInvalidArguments(e)
@@ -331,11 +361,11 @@ def __delitem__(self, key):
 
 	@classmethod
 	def _aliases(cls, aliases):
-	
+
 		""" this is intended to be a decorator used for functions in services.
 			yet the decorator is applied at class construction time and has thus no
 			access to self. but class wide aliases would be bad."""
-	
+
 		def decorator(func):
 			for alias in aliases:
 				cls._alias[alias] = func.__name__
@@ -419,7 +449,7 @@ def _delete(self, connid, objectid):
 		except KeyError:
 			raise RPCInvalidObject("No object with id {}".format(objectid))
 
-	def _deleteAllObjects(self, connid):
+	def _delete_all_objects(self, connid):
 		"""Deletes all objects created by this connection."""
 		try:
 			self._objects = {oid: (obj, connid_) for oid, (obj, connid_) in iteritems(self._objects) if connid_ != connid}
@@ -451,7 +481,7 @@ def __init__(self, *args, **kwargs):
 	# generator
 	def random(self, size, num):
 		return random(size, num)
-	
+
 	def echo(self, data):
 		return data
 

cerpcerus/rpcbase.py

@@ -1,5 +1,7 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+from builtins import str
+
 import inspect, itertools
 import unittest
 

cerpcerus/test/test_utils.py

@@ -210,6 +210,17 @@ def IProtocolConnector(transport, protocol):
 	"""
 	transport: not a twisted transport, but a protocol
 	protocol: not a twisted protocol but any class (e.g. RPCBase)
+	
+	this is needed because RPCBase cannot inherit from user defined class. But if different protocols
+		are to be supported, there needs to be a way to connect the twisted protcol callbacks.
+	
+	btw:
+	def asd(protcol_class):
+		class rpc_class(protcol_class):
+			pass
+		return rpc_class
+	a = asd()
+	that works...
 	"""
 	transport.recv_data = protocol.recv_data
 	transport.connection_made = protocol.connection_made

cerpcerus/utils.py

@@ -7,20 +7,24 @@
 import cerpcerus
 from cerpcerus.rpcbase import RPCUserError
 
+from utils import randblob
+
 def scandir_rec(path):
 	for direntry in os.scandir(path):
 		if direntry.is_dir():
 			yield from scandir_rec(direntry)
 		elif direntry.is_file():
 			yield direntry
 
+randomdata = randblob(1024*1024) # 16MB
+
 class TestService(cerpcerus.rpc.DebugService):
 	def __init__(self, reactor, conn):
 		cerpcerus.rpc.DebugService.__init__(self, True)
 		self.shared = Path("D:/PUBLIC").resolve(strict=True) # needs to be absolute, existing path
 
 	def list(self):
-		return tuple((Path(direntry).relative_to(self.shared).path, direntry.stat().st_size) for direntry in scandir_rec(self.shared))
+		return tuple((str(Path(direntry).relative_to(self.shared)), direntry.stat().st_size) for direntry in scandir_rec(self.shared))
 
 	def file(self, relpath, seek=0):
 		try:
@@ -44,6 +48,10 @@ def file(self, relpath, seek=0):
 		except OSError:
 			raise RPCUserError("Invalid file")
 
+	def infinite(self):
+		while True:
+			yield randomdata
+
 class MySSLContextFactory(cerpcerus.GenericRPCSSLContextFactory):
 
 	def __init__(self):

examples/filesend_server.py

@@ -0,0 +1,106 @@
+-- RPC message type --
+
+total messages: 3*2 + 3*2 + 2 + 3 + 6 = 23
+
+guidelines:
+	don't create new client-to-server messages based on return value
+		sometimes different api client functions might be needed, but the same messages should be used
+
+	distinguish based on sent information, so that messages can be handled by server quickly and easily
+		_with_streams would not be strictly necessary, but it increases performance if the server does not have to find out
+		if any of the arguments are streams and need to be handles differently
+
+	tldr:
+		make client-to-server and server-to-client messages orthogonal
+
+# Client to server, no response, errors are ignored silently
+
+NOTIFY:
+	function call
+
+NOTIFYMETHOD
+	method call on OBJECT 
+
+NOTIFYMETHOD_BY_RESULT
+	method call on OBJECT, referred to by RESULT
+
+*_WITH_STREAMS # not implemented for all methods yet
+	arguments can contain `Streamables`
+	separate messages for performance reasons
+	could be implemented as another parameter in request. eg.
+	client could send 0, 1, 2:
+		0: auto, server has to figure out if streams are included
+		1: client says there are no streams
+		2: client says there are streams
+    actually 0 is not neccessary as the work done would be the same as 2. "no streams" is just a performance optimisation
+
+# Client to server, return values are
+- RESULT, OBJECT, ERROR
+- STREAM_RESULT, STREAM_END, STREAM_ERROR
+
+CALL
+	function call
+
+CALLMETHOD
+	method call on OBJECT 
+
+CALLMETHOD_BY_RESULT
+	method call on OBJECT, referred to by RESULT
+
+*_WITH_STREAMS # not implemented for all methods yet
+	arguments can contain `Streamables`
+	separate messages for performance reasons
+
+# Client to server, no response, errors are ignored silently
+
+DELINSTANCE # should be DELOBJECT for symmetry?
+	deletes remote OBJECT
+
+DELINSTANCE_BY_RESULT
+	deletes remote OBJECT, referred to by RESULT
+
+# Client to server, streams, only indirectly user initiated
+
+STREAM_ARGUMENT
+	used with the *_WITH_STREAMS messages to deliver data
+
+ARGUMENT_ERROR
+	used with the *_WITH_STREAMS messages to signal errors
+
+ARGUMENT_END
+	used with the *_WITH_STREAMS messages to mark the end of a stream
+
+# Server to client
+
+RESULT
+	normal value, i.e. no `ObjectId` or `GeneratorType` which are handled separately
+	can send *METHOD_BY_RESULT messages using this return type
+
+OBJECT
+	object instance
+	can send *METHOD messages using this return type
+
+ERROR
+	signals an error, possible errors are
+	- GeneralError
+	- NoSuchFunction
+	- WrongArguments
+	- UserError
+	- NoService
+	- Deferred
+	- InvalidObject
+	- ResourceExhausted
+
+# Server to client, streams, only indirectly user initiated
+
+STREAM_RESULT
+	stream value, basically an iterable result
+
+STREAM_ERROR
+	signals error during stream
+
+STREAM_END
+	marks the end of a stream
+
+# Notes
+"STREAM" messages are not needed, but the client needs to call special `_stream*` functions on the library so the returned STREAM_* messages can be handles correctly

message-types.txt

@@ -0,0 +1,4 @@
+future
+twisted
+msgpack-python
+pyOpenSSL

requirements.txt

@@ -16,6 +16,7 @@
 	packages=["cerpcerus"],
 	test_suite="test",
 	install_requires=[
+		"future",
 		"twisted",
 		"msgpack-python",
 		"pyOpenSSL",