Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#!/usr/bin/env python
import pytest
from circuits.core import Component, Event, TimeoutError, handler
class wait(Event):
"""wait Event"""
success = True
class call(Event):
"""call Event"""
success = True
class hello(Event):
"""hello Event"""
success = True
from circuits.core import Component, Event, TimeoutError, handler
class wait(Event):
"""wait Event"""
success = True
class call(Event):
"""call Event"""
success = True
class hello(Event):
"""hello Event"""
success = True
class App(Component):
@handler('wait')
def _on_wait(self, timeout=-1):
result = self.fire(hello())
try:
yield self.wait('hello', timeout=timeout)
except TimeoutError as e:
yield e
else:
yield result
#!/usr/bin/env python
import pytest
from circuits.core import Component, Event, TimeoutError, handler
class wait(Event):
"""wait Event"""
success = True
class call(Event):
"""call Event"""
success = True
class hello(Event):
"""hello Event"""
success = True
class App(Component):
@handler('wait')
def _on_wait(self, timeout=-1):
result = self.fire(hello())
super(Connected, self).__init__(host, port)
class Disconnected(Event):
"""Disconnected Event
This Event is sent when a client has disconnected
@note: This event is for Client Components.
"""
def __init__(self):
"x.__init__(...) initializes x; see x.__class__.__doc__ for signature"
super(Disconnected, self).__init__()
class Read(Event):
"""Read Event
This Event is sent when a client or server connection has read any data.
@note: This event is used for both Client and Server Components.
:param args: Client: (data) Server: (sock, data)
:type tuple: tuple
"""
def __init__(self, *args):
"x.__init__(...) initializes x; see x.__class__.__doc__ for signature"
super(Read, self).__init__(*args)
class Error(Event):
.. note::
- This event is never sent, it is used to close.
- This event is used for both Client and Server Components.
:param args: Client: () Server: (sock)
:type tuple: tuple
"""
def __init__(self, *args):
"x.__init__(...) initializes x; see x.__class__.__doc__ for signature"
super(close, self).__init__(*args)
class ready(Event):
"""ready Event
This Event is used to notify the rest of the system that the underlying
Client or Server Component is ready to begin processing connections or
incoming/outgoing data. (This is triggered as a direct result of having
the capability to support multiple client/server components with a single
poller component instance in a system).
.. note::
This event is used for both Client and Server Components.
:param component: The Client/Server Component that is ready.
:type tuple: Component (Client/Server)
:param bind: The (host, port) the server has bound to.
the capability to support multiple client/server components with a snigle
poller component instnace in a system).
@note: This event is used for both Client and Server Components.
:param component: The Client/Server Component that is ready.
:type tuple: Component (Client/Server)
"""
def __init__(self, component):
"x.__init__(...) initializes x; see x.__class__.__doc__ for signature"
super(Ready, self).__init__(component)
class Closed(Event):
"""Closed Event
This Event is sent when a server has closed it's listening socket.
@note: This event is for Server components.
"""
def __init__(self):
"x.__init__(...) initializes x; see x.__class__.__doc__ for signature"
super(Closed, self).__init__()
###
### Components
###
import select
from collections import deque
try:
import serial
HAS_SERIAL = True
except ImportError:
HAS_SERIAL = False
from circuits.core import Event, Component
TIMEOUT = 0.2
BUFSIZE = 4096
class EOF(Event): pass
class Seek(Event): pass
class Read(Event): pass
class Close(Event): pass
class Write(Event): pass
class Error(Event): pass
class Opened(Event): pass
class Closed(Event): pass
class File(Component):
def __init__(self, *args, **kwargs):
channel = kwargs.get("channel", File.channel)
super(File, self).__init__(channel=channel)
self.autoclose = kwargs.get("autoclose", True)
self.encoding = kwargs.get("encoding", "utf-8")
Component to daemonize a system into the background and detach it from its
controlling PTY. Supports PID file writing, logging stdin, stdout and stderr
and changing the current working directory.
"""
from os import (
_exit, chdir, closerange, dup2, fork, getpid, remove, setsid, umask,
)
from os.path import isabs
from resource import RLIM_INFINITY, RLIMIT_NOFILE, getrlimit
from sys import stderr, stdin, stdout
from circuits.core import Component, Event, handler
class daemonize(Event):
"""daemonize Event"""
class daemonized(Event):
"""daemonized Event"""
class deletepid(Event):
""""deletepid Event"""
class writepid(Event):
""""writepid Event"""
class Daemon(Component):
import json
from circuits.core import Event
from circuits.six import PY3, text_type
META_EXCLUDE = set(dir(Event()))
META_EXCLUDE.add("node_call_id")
META_EXCLUDE.add("node_sock")
META_EXCLUDE.add("node_without_result")
META_EXCLUDE.add("success_channels")
def load_event(s):
data = json.loads(s)
name = data["name"] if PY3 else data["name"].encode("utf-8")
args = []
for arg in data["args"]:
if isinstance(arg, text_type):
arg = arg.encode("utf-8")
args.append(arg)
"""Read Event
This Event is sent when a client or server connection has read any data.
@note: This event is used for both Client and Server Components.
:param args: Client: (data) Server: (sock, data)
:type tuple: tuple
"""
def __init__(self, *args):
"x.__init__(...) initializes x; see x.__class__.__doc__ for signature"
super(Read, self).__init__(*args)
class Error(Event):
"""Error Event
This Event is sent when a client or server connection has an error.
@note: This event is used for both Client and Server Components.
:param args: Client: (error) Server: (sock, error)
:type tuple: tuple
"""
def __init__(self, *args):
"x.__init__(...) initializes x; see x.__class__.__doc__ for signature"
super(Error, self).__init__(*args)
class Write(Event):