我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用inspect.isgeneratorfunction()。
def _process_rep_event(self, socket, addr, data): """ Process a REP socket's event. Parameters ---------- socket : zmq.Socket Socket that generated the event. addr : AgentAddress AgentAddress associated with the socket that generated the event. data : bytes Data received on the socket. """ message = deserialize_message(message=data, serializer=addr.serializer) handler = self.handler[socket] if inspect.isgeneratorfunction(handler): generator = handler(self, message) socket.send(serialize_message(next(generator), addr.serializer)) execute_code_after_yield(generator) else: reply = handler(self, message) socket.send(serialize_message(reply, addr.serializer))
def trollbox_wrapper(handler): async def decorator(data): if len(data) != 5: return type_ = data[0] message_id = data[1] username = data[2] text = data[3] reputation = data[4] kwargs = { "id": message_id, "username": username, "type": type_, "text": text, "reputation": reputation } if inspect.isgeneratorfunction(handler): await handler(**kwargs) else: handler(**kwargs) return decorator
def create_bench(name, env): srmock = helpers.StartResponseMock() function = name.lower().replace('-', '_') app = eval('create.{0}(BODY, HEADERS)'.format(function)) def bench(): app(env, srmock) if srmock.status != '200 OK': raise AssertionError(srmock.status + ' != 200 OK') def bench_generator(): exhaust(app(env, srmock)) if srmock.status != '200 OK': raise AssertionError(srmock.status + ' != 200 OK') if inspect.isgeneratorfunction(app): return bench_generator else: return bench
def __service(func, resource, message): assert isinstance(resource, Resource) assert isinstance(message, simpype.Message) message.location = resource if inspect.isgeneratorfunction(func): mid = str(message.id)+str(message.seq_num) a_serve = resource.env.process(func(resource, message)) resource.task[mid] = Task(message.sim, message, a_serve) try: yield a_serve message.timestamp('resource.serve') except simpy.Interrupt as interrupt: message.timestamp('resource.'+str(interrupt.cause)) del resource.task[mid] else: func(resource, message) message.timestamp('resource.serve') if message.next: resource.send(message) else: message.done()
def default(self, obj): if hasattr(obj, "to_json"): return self.default(obj.to_json()) elif hasattr(obj, "__dict__"): d = dict( (key, value) for key, value in inspect.getmembers(obj) if not key.startswith("__") and not inspect.isabstract(value) and not inspect.isbuiltin(value) and not inspect.isfunction(value) and not inspect.isgenerator(value) and not inspect.isgeneratorfunction(value) and not inspect.ismethod(value) and not inspect.ismethoddescriptor(value) and not inspect.isroutine(value) ) return self.default(d) return obj
def test_excluding_predicates(self): self.istest(inspect.isbuiltin, 'sys.exit') if check_impl_detail(): self.istest(inspect.isbuiltin, '[].append') self.istest(inspect.iscode, 'mod.spam.__code__') self.istest(inspect.isframe, 'tb.tb_frame') self.istest(inspect.isfunction, 'mod.spam') self.istest(inspect.isfunction, 'mod.StupidGit.abuse') self.istest(inspect.ismethod, 'git.argue') self.istest(inspect.ismodule, 'mod') self.istest(inspect.istraceback, 'tb') self.istest(inspect.isdatadescriptor, 'collections.defaultdict.default_factory') self.istest(inspect.isgenerator, '(x for x in range(2))') self.istest(inspect.isgeneratorfunction, 'generator_function_example') if hasattr(types, 'GetSetDescriptorType'): self.istest(inspect.isgetsetdescriptor, 'type(tb.tb_frame).f_locals') else: self.assertFalse(inspect.isgetsetdescriptor(type(tb.tb_frame).f_locals)) if hasattr(types, 'MemberDescriptorType'): self.istest(inspect.ismemberdescriptor, 'type(lambda: None).__globals__') else: self.assertFalse(inspect.ismemberdescriptor(datetime.timedelta.days))
def _handle_errors(f): if inspect.isgeneratorfunction(f): def handler(*args, **kwargs): try: for data in f(*args, **kwargs): yield data except grpc.RpcError as exc: _translate_exception(exc) else: def handler(*args, **kwargs): try: return f(*args, **kwargs) except grpc.RpcError as exc: _translate_exception(exc) return functools.wraps(f)(handler)
def test_excluding_predicates(self): self.istest(inspect.isbuiltin, 'sys.exit') self.istest(inspect.isbuiltin, '[].append') self.istest(inspect.iscode, 'mod.spam.func_code') self.istest(inspect.isframe, 'tb.tb_frame') self.istest(inspect.isfunction, 'mod.spam') self.istest(inspect.ismethod, 'mod.StupidGit.abuse') self.istest(inspect.ismethod, 'git.argue') self.istest(inspect.ismodule, 'mod') self.istest(inspect.istraceback, 'tb') self.istest(inspect.isdatadescriptor, '__builtin__.file.closed') self.istest(inspect.isdatadescriptor, '__builtin__.file.softspace') self.istest(inspect.isgenerator, '(x for x in xrange(2))') self.istest(inspect.isgeneratorfunction, 'generator_function_example') if hasattr(types, 'GetSetDescriptorType'): self.istest(inspect.isgetsetdescriptor, 'type(tb.tb_frame).f_locals') else: self.assertFalse(inspect.isgetsetdescriptor(type(tb.tb_frame).f_locals)) if hasattr(types, 'MemberDescriptorType'): self.istest(inspect.ismemberdescriptor, 'datetime.timedelta.days') else: self.assertFalse(inspect.ismemberdescriptor(datetime.timedelta.days))
def _PrintResult(component_trace, verbose=False): """Prints the result of the Fire call to stdout in a human readable way.""" # TODO: Design human readable deserializable serialization method # and move serialization to it's own module. result = component_trace.GetResult() if isinstance(result, (list, set, types.GeneratorType)): for i in result: print(_OneLineResult(i)) elif inspect.isgeneratorfunction(result): raise NotImplementedError elif isinstance(result, dict): print(_DictAsString(result, verbose)) elif isinstance(result, tuple): print(_OneLineResult(result)) elif isinstance(result, (bool, six.string_types, six.integer_types, float, complex)): print(result) elif result is not None: print(helputils.HelpString(result, component_trace, verbose))
def __getattr__(self, name): if name.startswith(self.factory_method_prefix): raise AttributeError(name) factory = self.get_factory_method_for(name) if inspect.isgeneratorfunction(factory): with self.wrapped_attribute_error(): generator = factory() instance = next(generator) self.attribute_generators.append(generator) else: with self.wrapped_attribute_error(): instance = factory() setattr(self, name, instance) self.attributes_set[name] = instance return instance
def test_excluding_predicates(self): self.istest(inspect.isbuiltin, 'sys.exit') if check_impl_detail(): self.istest(inspect.isbuiltin, '[].append') self.istest(inspect.iscode, 'mod.spam.func_code') self.istest(inspect.isframe, 'tb.tb_frame') self.istest(inspect.isfunction, 'mod.spam') self.istest(inspect.ismethod, 'mod.StupidGit.abuse') self.istest(inspect.ismethod, 'git.argue') self.istest(inspect.ismodule, 'mod') self.istest(inspect.istraceback, 'tb') self.istest(inspect.isdatadescriptor, '__builtin__.file.closed') self.istest(inspect.isdatadescriptor, '__builtin__.file.softspace') self.istest(inspect.isgenerator, '(x for x in xrange(2))') self.istest(inspect.isgeneratorfunction, 'generator_function_example') if hasattr(types, 'GetSetDescriptorType'): self.istest(inspect.isgetsetdescriptor, 'type(tb.tb_frame).f_locals') else: self.assertFalse(inspect.isgetsetdescriptor(type(tb.tb_frame).f_locals)) if hasattr(types, 'MemberDescriptorType'): self.istest(inspect.ismemberdescriptor, 'type(lambda: None).func_globals') else: self.assertFalse(inspect.ismemberdescriptor(type(lambda: None).func_globals))
def default(self, obj): # if hasattr(obj, "to_json"): # return self.default(obj.to_json()) if isinstance(obj, Enum): return obj.name elif hasattr(obj, "__dict__"): d = dict( (key, value) for key, value in inspect.getmembers(obj) if not key.startswith("__") and not inspect.isabstract(value) and not inspect.isbuiltin(value) and not inspect.isfunction(value) and not inspect.isgenerator(value) and not inspect.isgeneratorfunction(value) and not inspect.ismethod(value) and not inspect.ismethoddescriptor(value) and not inspect.isroutine(value) and not self.isempty(value) and not value is None ) return self.default(d) return obj
def protocol(func: typing.Callable[..., ProtocolGenerator[R]]) -> typing.Callable[..., Consumer[R]]: """Wraps a Ohne I/O protocol function. Under the hood this wraps the generator inside a :class:`~ohneio.Consumer`. Args: func (callable): Protocol function to wrap. (Protocol functions have to be generators) Returns: callable: wrapped function. """ if not callable(func): # pragma: no cover # This is for users misusing the library, type hinting already checks this raise ValueError("A protocol needs to a be a callable") if not inspect.isgeneratorfunction(func): # pragma: no cover # This is for users misusing the library, type hinting already checks this raise ValueError("A protocol needs to be a generator function") @functools.wraps(func) def wrapper(*args, **kwargs): return Consumer(func(*args, **kwargs)) return wrapper
def tcp_client(client_prog, family=socket.AF_INET, timeout=10): if not inspect.isgeneratorfunction(client_prog): raise TypeError('client_prog: a generator function was expected') sock = socket.socket(family, socket.SOCK_STREAM) if timeout is None: raise RuntimeError('timeout is required') if timeout <= 0: raise RuntimeError('only blocking sockets are supported') sock.settimeout(timeout) srv = Client(sock, client_prog, timeout) return srv
def add_route(app, fn): """ ???????????WEB APP??? :param app: WEB APP?? :param fn: ???? """ # ????????????? method = getattr(fn, '__method__', None) path = getattr(fn, '__route__', None) if path is None or method is None: return # ???????????? if not asyncio.iscoroutinefunction(fn) and not inspect.isgeneratorfunction(fn): fn = asyncio.coroutine(fn) logging.info('add route function: %s(%s), method(%s), path(%s)' % (fn.__name__, ', '.join(inspect.signature(fn).parameters.keys()), method, path, )) app.router.add_route(method, path, fn)
def __call__(self, func): if isgeneratorfunction(func): def inner(*args, **kwds): with self._recreate_cm(): yield from func(*args, **kwds) elif is_contextmanager(func): @contextmanager def inner(*args, **kwds): with self._recreate_cm(): with func(*args, **kwds) as ret: yield ret else: def inner(*args, **kwds): with self._recreate_cm(): return func(*args, **kwds) return wraps(func)(inner) # Some python version have a different signature for '_GeneratorContextManager.__init__', so we must adapt:
def _format_coroutine(coro): assert iscoroutine(coro) coro_name = getattr(coro, '__qualname__', coro.__name__) filename = coro.gi_code.co_filename if (isinstance(coro, CoroWrapper) and not inspect.isgeneratorfunction(coro.func)): filename, lineno = events._get_function_source(coro.func) if coro.gi_frame is None: coro_repr = '%s() done, defined at %s:%s' % (coro_name, filename, lineno) else: coro_repr = '%s() running, defined at %s:%s' % (coro_name, filename, lineno) elif coro.gi_frame is not None: lineno = coro.gi_frame.f_lineno coro_repr = '%s() running at %s:%s' % (coro_name, filename, lineno) else: lineno = coro.gi_code.co_firstlineno coro_repr = '%s() done, defined at %s:%s' % (coro_name, filename, lineno) return coro_repr
def add_route(app, fn): ''' ??????URL???? ''' method = getattr(fn, '__method__', None) path = getattr(fn, '__route__', None) if path is None or method is None: raise ValueError('@get or @post not defined in %s.' % str(fn)) if not asyncio.iscoroutinefunction(fn) and not inspect.isgeneratorfunction(fn): fn = asyncio.coroutine(fn) logging.info('add route %s %s => %s(%s)' % (method, path, fn.__name__, ', '.join(inspect.signature(fn).parameters.keys()))) app.router.add_route(method, path, RequestHandler(app, fn)) # ??????? # ???handler???????????????
def __getattr__(self, item): attr = getattr(self.delegate, item) if inspect.iscoroutinefunction(attr) or hasattr(attr, "_is_coroutine") and attr._is_coroutine or inspect.iscoroutine( attr): async def wrapper(*args, **kwargs): return self._wrap(await attr(*args, **kwargs)) return wrapper() if inspect.iscoroutine(attr) else wrapper elif inspect.isgeneratorfunction(attr) or inspect.isgenerator(attr): def wrapper(*args, **kwargs): for entry in attr(*args, **kwargs): yield self._wrap(entry) return wrapper if inspect.isgeneratorfunction(attr) else wrapper() elif inspect.isfunction(attr): def wrapper(*args, **kwargs): return self._wrap(attr(*args, **kwargs)) return wrapper else: return self._wrap(attr)
def _call_handler(self, *, handler, args, kwargs, loop=None, start=True, executor=None): if loop is None: loop = asyncio.get_event_loop() if (inspect.iscoroutinefunction(handler) or inspect.isgeneratorfunction(handler)): # Get a coro/future f = handler(*args, **kwargs) else: # run_in_executor doesn't support kwargs handler = functools.partial(handler, *args, **kwargs) # Get result/coro/future f = loop.run_in_executor(executor, handler) if start: # Wrap future in a task, schedule it for execution f = asyncio.ensure_future(f, loop=loop) # Return a coro that awaits our existing future return self._result_tuple(handler, f)
def _call_meth(self, match_info, name): # call meth with variable segments of the request as arguments. meth = getattr(self, name) if (not asyncio.iscoroutinefunction(meth) and not inspect.isgeneratorfunction(meth)): meth = asyncio.coroutine(meth) # get variable segments for the current provider. var = {k: v for k, v in match_info.items() if not k.startswith('_')} # get method signature and apply variable segments req, _, kw, _ = inspect.getargspec(getattr(meth, '__wrapped__', meth)) if kw is None: rv = yield from meth(**{k: v for k, v in var.items() if k in req}) else: rv = yield from meth(**var) # any kerword arguments is accepted return rv
def __init__(self, fn, callback=None): # this will except if <fn> is not a coroutine or generator function # that can yield assert inspect.isgeneratorfunction(fn) or hasattr(fn, "__call__") Task._next_id += 1 self.id = Task._next_id self.fn = fn() self.state = None if callable(callback) and not inspect.getargspec(callback).args: def cb(_): callback() self.callback = cb else: self.callback = callback
def default(self, obj): if hasattr(obj, "to_json"): return self.default(obj.to_json()) elif hasattr(obj, "__dict__"): data = dict( (key, value) for key, value in inspect.getmembers(obj) if not key.startswith("__") and not inspect.isabstract(value) and not inspect.isbuiltin(value) and not inspect.isfunction(value) and not inspect.isgenerator(value) and not inspect.isgeneratorfunction(value) and not inspect.ismethod(value) and not inspect.ismethoddescriptor(value) and not inspect.isroutine(value) ) return self.default(data) return obj
def _run_request(self, request, where, cpu, gen, *args, **kwargs): """Internal use only. """ if isinstance(gen, str): name = gen else: name = gen.func_name if name in self._xfer_funcs: code = None else: # if not inspect.isgeneratorfunction(gen): # logger.warning('"%s" is not a valid generator function', name) # raise StopIteration([]) code = inspect.getsource(gen).lstrip() def _run_req(task=None): msg = {'req': 'job', 'auth': self._auth, 'job': _DispycosJob_(request, task, name, where, cpu, code, args, kwargs)} if (yield self.scheduler.deliver(msg, timeout=MsgTimeout)) == 1: reply = yield task.receive() if isinstance(reply, Task): if self.status_task: msg = DispycosTaskInfo(reply, args, kwargs, time.time()) self.status_task.send(DispycosStatus(Scheduler.TaskCreated, msg)) if not request.endswith('async'): reply = yield task.receive() else: reply = None raise StopIteration(reply) yield Task(_run_req).finish()
def __init__(self, method, name=None): """'method' must be generator method; this is used to create tasks. If 'name' is not given, method's function name is used for registering. """ if not inspect.isgeneratorfunction(method): raise RuntimeError('RTI method must be generator function') self._method = method if name: self._name = name else: self._name = method.__name__ if not RTI._pycos: RTI._pycos = Pycos.instance() self._location = None self._mid = None
def register(self): """RTI must be registered so it can be located. """ if self._location: return -1 if not inspect.isgeneratorfunction(self._method): return -1 RTI._pycos._lock.acquire() if RTI._pycos._rtis.get(self._name, None) is None: RTI._pycos._rtis[self._name] = self RTI._pycos._lock.release() return 0 else: RTI._pycos._lock.release() return -1
def __get_generator(task, *args, **kwargs): if args: target = args[0] args = args[1:] else: target = kwargs.pop('target', None) args = kwargs.pop('args', ()) kwargs = kwargs.pop('kwargs', kwargs) if not inspect.isgeneratorfunction(target): raise Exception('%s is not a generator!' % target.__name__) if target.func_defaults and \ 'task' in target.func_code.co_varnames[:target.func_code.co_argcount][-len(target.func_defaults):]: kwargs['task'] = task return target(*args, **kwargs)
def _run_request(self, request, where, cpu, gen, *args, **kwargs): """Internal use only. """ if isinstance(gen, str): name = gen else: name = gen.__name__ if name in self._xfer_funcs: code = None else: # if not inspect.isgeneratorfunction(gen): # logger.warning('"%s" is not a valid generator function', name) # raise StopIteration([]) code = inspect.getsource(gen).lstrip() def _run_req(task=None): msg = {'req': 'job', 'auth': self._auth, 'job': _DispycosJob_(request, task, name, where, cpu, code, args, kwargs)} if (yield self.scheduler.deliver(msg, timeout=MsgTimeout)) == 1: reply = yield task.receive() if isinstance(reply, Task): if self.status_task: msg = DispycosTaskInfo(reply, args, kwargs, time.time()) self.status_task.send(DispycosStatus(Scheduler.TaskCreated, msg)) if not request.endswith('async'): reply = yield task.receive() else: reply = None raise StopIteration(reply) yield Task(_run_req).finish()
def __get_generator(task, *args, **kwargs): if args: target = args[0] args = args[1:] else: target = kwargs.pop('target', None) args = kwargs.pop('args', ()) kwargs = kwargs.pop('kwargs', kwargs) if not inspect.isgeneratorfunction(target): raise Exception('%s is not a generator!' % target.__name__) if target.__defaults__ and \ 'task' in target.__code__.co_varnames[:target.__code__.co_argcount][-len(target.__defaults__):]: kwargs['task'] = task return target(*args, **kwargs)
def is_awaitable(obj): # There is no single method which can answer in any case, should wait or not - so need to create one # for the suspected cases : func, coro, gen-coro, future, # class with sync __call__, class with async __call__, # sync method, async method if inspect.isawaitable(obj) or inspect.iscoroutinefunction(obj) or inspect.iscoroutine(obj): return True elif inspect.isgeneratorfunction(obj): return True elif CallChain.is_user_defined_class(obj): if hasattr(obj, '__call__'): return CallChain.is_awaitable(obj.__call__) return False else: return False
def pre_process_extensions(self, extensions, request, action_args): # List of callables for post-processing extensions post = [] for ext in extensions: if inspect.isgeneratorfunction(ext): response = None # If it's a generator function, the part before the # yield is the preprocessing stage try: with ResourceExceptionHandler(): gen = ext(req=request, **action_args) response = next(gen) except Fault as ex: response = ex # We had a response... if response: return response, [] # No response, queue up generator for post-processing post.append(gen) else: # Regular functions only perform post-processing post.append(ext) # Run post-processing in the reverse order return None, reversed(post)
def isgenerator(o): if isinstance(o, UnboundMethod): o = o._func return inspect.isgeneratorfunction(o) or inspect.isgenerator(o)
def _validContext(func): # Defined inside USBContext so we can access "self.__*". @contextlib.contextmanager def refcount(self): with self.__context_cond: if not self.__context_p and self.__auto_open: # BBB warnings.warn( 'Use "with USBContext() as context:" for safer cleanup' ' on interpreter shutdown. See also USBContext.open().', DeprecationWarning, ) self.open() self.__context_refcount += 1 try: yield finally: with self.__context_cond: self.__context_refcount -= 1 if not self.__context_refcount: self.__context_cond.notifyAll() if inspect.isgeneratorfunction(func): def wrapper(self, *args, **kw): with refcount(self): if self.__context_p: # pylint: disable=not-callable for value in func(self, *args, **kw): # pylint: enable=not-callable yield value else: def wrapper(self, *args, **kw): with refcount(self): if self.__context_p: # pylint: disable=not-callable return func(self, *args, **kw) # pylint: enable=not-callable functools.update_wrapper(wrapper, func) return wrapper # pylint: enable=no-self-argument,protected-access
def _process_async_rep_event(self, socket, channel, data): """ Process a ASYNC_REP socket's event. Parameters ---------- socket : zmq.Socket Socket that generated the event. channel : AgentChannel AgentChannel associated with the socket that generated the event. data : bytes Data received on the socket. """ message = deserialize_message(message=data, serializer=channel.serializer) address_uuid, request_uuid, data, address = message client_address = address.twin() if not self.registered(client_address): self.connect(address) handler = self.handler[socket] is_generator = inspect.isgeneratorfunction(handler) if is_generator: generator = handler(self, data) reply = next(generator) else: reply = handler(self, data) self.send(client_address, (address_uuid, request_uuid, reply)) if is_generator: execute_code_after_yield(generator)
def _process_sync_pub_event(self, socket, channel, data): """ Process a SYNC_PUB socket's event. Parameters ---------- socket : zmq.Socket Socket that generated the event. channel : AgentChannel AgentChannel associated with the socket that generated the event. data : bytes Data received on the socket. """ message = deserialize_message(message=data, serializer=channel.serializer) address_uuid, request_uuid, data = message handler = self.handler[socket] is_generator = inspect.isgeneratorfunction(handler) if is_generator: generator = handler(self, data) reply = next(generator) else: reply = handler(self, data) message = (address_uuid, request_uuid, reply) self._send_channel_sync_pub(channel=channel, message=message, topic=address_uuid, general=False) if is_generator: execute_code_after_yield(generator)
def add_route(app,fn): method = getattr(fn,'__method__',None) path = getattr(fn,'__route__',None) if path is None or method is None: return ValueError('@get or @post not defined in %s' % str(fn)) if not asyncio.iscoroutinefunction(fn) and not inspect.isgeneratorfunction(fn): fn = asyncio.coroutine(fn) logging.info('add route %s %s => %s(%s)' % (method,path,fn.__name__,','.join(inspect.signature(fn).parameters.keys()))) app.router.add_route(method,path,RequestHandler(app,fn)) #????URL??
def ticker_wrapper(handler): async def decorator(data): currency_pair = data[0] last = data[1] lowest_ask = data[2] highest_bid = data[3] percent_change = data[4] base_volume = data[5] quote_volume = data[6] is_frozen = data[7] day_high = data[8] day_low = data[9] kwargs = { "currency_pair": currency_pair, "last": last, "lowest_ask": lowest_ask, "highest_bid": highest_bid, "percent_change": percent_change, "base_volume": base_volume, "quote_volume": quote_volume, "is_frozen": is_frozen, "day_high": day_high, "day_low": day_low } if inspect.isgeneratorfunction(handler): await handler(**kwargs) else: handler(**kwargs) return decorator
def trades_wrapper(topic, handler): async def decorator(data): for event in data: event["currency_pair"] = topic if inspect.isgeneratorfunction(handler): await handler(**event) else: handler(**event) return decorator
def add_route(app, fn): method = getattr(fn, '__method__', None) path = getattr(fn, '__route__', None) if path is None or method is None: raise ValueError('@Get or @Post not defined in {}.'.format(str(fn))) if not asyncio.iscoroutinefunction(fn) and not inspect.isgeneratorfunction(fn): fn = asyncio.coroutine(fn) logging.info('add route {} {} => {}({})'.format(method, path, fn.__name__, ', '.join(inspect.signature(fn).parameters.keys()))) app.router.add_route(method, path, RequestHandler(app, fn))
def is_cython_or_generator(fn): """Returns whether this function is either a generator function or a Cythonized function.""" if hasattr(fn, '__func__'): fn = fn.__func__ # Class method, static method if inspect.isgeneratorfunction(fn): return True name = type(fn).__name__ return \ name == 'generator' or \ name == 'method_descriptor' or \ name == 'cython_function_or_method' or \ name == 'builtin_function_or_method'
def return_list(func): import inspect from functools import wraps assert inspect.isgeneratorfunction(func) @wraps(func) def wrapped(*args, **kargs): return list(func(*args, **kargs)) return wrapped
def __enqueue(func, pipe, message): assert isinstance(pipe, Pipe) assert isinstance(message, simpype.Message) message.location = pipe message.resource = pipe.resource if inspect.isgeneratorfunction(func): result = yield pipe.env.process(func(pipe, message)) else: result = func(pipe, message) pipe.full() return result
