openstack/deb-python-pyngus
Code Issues Proposed changes

Integrate Gordon Sim's PEP8 cleanups

Browse Source
This commit is contained in:
Kenneth Giusti 2014-01-20 14:12:37 -05:00
parent ea3ee25ff4
commit 875bfc9fe8
4 changed files with 137 additions and 163 deletions

129
python/connection.py

@ -1,43 +1,38 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.
# #
# Licensed to the Apache Software Foundation (ASF) under one # Licensed under the Apache License, Version 2.0 (the "License"); you may
# or more contributor license agreements. See the NOTICE file # not use this file except in compliance with the License. You may obtain
# distributed with this work for additional information # a copy of the License at
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# #
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
__all__ = [ __all__ = [
"ConnectionEventHandler", "ConnectionEventHandler",
"Connection" "Connection"
] ]
import logging import logging
from link import ReceiverLink
from link import SenderLink
import proton import proton
from link import SenderLink, ReceiverLink
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
#
# An implementation of an AMQP 1.0 Connection
#
class ConnectionEventHandler(object): class ConnectionEventHandler(object):
""" """An implementation of an AMQP 1.0 Connection."""
"""
def connection_active(self, connection): def connection_active(self, connection):
"""connection handshake completed""" """Connection handshake has completed."""
LOG.debug("connection_active (ignored)") LOG.debug("connection_active (ignored)")
def connection_remote_closed(self, connection, error=None): def connection_remote_closed(self, connection, error=None):
@ -69,14 +64,9 @@ class ConnectionEventHandler(object):
class Connection(object): class Connection(object):
"""
"""
EOS = -1 # indicates 'I/O stream closed' EOS = -1 # indicates 'I/O stream closed'
def __init__(self, container, name, eventHandler=None, properties={}): def __init__(self, container, name, eventHandler=None, properties={}):
"""
"""
self._name = name self._name = name
self._container = container self._container = container
self._handler = eventHandler self._handler = eventHandler
@ -149,15 +139,11 @@ Associate an arbitrary user object with this Connection.
""") """)
def open(self): def open(self):
"""
"""
self._pn_connection.open() self._pn_connection.open()
self._pn_session = self._pn_connection.session() self._pn_session = self._pn_connection.session()
self._pn_session.open() self._pn_session.open()
def close(self, error=None): def close(self, error=None):
"""
"""
for l in self._sender_links.itervalues(): for l in self._sender_links.itervalues():
l.close(error) l.close(error)
for l in self._receiver_links.itervalues(): for l in self._receiver_links.itervalues():
@ -173,8 +159,6 @@ Associate an arbitrary user object with this Connection.
| proton.Endpoint.REMOTE_CLOSED) | proton.Endpoint.REMOTE_CLOSED)
def destroy(self): def destroy(self):
"""
"""
self._sender_links.clear() self._sender_links.clear()
self._receiver_links.clear() self._receiver_links.clear()
self._container._remove_connection(self._name) self._container._remove_connection(self._name)
@ -187,7 +171,7 @@ Associate an arbitrary user object with this Connection.
if pn_link.is_sender and pn_link.name not in self._sender_links: if pn_link.is_sender and pn_link.name not in self._sender_links:
LOG.debug("Remotely initiated Sender needs init") LOG.debug("Remotely initiated Sender needs init")
self._sender_links[pn_link.name] = pn_link self._sender_links[pn_link.name] = pn_link
pn_link.context = None # @todo: update proton.py pn_link.context = None # @todo: update proton.py
req_source = "" req_source = ""
if pn_link.remote_source.dynamic: if pn_link.remote_source.dynamic:
req_source = None req_source = None
@ -200,7 +184,7 @@ Associate an arbitrary user object with this Connection.
elif pn_link.is_receiver and pn_link.name not in self._receiver_links: elif pn_link.is_receiver and pn_link.name not in self._receiver_links:
LOG.debug("Remotely initiated Receiver needs init") LOG.debug("Remotely initiated Receiver needs init")
self._receiver_links[pn_link.name] = pn_link self._receiver_links[pn_link.name] = pn_link
pn_link.context = None # @todo: update proton.py pn_link.context = None # @todo: update proton.py
req_target = "" req_target = ""
if pn_link.remote_target.dynamic: if pn_link.remote_target.dynamic:
req_target = None req_target = None
@ -211,22 +195,24 @@ Associate an arbitrary user object with this Connection.
{"source-address": {"source-address":
pn_link.remote_source.address}) pn_link.remote_source.address})
_REMOTE_REQ = (proton.Endpoint.LOCAL_UNINIT|proton.Endpoint.REMOTE_ACTIVE) _REMOTE_REQ = (proton.Endpoint.LOCAL_UNINIT
_REMOTE_CLOSE = (proton.Endpoint.LOCAL_ACTIVE|proton.Endpoint.REMOTE_CLOSED) | proton.Endpoint.REMOTE_ACTIVE)
_ACTIVE = (proton.Endpoint.LOCAL_ACTIVE|proton.Endpoint.REMOTE_ACTIVE) _REMOTE_CLOSE = (proton.Endpoint.LOCAL_ACTIVE
_CLOSED = (proton.Endpoint.LOCAL_CLOSED|proton.Endpoint.REMOTE_CLOSED) | proton.Endpoint.REMOTE_CLOSED)
_ACTIVE = (proton.Endpoint.LOCAL_ACTIVE | proton.Endpoint.REMOTE_ACTIVE)
_CLOSED = (proton.Endpoint.LOCAL_CLOSED | proton.Endpoint.REMOTE_CLOSED)
_LOCAL_UNINIT = proton.Endpoint.LOCAL_UNINIT
def process(self, now): def process(self, now):
"""
"""
self._next_tick = self._pn_transport.tick(now) self._next_tick = self._pn_transport.tick(now)
# wait until SASL has authenticated # wait until SASL has authenticated
# @todo Server-side SASL # @todo Server-side SASL
if self._pn_sasl: if self._pn_sasl:
if self._pn_sasl.state not in (proton.SASL.STATE_PASS, if self._pn_sasl.state not in (proton.SASL.STATE_PASS,
proton.SASL.STATE_FAIL): proton.SASL.STATE_FAIL):
LOG.debug("SASL in progress. State=%s", str(self._pn_sasl.state)) LOG.debug("SASL in progress. State=%s",
str(self._pn_sasl.state))
self._handler.sasl_step(self, self._pn_sasl) self._handler.sasl_step(self, self._pn_sasl)
return return
@ -240,15 +226,15 @@ Associate an arbitrary user object with this Connection.
self._active = True self._active = True
self._handler.connection_active(self) self._handler.connection_active(self)
pn_session = self._pn_connection.session_head(proton.Endpoint.LOCAL_UNINIT) pn_session = self._pn_connection.session_head(self._LOCAL_UNINIT)
while pn_session: while pn_session:
LOG.debug("Opening remotely initiated session") LOG.debug("Opening remotely initiated session")
pn_session.open() pn_session.open()
pn_session = pn_session.next(proton.Endpoint.LOCAL_UNINIT) pn_session = pn_session.next(self._LOCAL_UNINIT)
pn_link = self._pn_connection.link_head(self._REMOTE_REQ) pn_link = self._pn_connection.link_head(self._REMOTE_REQ)
while pn_link: while pn_link:
next_link = pn_link.next(proton.Endpoint.LOCAL_UNINIT) next_link = pn_link.next(self._LOCAL_UNINIT)
if pn_link.state == self._REMOTE_REQ: if pn_link.state == self._REMOTE_REQ:
self._link_requested(pn_link) self._link_requested(pn_link)
@ -266,11 +252,13 @@ Associate an arbitrary user object with this Connection.
if pn_link.is_sender: if pn_link.is_sender:
sender_link = pn_link.context sender_link = pn_link.context
assert isinstance(sender_link, SenderLink) assert isinstance(sender_link, SenderLink)
sender_link._handler.sender_active(sender_link) if sender_link._handler:
sender_link._handler.sender_active(sender_link)
else: else:
receiver_link = pn_link.context receiver_link = pn_link.context
assert isinstance(receiver_link, ReceiverLink) assert isinstance(receiver_link, ReceiverLink)
receiver_link._handler.receiver_active(receiver_link) if receiver_link._handler:
receiver_link._handler.receiver_active(receiver_link)
pn_link = next_link pn_link = next_link
# process the work queue # process the work queue
@ -298,11 +286,12 @@ Associate an arbitrary user object with this Connection.
if pn_link.context: if pn_link.context:
if pn_link.is_sender: if pn_link.is_sender:
sender_link = pn_link.context sender_link = pn_link.context
sender_link._handler.sender_remote_closed(sender_link, None) handler = pn_link.context._handler
handler.sender_remote_closed(sender_link, None)
else: else:
receiver_link = pn_link.context receiver_link = pn_link.context
receiver_link._handler.receiver_remote_closed(receiver_link, handler = pn_link.context._handler
None) handler.receiver_remote_closed(receiver_link, None)
pn_link = next_link pn_link = next_link
pn_link = self._pn_connection.link_head(self._CLOSED) pn_link = self._pn_connection.link_head(self._CLOSED)
@ -350,8 +339,6 @@ Associate an arbitrary user object with this Connection.
@property @property
def needs_input(self): def needs_input(self):
"""
"""
if self._read_done: if self._read_done:
return self.EOS return self.EOS
capacity = self._pn_transport.capacity() capacity = self._pn_transport.capacity()
@ -361,8 +348,6 @@ Associate an arbitrary user object with this Connection.
return self.EOS return self.EOS
def process_input(self, in_data): def process_input(self, in_data):
"""
"""
c = self.needs_input c = self.needs_input
if c <= 0: if c <= 0:
return c return c
@ -380,8 +365,6 @@ Associate an arbitrary user object with this Connection.
@property @property
def has_output(self): def has_output(self):
"""
"""
if self._write_done: if self._write_done:
return self.EOS return self.EOS
@ -392,8 +375,6 @@ Associate an arbitrary user object with this Connection.
return self.EOS return self.EOS
def output_data(self): def output_data(self):
"""
"""
c = self.has_output c = self.has_output
if c <= 0: if c <= 0:
return None return None
@ -411,7 +392,7 @@ Associate an arbitrary user object with this Connection.
def create_sender(self, source_address, target_address=None, def create_sender(self, source_address, target_address=None,
eventHandler=None, name=None, properties={}): eventHandler=None, name=None, properties={}):
"""Factory for Sender links""" """Factory method for Sender links."""
ident = name or str(source_address) ident = name or str(source_address)
if ident in self._sender_links: if ident in self._sender_links:
raise KeyError("Sender %s already exists!" % ident) raise KeyError("Sender %s already exists!" % ident)
@ -433,11 +414,12 @@ Associate an arbitrary user object with this Connection.
if pn_link.remote_source.dynamic and not source_override: if pn_link.remote_source.dynamic and not source_override:
raise Exception("A source address must be supplied!") raise Exception("A source address must be supplied!")
source_addr = source_override or pn_link.remote_source.address source_addr = source_override or pn_link.remote_source.address
self._sender_links[link_handle] = SenderLink(self, pn_link, link = SenderLink(self, pn_link,
source_addr, source_addr,
pn_link.remote_target.address, pn_link.remote_target.address,
event_handler, properties) event_handler, properties)
return self._sender_links[link_handle] self._sender_links[link_handle] = link
return link
def reject_sender(self, link_handle, reason): def reject_sender(self, link_handle, reason):
pn_link = self._sender_links.get(link_handle) pn_link = self._sender_links.get(link_handle)
@ -449,7 +431,7 @@ Associate an arbitrary user object with this Connection.
def create_receiver(self, target_address, source_address=None, def create_receiver(self, target_address, source_address=None,
eventHandler=None, name=None, properties={}): eventHandler=None, name=None, properties={}):
"""Factory for Receive links""" """Factory method for creating Receive links."""
ident = name or str(target_address) ident = name or str(target_address)
if ident in self._receiver_links: if ident in self._receiver_links:
raise KeyError("Receiver %s already exists!" % ident) raise KeyError("Receiver %s already exists!" % ident)
@ -470,11 +452,12 @@ Associate an arbitrary user object with this Connection.
if pn_link.remote_target.dynamic and not target_override: if pn_link.remote_target.dynamic and not target_override:
raise Exception("A target address must be supplied!") raise Exception("A target address must be supplied!")
target_addr = target_override or pn_link.remote_target.address target_addr = target_override or pn_link.remote_target.address
self._receiver_links[link_handle] = ReceiverLink(self, pn_link, link = ReceiverLink(self, pn_link,
target_addr, target_addr,
pn_link.remote_source.address, pn_link.remote_source.address,
event_handler, properties) event_handler, properties)
return self._receiver_links[link_handle] self._receiver_links[link_handle] = link
return link
def reject_receiver(self, link_handle, reason): def reject_receiver(self, link_handle, reason):
pn_link = self._receiver_links.get(link_handle) pn_link = self._receiver_links.get(link_handle)
@ -484,7 +467,6 @@ Associate an arbitrary user object with this Connection.
# @todo support reason for close # @todo support reason for close
pn_link.close() pn_link.close()
def _remove_sender(self, name): def _remove_sender(self, name):
if name in self._sender_links: if name in self._sender_links:
del self._sender_links[name] del self._sender_links[name]
@ -492,4 +474,3 @@ Associate an arbitrary user object with this Connection.
def _remove_receiver(self, name): def _remove_receiver(self, name):
if name in self._receiver_links: if name in self._receiver_links:
del self._receiver_links[name] del self._receiver_links[name]

40
python/container.py

@ -1,41 +1,38 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.
# #
# Licensed to the Apache Software Foundation (ASF) under one # Licensed under the Apache License, Version 2.0 (the "License"); you may
# or more contributor license agreements. See the NOTICE file # not use this file except in compliance with the License. You may obtain
# distributed with this work for additional information # a copy of the License at
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, # Unless required by applicable law or agreed to in writing, software
# software distributed under the License is distributed on an # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# KIND, either express or implied. See the License for the # License for the specific language governing permissions and limitations
# specific language governing permissions and limitations # under the License.
# under the License.
#
__all__ = [ __all__ = [
"ContainerEventHandler", "ContainerEventHandler",
"Container" "Container"
] ]
import heapq
import logging
import heapq, logging
from connection import Connection from connection import Connection
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
#
# An implementation of an AMQP 1.0 container
#
class ContainerEventHandler(object): class ContainerEventHandler(object):
# @todo - ContainerEventHandler # @todo - ContainerEventHandler
pass pass
# An implementation of an AMQP 1.0 container
class Container(object): class Container(object):
def __init__(self, name, eventHandler=None, properties={}): def __init__(self, name, eventHandler=None, properties={}):
self._name = name self._name = name
@ -93,4 +90,3 @@ class Container(object):
def _remove_connection(self, name): def _remove_connection(self, name):
if name in self._connections: if name in self._connections:
del self._connections[name] del self._connections[name]

68
python/link.py

@ -1,36 +1,36 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.
# #
# Licensed to the Apache Software Foundation (ASF) under one # Licensed under the Apache License, Version 2.0 (the "License"); you may
# or more contributor license agreements. See the NOTICE file # not use this file except in compliance with the License. You may obtain
# distributed with this work for additional information # a copy of the License at
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# #
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
__all__ = [ __all__ = [
"SenderEventHandler", "SenderEventHandler",
"SenderLink", "SenderLink",
"ReceiverEventHandler", "ReceiverEventHandler",
"ReceiverLink" "ReceiverLink"
] ]
import collections, logging import collections
import logging
import proton import proton
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
class _Link(object): class _Link(object):
"""Generic Link base class""" """A generic Link base class."""
def __init__(self, connection, pn_link, def __init__(self, connection, pn_link,
target_address, source_address, target_address, source_address,
handler, properties): handler, properties):
@ -73,8 +73,6 @@ class _Link(object):
return self._name return self._name
def open(self): def open(self):
"""
"""
self._pn_link.open() self._pn_link.open()
def _get_user_context(self): def _get_user_context(self):
@ -91,7 +89,7 @@ Associate an arbitrary application object with this link.
@property @property
def source_address(self): def source_address(self):
"""If link is a sender, source is determined by the local value, else """If link is a sender, source is determined by the local value, else
use the remote use the remote.
""" """
if self._pn_link.is_sender: if self._pn_link.is_sender:
return self._pn_link.source.address return self._pn_link.source.address
@ -101,7 +99,7 @@ Associate an arbitrary application object with this link.
@property @property
def target_address(self): def target_address(self):
"""If link is a receiver, target is determined by the local value, else """If link is a receiver, target is determined by the local value, else
use the remote use the remote.
""" """
if self._pn_link.is_receiver: if self._pn_link.is_receiver:
return self._pn_link.target.address return self._pn_link.target.address
@ -123,9 +121,8 @@ Associate an arbitrary application object with this link.
self._pn_link.context = None self._pn_link.context = None
self._pn_link = None self._pn_link = None
class SenderEventHandler(object): class SenderEventHandler(object):
"""
"""
def sender_active(self, sender_link): def sender_active(self, sender_link):
LOG.debug("sender_active (ignored)") LOG.debug("sender_active (ignored)")
@ -160,23 +157,22 @@ class SenderLink(_Link):
# @todo - think about send-settle-mode configuration # @todo - think about send-settle-mode configuration
def send(self, message, delivery_callback=None, handle=None, deadline=None): def send(self, message, delivery_callback=None,
""" handle=None, deadline=None):
""" self._pending_sends.append((message, delivery_callback, handle,
self._pending_sends.append( (message, delivery_callback, handle, deadline))
deadline) )
# @todo deadline not supported yet # @todo deadline not supported yet
assert not deadline, "send timeout not supported yet!" assert not deadline, "send timeout not supported yet!"
if deadline and (self._next_deadline == 0 or if deadline and (self._next_deadline == 0 or
self._next_deadline > deadline): self._next_deadline > deadline):
self._next_deadline = deadline self._next_deadline = deadline
delivery = self._pn_link.delivery( "tag-%x" % self._next_tag ) delivery = self._pn_link.delivery("tag-%x" % self._next_tag)
self._next_tag += 1 self._next_tag += 1
if delivery.writable: if delivery.writable:
send_req = self._pending_sends.popleft() send_req = self._pending_sends.popleft()
self._write_msg( delivery, send_req ) self._write_msg(delivery, send_req)
return 0 return 0
@ -213,11 +209,12 @@ class SenderLink(_Link):
proton.Disposition.REJECTED: SenderLink.REJECTED, proton.Disposition.REJECTED: SenderLink.REJECTED,
proton.Disposition.RELEASED: SenderLink.RELEASED, proton.Disposition.RELEASED: SenderLink.RELEASED,
proton.Disposition.MODIFIED: SenderLink.MODIFIED, proton.Disposition.MODIFIED: SenderLink.MODIFIED,
} }
if delivery.tag in self._pending_acks: if delivery.tag in self._pending_acks:
if delivery.settled: # remote has finished if delivery.settled: # remote has finished
LOG.debug("delivery updated, remote state=%s", str(delivery.remote_state)) LOG.debug("delivery updated, remote state=%s",
str(delivery.remote_state))
send_req = self._pending_acks.pop(delivery.tag) send_req = self._pending_acks.pop(delivery.tag)
state = _disposition_state_map.get(delivery.remote_state, state = _disposition_state_map.get(delivery.remote_state,
@ -241,7 +238,7 @@ class SenderLink(_Link):
# send_req = (msg, cb, handle, deadline) # send_req = (msg, cb, handle, deadline)
msg = send_req[0] msg = send_req[0]
cb = send_req[1] cb = send_req[1]
self._pn_link.send( msg.encode() ) self._pn_link.send(msg.encode())
self._pn_link.advance() self._pn_link.advance()
if cb: # delivery callback given if cb: # delivery callback given
assert delivery.tag not in self._pending_acks assert delivery.tag not in self._pending_acks
@ -273,7 +270,7 @@ class ReceiverLink(_Link):
target_address, source_address, target_address, source_address,
eventHandler, properties) eventHandler, properties)
self._next_handle = 0 self._next_handle = 0
self._unsettled_deliveries = {} # indexed by handle self._unsettled_deliveries = {} # indexed by handle
# @todo - think about receiver-settle-mode configuration # @todo - think about receiver-settle-mode configuration
@ -327,4 +324,3 @@ class ReceiverLink(_Link):
delivery = self._unsettled_deliveries.pop(handle) delivery = self._unsettled_deliveries.pop(handle)
delivery.update(result) delivery.update(result)
delivery.settle() delivery.settle()

63
python/sockets.py

@ -1,28 +1,28 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.
# #
# Licensed to the Apache Software Foundation (ASF) under one # Licensed under the Apache License, Version 2.0 (the "License"); you may
# or more contributor license agreements. See the NOTICE file # not use this file except in compliance with the License. You may obtain
# distributed with this work for additional information # a copy of the License at
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# #
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
__all__ = [ __all__ = [
"read_socket_input", "read_socket_input",
"write_socket_output" "write_socket_output"
] ]
import socket, errno, logging import errno
import logging
import socket
from connection import Connection from connection import Connection
@ -32,6 +32,7 @@ LOG = logging.getLogger(__name__)
processing. processing.
""" """
def read_socket_input(connection, socket_obj): def read_socket_input(connection, socket_obj):
"""Read from the network layer and processes all data read. Can """Read from the network layer and processes all data read. Can
support both blocking and non-blocking sockets. support both blocking and non-blocking sockets.
@ -44,57 +45,58 @@ def read_socket_input(connection, socket_obj):
try: try:
sock_data = socket_obj.recv(count) sock_data = socket_obj.recv(count)
except socket.timeout, e: except socket.timeout as e:
LOG.debug("Socket timeout exception %s", str(e)) LOG.debug("Socket timeout exception %s", str(e))
raise # caller must handle raise # caller must handle
except socket.error, e: except socket.error as e:
LOG.debug("Socket error exception %s", str(e)) LOG.debug("Socket error exception %s", str(e))
err = e.args[0] err = e.args[0]
# ignore non-fatal errors # ignore non-fatal errors
if (err != errno.EAGAIN and if (err != errno.EAGAIN and
err != errno.EWOULDBLOCK and err != errno.EWOULDBLOCK and
err != errno.EINTR): err != errno.EINTR):
# otherwise, unrecoverable: # otherwise, unrecoverable:
connection.close_input() connection.close_input()
raise # caller must handle raise # caller must handle
except Exception, e: # beats me... assume fatal except Exception as e: # beats me... assume fatal
LOG.debug("unknown socket exception %s", str(e)) LOG.debug("unknown socket exception %s", str(e))
connection.close_input() connection.close_input()
raise # caller must handle raise # caller must handle
if sock_data: if sock_data:
count = connection.process_input( sock_data ) count = connection.process_input(sock_data)
else: else:
LOG.debug("Socket closed") LOG.debug("Socket closed")
count = Connection.EOS count = Connection.EOS
connection.close_input() connection.close_input()
return count return count
def write_socket_output(connection, socket_obj): def write_socket_output(connection, socket_obj):
"""Write data to the network layer. Can support both blocking and """Write data to the network layer. Can support both blocking and
non-blocking sockets. non-blocking sockets.
""" """
count = connection.has_output count = connection.has_output
if count <= 0: if count <= 0:
return count # 0 or EOS return count # 0 or EOS
data = connection.output_data() data = connection.output_data()
try: try:
count = socket_obj.send(data) count = socket_obj.send(data)
except socket.timeout, e: except socket.timeout as e:
LOG.debug("Socket timeout exception %s", str(e)) LOG.debug("Socket timeout exception %s", str(e))
raise # caller must handle raise # caller must handle
except socket.error, e: except socket.error as e:
LOG.debug("Socket error exception %s", str(e)) LOG.debug("Socket error exception %s", str(e))
err = e.args[0] err = e.args[0]
# ignore non-fatal errors # ignore non-fatal errors
if (err != errno.EAGAIN and if (err != errno.EAGAIN and
err != errno.EWOULDBLOCK and err != errno.EWOULDBLOCK and
err != errno.EINTR): err != errno.EINTR):
# otherwise, unrecoverable # otherwise, unrecoverable
connection.close_output() connection.close_output()
raise raise
except Exception, e: # beats me... assume fatal except Exception as e: # beats me... assume fatal
LOG.debug("unknown socket exception %s", str(e)) LOG.debug("unknown socket exception %s", str(e))
connection.close_output() connection.close_output()
raise raise
@ -106,4 +108,3 @@ def write_socket_output(connection, socket_obj):
count = Connection.EOS count = Connection.EOS
connection.close_output() connection.close_output()
return count return count