Newer
Older
#!/usr/bin/env python
#coding=utf8
import ssl
import time
import socket
import select
import logging
from sjrpc.core import RpcConnection, ConnectionManager
class SimpleRpcServer(ConnectionManager):
'''
A simple RPC Server that wait for new connections and dispatch messages
from thoses are already established.
:param sock: the :class:`socket` object to bind to the server connection
:param default_handler: the default handler to bind to the new client
connections
'''
def __init__(self, sock, default_handler=None, on_disconnect=None):
super(SimpleRpcServer, self).__init__()
sock.setblocking(False)
self._listening_sock = sock
self.register(sock, self._handle_master_event)
self._clients = {}
self._default_handler = default_handler
self._on_disconnect = on_disconnect
def _accept_connection(self):
return self._listening_sock.accept()
def shutdown(self):
super(SimpleRpcServer, self).shutdown()
time.sleep(ConnectionManager.POLL_TIMEOUT)
for connection in self._clients.values():
connection.shutdown(self._on_disconnect)
self._listening_sock.close()
def shutdown_client(self, fd):
conn = self._clients.get(fd)
try:
del self._clients[fd]
except KeyError:
pass
if conn is not None:
conn.shutdown(callback=self._on_disconnect)
Antoine Millet
committed
def all_connections(self):
return set(self._clients.values())
def _handle_master_event(self, fd, events):
# Event concerns the listening socket:
if events & select.EPOLLIN:
accepted = self._accept_connection()
if accepted is not None:
sock, address = accepted
sock.setblocking(False)
connection = RpcConnection(sock, self,
handler=self._default_handler)
self.register(connection.get_fd(), self._handle_client_event)
self._clients[connection.get_fd()] = connection
def _handle_client_event(self, fd, events):
connection = self._clients[fd]
if events & select.EPOLLIN:
# Data are ready to be readed on socket
try:
connection.receive()
except socket.error as err:
logging.debug('Socket error while receiving from client '
'fd/%s: %s', fd, err)
self.shutdown_client(fd)
except Exception as err:
logging.debug('Unknown error while receiving from client '
'fd/%s: %s', fd, err)
self.shutdown_client(fd)
Antoine Millet
committed
if events & select.EPOLLOUT:
# Data are ready to be written on socket
try:
connection.send()
except socket.error as err:
logging.debug('Socket error while sending to the client '
'fd/%s: %s', fd, err)
except Exception as err:
logging.debug('Unknown error while sending to the client '
'fd/%s: %s', fd, err)
self.shutdown_client(fd)
if events & select.EPOLLHUP:
logging.debug('Socket HUP fd/%s', fd)
self.shutdown_client(fd)
class SimpleSslRpcServer(SimpleRpcServer):
'''
A simple RPC Server that wait for new connections and dispatch messages
from thoses are already established. This server version enable SSL on
client sockets.
:param sock: the :class:`socket` object to bind to the server connection
:param default_handler: the default handler to bind to the new client
connections
'''
def __init__(self, sock, certfile=None, keyfile=None, **kwargs):
self._certfile = certfile
self._keyfile = keyfile
super(SimpleSslRpcServer, self).__init__(sock, **kwargs)
def _accept_connection(self):
sock, address = self._listening_sock.accept()
try:
sslsock = ssl.wrap_socket(sock, server_side=True,
keyfile=self._keyfile,
certfile=self._certfile,
ssl_version=ssl.PROTOCOL_TLSv1,
do_handshake_on_connect=True)
except ssl.SSLError as err:
logging.debug('Error when accepting ssl connection: %s', err)
else:
return sslsock, address