Skip to main content
aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authoraleherbau2011-03-31 14:43:18 +0000
committeraleherbau2011-03-31 14:43:18 +0000
commit60783280bd63e742b2b70cdbdff8d93758ffe820 (patch)
treef056f0359bb04fc28ab3590b3dd412d4278b31b3 /python/src/tcf/channel/AbstractChannel.py
parentab638aab968e71abfb8fde370b25c211434a2acd (diff)
downloadorg.eclipse.tcf-60783280bd63e742b2b70cdbdff8d93758ffe820.tar.gz
org.eclipse.tcf-60783280bd63e742b2b70cdbdff8d93758ffe820.tar.xz
org.eclipse.tcf-60783280bd63e742b2b70cdbdff8d93758ffe820.zip
Initial commit of TCF python scripting support.
Diffstat (limited to 'python/src/tcf/channel/AbstractChannel.py')
-rw-r--r--python/src/tcf/channel/AbstractChannel.py753
1 files changed, 753 insertions, 0 deletions
diff --git a/python/src/tcf/channel/AbstractChannel.py b/python/src/tcf/channel/AbstractChannel.py
new file mode 100644
index 000000000..f343d58ef
--- /dev/null
+++ b/python/src/tcf/channel/AbstractChannel.py
@@ -0,0 +1,753 @@
+# *******************************************************************************
+# * Copyright (c) 2011 Wind River Systems, Inc. and others.
+# * All rights reserved. This program and the accompanying materials
+# * are made available under the terms of the Eclipse Public License v1.0
+# * which accompanies this distribution, and is available at
+# * http://www.eclipse.org/legal/epl-v10.html
+# *
+# * Contributors:
+# * Wind River Systems - initial API and implementation
+# *******************************************************************************
+
+import sys, threading, time, types
+from exceptions import Exception, IOError
+from tcf import protocol, transport, services, peer, errors
+from tcf.services import locator
+from tcf.channel import STATE_CLOSED, STATE_OPEN, STATE_OPENING
+from tcf.channel import Token, fromJSONSequence, toJSONSequence
+
+EOS = -1 # End Of Stream
+EOM = -2 # End Of Message
+
+
+class Message(object):
+ def __init__(self, typeCode):
+ if type(typeCode) is types.IntType:
+ typeCode = chr(typeCode)
+ self.type = typeCode
+ self.service = None
+ self.name = None
+ self.data = None
+ self.is_canceled = None
+ self.is_sent = None
+ self.token = None
+ self.trace = ()
+ def __str__(self):
+ return "%s %s %s" % (self.type, self.service, self.name)
+
+class ReaderThread(threading.Thread):
+ def __init__(self, channel, handleInput):
+ super(ReaderThread, self).__init__(name="TCF Reader Thread")
+ self.channel = channel
+ self.handleInput = handleInput
+ self.buf = bytearray()
+ self.eos_err_report = None
+ self.daemon = True
+
+ def error(self):
+ raise IOError("Protocol syntax error")
+
+ def readBytes(self, end, buf=None):
+ if buf is None:
+ buf = bytearray()
+ while True:
+ n = self.channel.read()
+ if n <= 0:
+ if n == end: break
+ if n == EOM: raise IOError("Unexpected end of message")
+ if n < 0: raise IOError("Communication channel is closed by remote peer")
+ buf.append(n)
+ return buf
+
+ def readString(self):
+ del self.buf[:]
+ bytes = self.readBytes(0, self.buf)
+ return bytes.decode("UTF8")
+
+ def run(self):
+ try:
+ while True:
+ n = self.channel.read()
+ if n == EOM: continue
+ if n == EOS:
+ try:
+ self.eos_err_report = self.readBytes(EOM)
+ reportLen = len(self.eos_err_report)
+ if reportLen == 0 or reportLen == 1 and self.eos_err_report[0] == 0:
+ self.eos_err_report = None
+ except:
+ pass
+ break
+ msg = Message(n)
+ if self.channel.read() != 0: self.error()
+ typeCode = msg.type
+ if typeCode == 'C':
+ msg.token = Token(self.readBytes(0))
+ msg.service = self.readString()
+ msg.name = self.readString()
+ msg.data = self.readBytes(EOM)
+ elif typeCode in 'PRN':
+ msg.token = Token(self.readBytes(0))
+ msg.data = self.readBytes(EOM)
+ elif typeCode == 'E':
+ msg.service = self.readString()
+ msg.name = self.readString()
+ msg.data = self.readBytes(EOM)
+ elif typeCode == 'F':
+ msg.data = self.readBytes(EOM)
+ else:
+ self.error()
+ protocol.invokeLater(self.handleInput, msg)
+ delay = self.channel.local_congestion_level
+ if delay > 0: time.sleep(delay / 1000.0)
+ protocol.invokeLater(self.handleEOS)
+ except Exception as x:
+ try:
+ x.tb = sys.exc_info()[2]
+ protocol.invokeLater(self.channel.terminate, x)
+ except:
+ # TCF event dispatcher has shut down
+ pass
+
+ def handleEOS(self):
+ if not self.channel.out_tokens and not self.eos_err_report:
+ self.channel.close()
+ else:
+ x = IOError("Communication channel is closed by remote peer")
+ if self.eos_err_report:
+ try:
+ args = fromJSONSequence(self.eos_err_report)
+ if len(args) > 0 and args[0] is not None:
+ x.caused_by = Exception(errors.toErrorString(args[0]))
+ except IOError:
+ pass
+ self.channel.terminate(x)
+
+
+class AbstractChannel(object):
+ """
+ AbstractChannel implements communication link connecting two end points (peers).
+ The channel asynchronously transmits messages: commands, results and events.
+
+ Clients can subclass AbstractChannel to support particular transport (wire) protocol.
+ Also, see StreamChannel for stream oriented transport protocols.
+ """
+
+ def __init__(self, remote_peer, local_peer=None):
+ self.remote_peer = remote_peer
+ self.local_peer = local_peer # TODO
+ self.inp_thread = ReaderThread(self, self.__handleInput)
+ self.out_thread = threading.Thread(target=self.__write_output,name="TCF Channel Transmitter")
+ self.out_thread.daemon = True
+ self.out_tokens = {}
+ self.out_queue = []
+ self.out_lock = threading.Condition()
+ self.pending_command_limit = 32
+ self.remote_service_by_class = {}
+ self.local_service_by_name = {}
+ self.remote_service_by_name = {}
+ self.channel_listeners = []
+ self.event_listeners = {}
+ self.command_servers = {}
+ self.redirect_queue = []
+ self.redirect_command = None
+ self.notifying_channel_opened = False
+ self.registered_with_trasport = False
+ self.state = STATE_OPENING
+ self.proxy = None
+ self.zero_copy = False
+
+ self.local_congestion_level = -100
+ self.remote_congestion_level = -100
+ self.local_congestion_cnt = 0
+ self.local_congestion_time = 0
+ self.local_service_by_class = {}
+ self.trace_listeners = []
+
+ def __write_output(self):
+ try:
+ while True:
+ msg = None
+ last = False
+ with self.out_lock:
+ while len(self.out_queue) == 0:
+ self.out_lock.wait()
+ msg = self.out_queue.pop(0)
+ if not msg: break
+ last = len(self.out_queue) == 0
+ if msg.is_canceled:
+ if last: self.flush()
+ continue
+ msg.is_sent = True
+ if msg.trace:
+ protocol.invokeLater(self.__traceMessageSent, msg)
+ self.write(msg.type)
+ self.write(0)
+ if msg.token:
+ self.write(msg.token.id)
+ self.write(0)
+ if msg.service:
+ self.write(msg.service.encode("UTF8"))
+ self.write(0)
+ if msg.name:
+ self.write(msg.name.encode("UTF8"))
+ self.write(0)
+ if msg.data:
+ self.write(msg.data)
+ self.write(EOM)
+ delay = 0
+ level = self.remote_congestion_level
+ if level > 0: delay = level * 10
+ if last or delay > 0: self.flush()
+ if delay > 0: time.sleep(delay / 1000.0)
+ #else yield()
+ self.write(EOS)
+ self.write(EOM)
+ self.flush()
+ except Exception as x:
+ try:
+ protocol.invokeLater(self.terminate, x)
+ except:
+ # TCF event dispatcher has shut down
+ pass
+
+ def __traceMessageSent(self, m):
+ for l in m.trace:
+ try:
+ id = None
+ if m.token is not None:
+ id = m.token.getID()
+ l.onMessageSent(m.type, id, m.service, m.name, m.data)
+ except Exception as x:
+ protocol.log("Exception in channel listener", x)
+
+ def start(self):
+ assert protocol.isDispatchThread()
+ protocol.invokeLater(self.__initServices)
+ self.inp_thread.start()
+ self.out_thread.start()
+
+ def __initServices(self):
+ try:
+ if self.proxy: return
+ if self.state == STATE_CLOSED: return
+ services.onChannelCreated(self, self.local_service_by_name)
+ self.__makeServiceByClassMap(self.local_service_by_name, self.local_service_by_class)
+ args = self.local_service_by_name.keys()
+ self.sendEvent(protocol.getLocator(), "Hello", toJSONSequence((args,)))
+ except IOError as x:
+ self.terminate(x)
+
+ def redirect_id(self, peer_id):
+ """
+ Redirect this channel to given peer using this channel remote peer locator service as a proxy.
+ @param peer_id - peer that will become new remote communication endpoint of this channel
+ """
+ map = {}
+ map[peer.ATTR_ID] = peer_id
+ self.redirect(map)
+
+ def redirect(self, peer_attrs):
+ """
+ Redirect this channel to given peer using this channel remote peer locator service as a proxy.
+ @param peer_attrs - peer that will become new remote communication endpoint of this channel
+ """
+ channel = self
+ assert protocol.isDispatchThread()
+ if self.state == STATE_OPENING:
+ self.redirect_queue.append(peer_attrs)
+ else:
+ assert self.state == STATE_OPEN
+ assert self.redirect_command is None
+ try:
+ l = self.remote_service_by_class.get(locator.LocatorService)
+ if not l: raise IOError("Cannot redirect channel: peer " +
+ self.remote_peer.getID() + " has no locator service")
+ peer_id = peer_attrs.get(peer.ATTR_ID)
+ if peer_id and len(peer_attrs) == 1:
+ peer = l.getPeers().get(peer_id)
+ if not peer:
+ # Peer not found, must wait for a while until peer is discovered or time out
+ class Callback(object):
+ found = None
+ def __call__(self):
+ if self.found: return
+ self.channel.terminate(Exception("Peer " + peer_id + " not found"))
+ cb = Callback()
+ protocol.invokeLaterWithDelay(locator.DATA_RETENTION_PERIOD / 3, cb)
+ class Listener(locator.LocatorListener):
+ def peerAdded(self, peer):
+ if peer.getID() == peer_id:
+ cb.found = True
+ channel.state = STATE_OPEN
+ l.removeListener(self)
+ channel.redirect_id(peer_id)
+ l.addListener(Listener())
+ else:
+ class DoneRedirect(locator.DoneRedirect):
+ def doneRedirect(self, token, exc):
+ assert channel.redirect_command is token
+ channel.redirect_command = None
+ if channel.state != STATE_OPENING: return
+ if exc: channel.terminate(exc)
+ channel.remote_peer = peer
+ channel.remote_service_by_class.clear()
+ channel.remote_service_by_name.clear()
+ channel.event_listeners.clear()
+ self.redirect_command = l.redirect(peer_id, DoneRedirect())
+ else:
+ class TransientPeer(peer.TransientPeer):
+ def __init__(self, peer_attrs, parent):
+ super(TransientPeer, self).__init__(peer_attrs)
+ self.parent = parent
+ def openChannel(self):
+ c = self.parent.openChannel()
+ c.redirect(peer_attrs)
+ class DoneRedirect(locator.DoneRedirect):
+ def doneRedirect(self, token, exc):
+ assert channel.redirect_command is token
+ channel.redirect_command = None
+ if channel.state != STATE_OPENING: return
+ if exc: channel.terminate(exc)
+ parent = channel.remote_peer
+ channel.remote_peer = TransientPeer(peer_attrs, parent)
+ channel.remote_service_by_class.clear()
+ channel.remote_service_by_name.clear()
+ channel.event_listeners.clear()
+ self.redirect_command = l.redirect(peer_attrs, DoneRedirect())
+ self.state = STATE_OPENING
+ except Exception as x:
+ self.terminate(x)
+
+ def __makeServiceByClassMap(self, by_name, by_class):
+ for service in by_name.values():
+ for clazz in service.__class__.__bases__:
+ if clazz == services.Service: continue
+ # TODO
+ # if (!IService.class.isAssignableFrom(fs)) continue
+ by_class[clazz] = service
+
+ def getState(self):
+ return self.state
+
+ def addChannelListener(self, listener):
+ assert protocol.isDispatchThread()
+ assert listener
+ self.channel_listeners.append(listener)
+
+ def removeChannelListener(self, listener):
+ assert protocol.isDispatchThread()
+ self.channel_listeners.remove(listener)
+
+ def addTraceListener(self, listener):
+ if self.trace_listeners is None:
+ self.trace_listeners = []
+ else:
+ self.trace_listeners = self.trace_listeners[:]
+ self.trace_listeners.append(listener)
+
+ def removeTraceListener(self, listener):
+ self.trace_listeners = self.trace_listeners[:]
+ self.trace_listeners.remove(listener)
+ if len(self.trace_listeners) == 0: self.trace_listeners = None
+
+ def addEventListener(self, service, listener):
+ assert protocol.isDispatchThread()
+ svc_name = str(service)
+ listener.svc_name = svc_name
+ list = self.event_listeners.get(svc_name) or []
+ list.append(listener)
+ self.event_listeners[svc_name] = list
+
+ def removeEventListener(self, service, listener):
+ assert protocol.isDispatchThread()
+ svc_name = str(service)
+ list = self.event_listeners.get(svc_name)
+ if not list: return
+ for i in range(len(list)):
+ if list[i] is listener:
+ if len(list) == 1:
+ del self.event_listeners[svc_name]
+ else:
+ del list[i]
+ return
+
+ def addCommandServer(self, service, listener):
+ assert protocol.isDispatchThread()
+ svc_name = str(service)
+ if self.command_servers.get(svc_name):
+ raise Exception("Only one command server per service is allowed")
+ self.command_servers[svc_name] = listener
+
+ def removeCommandServer(self, service, listener):
+ assert protocol.isDispatchThread()
+ svc_name = str(service)
+ if self.command_servers.get(svc_name) is not listener:
+ raise Exception("Invalid command server")
+ del self.command_servers[svc_name]
+
+ def close(self):
+ assert protocol.isDispatchThread()
+ if self.state == STATE_CLOSED: return
+ try:
+ self.__sendEndOfStream(10000)
+ self._close(None)
+ except Exception as x:
+ self._close(x)
+
+ def terminate(self, error):
+ assert protocol.isDispatchThread()
+ if self.state == STATE_CLOSED: return
+ try:
+ self.__sendEndOfStream(500)
+ except Exception as x:
+ if not error: error = x
+ self._close(error)
+
+ def __sendEndOfStream(self, timeout):
+ with self.out_lock:
+ del self.out_queue[:]
+ self.out_queue.append(None)
+ self.out_lock.notify()
+ self.out_thread.join(timeout)
+
+ def _close(self, error):
+ assert self.state != STATE_CLOSED
+ self.state = STATE_CLOSED
+ # Closing channel underlying streams can block for a long time,
+ # so it needs to be done by a background thread.
+ thread = threading.Thread(target=self.stop, name="TCF Channel Cleanup")
+ thread.daemon = True
+ thread.start()
+ if error and isinstance(self.remote_peer, peer.AbstractPeer):
+ self.remote_peer.onChannelTerminated()
+ if self.registered_with_trasport:
+ self.registered_with_trasport = False
+ transport.channelClosed(self, error)
+ if self.proxy:
+ try:
+ self.proxy.onChannelClosed(error)
+ except Exception as x:
+ protocol.log("Exception in channel listener", x)
+ channel = self
+ class Runnable(object):
+ def __call__(self):
+ if channel.out_tokens:
+ x = None
+ if isinstance(error, Exception): x = error
+ elif error: x = Exception(error)
+ else: x = IOError("Channel is closed")
+ for msg in channel.out_tokens.values():
+ try:
+ s = str(msg)
+ if len(s) > 72: s = s[:72] + "...]"
+ y = IOError("Command " + s + " aborted")
+# y.initCause(x)
+ msg.token.getListener().terminated(msg.token, y)
+ except Exception as e:
+ protocol.log("Exception in command listener", e)
+ channel.out_tokens.clear()
+ if channel.channel_listeners:
+ for l in channel.channel_listeners:
+ if not l: break
+ try:
+ l.onChannelClosed(error)
+ except Exception as x:
+ protocol.log("Exception in channel listener", x)
+ elif error:
+ protocol.log("TCF channel terminated", error)
+ if channel.trace_listeners:
+ for l in channel.trace_listeners:
+ try:
+ l.onChannelClosed(error)
+ except Exception as x:
+ protocol.log("Exception in channel listener", x)
+ protocol.invokeLater(Runnable())
+
+ def getCongestion(self):
+ assert protocol.isDispatchThread()
+ level = len(self.out_tokens) * 100 / self.pending_command_limit - 100
+ if self.remote_congestion_level > level: level = self.remote_congestion_level
+ if level > 100: level = 100
+ return level
+
+ def getLocalPeer(self):
+ assert protocol.isDispatchThread()
+ return self.local_peer
+
+ def getRemotePeer(self):
+ assert protocol.isDispatchThread()
+ return self.remote_peer
+
+ def getLocalServices(self):
+ assert protocol.isDispatchThread()
+ assert self.state != STATE_OPENING
+ return self.local_service_by_name.keys()
+
+ def getRemoteServices(self):
+ assert protocol.isDispatchThread()
+ assert self.state != STATE_OPENING
+ return self.remote_service_by_name.keys()
+
+ def getLocalService(self, cls_or_name):
+ assert protocol.isDispatchThread()
+ assert self.state != STATE_OPENING
+ if type(cls_or_name) == types.StringType:
+ return self.local_service_by_name.get(cls_or_name)
+ else:
+ return self.local_service_by_class.get(cls_or_name)
+
+ def getRemoteService(self, cls_or_name):
+ assert protocol.isDispatchThread()
+ assert self.state != STATE_OPENING
+ if type(cls_or_name) == types.StringType:
+ return self.remote_service_by_name.get(cls_or_name)
+ else:
+ return self.remote_service_by_class.get(cls_or_name)
+
+ def setServiceProxy(self, service_interface, service_proxy):
+ if not self.notifying_channel_opened: raise Exception("setServiceProxe() can be called only from channel open call-back")
+ if not isinstance(self.remote_service_by_name.get(service_proxy.getName()), services.GenericProxy): raise Exception("Proxy already set")
+ if self.remote_service_by_class.get(service_interface): raise Exception("Proxy already set")
+ self.remote_service_by_class[service_interface] = service_proxy
+ self.remote_service_by_name[service_proxy.getName()] = service_proxy
+
+ def setProxy(self, proxy, services):
+ self.proxy = proxy
+ self.sendEvent(protocol.getLocator(), "Hello", toJSONSequence((services,)))
+ self.local_service_by_class.clear()
+ self.local_service_by_name.clear()
+
+ def addToOutQueue(self, msg):
+ msg.trace = self.trace_listeners
+ with self.out_lock:
+ self.out_queue.append(msg)
+ self.out_lock.notify()
+
+ def sendCommand(self, service, name, args, listener):
+ assert protocol.isDispatchThread()
+ if self.state == STATE_OPENING: raise Exception("Channel is waiting for Hello message")
+ if self.state == STATE_CLOSED: raise Exception("Channel is closed")
+ msg = Message('C')
+ msg.service = str(service)
+ msg.name = name
+ msg.data = args
+ channel = self
+ class CancelableToken(Token):
+ def __init__(self, listener):
+ super(CancelableToken, self).__init__(listener=listener)
+ def cancel(self):
+ assert protocol.isDispatchThread()
+ if channel.state != STATE_OPEN: return False
+ with channel.out_lock:
+ if msg.is_sent: return False
+ msg.is_canceled = True
+ del channel.out_tokens[msg.token.getID()]
+ return True
+ token = CancelableToken(listener)
+ msg.token = token
+ self.out_tokens[token.getID()] = msg
+ self.addToOutQueue(msg)
+ return token
+
+ def sendProgress(self, token, results):
+ assert protocol.isDispatchThread()
+ if self.state != STATE_OPEN: raise Exception("Channel is closed")
+ msg = Message('P')
+ msg.data = results
+ msg.token = token
+ self.addToOutQueue(msg)
+
+ def sendResult(self, token, results):
+ assert protocol.isDispatchThread()
+ if self.state != STATE_OPEN: raise Exception("Channel is closed")
+ msg = Message('R')
+ msg.data = results
+ msg.token = token
+ self.addToOutQueue(msg)
+
+ def rejectCommand(self, token):
+ assert protocol.isDispatchThread()
+ if self.state != STATE_OPEN: raise Exception("Channel is closed")
+ msg = Message('N')
+ msg.token = token
+ self.addToOutQueue(msg)
+
+ def sendEvent(self, service, name, args):
+ assert protocol.isDispatchThread()
+ if not (self.state == STATE_OPEN or self.state == STATE_OPENING and isinstance(service, locator.LocatorService)):
+ raise Exception("Channel is closed")
+ msg = Message('E')
+ msg.service = str(service)
+ msg.name = name
+ msg.data = args
+ self.addToOutQueue(msg)
+
+ def isZeroCopySupported(self):
+ return self.zero_copy
+
+ def __traceMessageReceived(self, m):
+ for l in self.trace_listeners:
+ try:
+ id = None
+ if m.token is not None:
+ id = m.token.getID()
+ l.onMessageReceived(m.type, id, m.service, m.name, m.data)
+ except Exception as x:
+ protocol.log("Exception in channel listener", x)
+
+ def __handleInput(self, msg):
+ assert protocol.isDispatchThread()
+ if self.state == STATE_CLOSED: return
+ if self.trace_listeners:
+ self.__traceMessageReceived(msg)
+ try:
+ token = None
+ typeCode = msg.type
+ if typeCode in 'PRN':
+ token_id = msg.token.getID()
+ cmd = self.out_tokens.get(token_id)
+ if cmd is None: raise Exception("Invalid token received: " + token_id)
+ if typeCode != 'P':
+ del self.out_tokens[token_id]
+ token = cmd.token
+ if typeCode == 'C':
+ if self.state == STATE_OPENING:
+ raise IOError("Received command " + msg.service + "." + msg.name + " before Hello message")
+ if self.proxy:
+ self.proxy.onCommand(msg.token, msg.service, msg.name, msg.data)
+ else:
+ token = msg.token
+ cmds = self.command_servers.get(msg.service)
+ if cmds:
+ cmds.command(token, msg.name, msg.data)
+ else:
+ self.rejectCommand(token)
+ elif typeCode == 'P':
+ token.getListener().progress(token, msg.data)
+ self.__sendCongestionLevel()
+ elif typeCode == 'R':
+ token.getListener().result(token, msg.data)
+ self.__sendCongestionLevel()
+ elif typeCode == 'N':
+ token.getListener().terminated(token, errors.ErrorReport(
+ "Command is not recognized", errors.TCF_ERROR_INV_COMMAND))
+ elif typeCode == 'E':
+ hello = msg.service == locator.NAME and msg.name == "Hello"
+ if hello:
+ self.remote_service_by_name.clear()
+ self.remote_service_by_class.clear()
+ data = fromJSONSequence(msg.data)[0]
+ services.onChannelOpened(self, data, self.remote_service_by_name)
+ self.__makeServiceByClassMap(self.remote_service_by_name, self.remote_service_by_class)
+ self.zero_copy = self.remote_service_by_name.has_key("ZeroCopy")
+ if self.proxy and self.state == STATE_OPEN:
+ self.proxy.onEvent(msg.service, msg.name, msg.data)
+ elif hello:
+ assert self.state == STATE_OPENING
+ self.state = STATE_OPEN
+ assert self.redirect_command is None
+ if self.redirect_queue:
+ self.redirect(self.redirect_queue.pop(0))
+ else:
+ self.notifying_channel_opened = True
+ if not self.registered_with_trasport:
+ transport.channelOpened(self)
+ self.registered_with_trasport = True
+ for l in self.channel_listeners:
+ if not l: break
+ try:
+ l.onChannelOpened()
+ except Exception as x:
+ protocol.log("Exception in channel listener", x)
+ self.notifying_channel_opened = False
+ else:
+ list = self.event_listeners.get(msg.service)
+ if list:
+ for l in list:
+ l.event(msg.name, msg.data)
+ self.__sendCongestionLevel()
+ elif typeCode == 'F':
+ len = len(msg.data)
+ if len > 0 and msg.data[len - 1] == '\0': len -= 1
+ self.remote_congestion_level = int(msg.data)
+ else:
+ assert False
+ except Exception as x:
+ x.tb = sys.exc_info()[2]
+ self.terminate(x)
+
+ def __sendCongestionLevel(self):
+ self.local_congestion_cnt += 1
+ if self.local_congestion_cnt < 8: return
+ self.local_congestion_cnt = 0
+ if self.state != STATE_OPEN: return
+ timeVal = int(time.time())
+ if timeVal - self.local_congestion_time < 500: return
+ assert protocol.isDispatchThread()
+ level = protocol.getCongestionLevel()
+ if level == self.local_congestion_level: return
+ i = (level - self.local_congestion_level) / 8
+ if i != 0: level = self.local_congestion_level + i
+ self.local_congestion_time = timeVal
+ with self.out_lock:
+ msg = None
+ if self.out_queue:
+ msg = self.out_queue[0]
+ if msg is None or msg.type != 'F':
+ msg = Message('F')
+ self.out_queue.insert(0, msg)
+ self.out_lock.notify()
+ data = "%i\0" % self.local_congestion_level
+ msg.data = data
+ msg.trace = self.trace_listeners
+ self.local_congestion_level = level
+
+ def read(self):
+ """
+ Read one byte from the channel input stream.
+ @return next data byte or EOS (-1) if end of stream is reached,
+ or EOM (-2) if end of message is reached.
+ @raises IOError
+ """
+ raise NotImplementedError("Abstract method")
+
+ def writeByte(self, n):
+ """
+ Write one byte into the channel output stream.
+ The method argument can be one of two special values:
+ EOS (-1) end of stream marker
+ EOM (-2) end of message marker.
+ The stream can put the byte into a buffer instead of transmitting it right away.
+ @param n - the data byte.
+ @raises IOError
+ """
+ raise NotImplementedError("Abstract method")
+
+ def flush(self):
+ """
+ Flush the channel output stream.
+ All buffered data should be transmitted immediately.
+ @raises IOError
+ """
+ raise NotImplementedError("Abstract method")
+
+ def stop(self):
+ """
+ Stop (close) channel underlying streams.
+ If a thread is blocked by read() or write(), it should be
+ resumed (or interrupted).
+ @raises IOError
+ """
+ raise NotImplementedError("Abstract method")
+
+ def write(self, buf):
+ """
+ Write array of bytes into the channel output stream.
+ The stream can put bytes into a buffer instead of transmitting it right away.
+ @param buf
+ @raises IOError
+ """
+ assert threading.currentThread() == self.out_thread
+ for i in buf:
+ self.writeByte(ord(buf[i]) & 0xff)

Back to the top