"@PACKAGE_NAME@",
"msgq_socket").replace("${prefix}",
"@prefix@")
-
+
def __init__(self, socket_file=None, verbose=False):
"""Initialize the MsgQ master.
-
+
The socket_file specifies the path to the UNIX domain socket
that the msgq process listens on. If it is None, the
environment variable BIND10_MSGQ_SOCKET_FILE is used. If that
self.poller = select.poll()
except AttributeError:
self.kqueue = select.kqueue()
-
+
def add_kqueue_socket(self, socket, write_filter=False):
"""Add a kquque filter for a socket. By default the read
filter is used; if write_filter is set to True, the write
self.socket_file)
self.listen_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-
+
if os.path.exists(self.socket_file):
os.remove(self.socket_file)
try:
if self.verbose:
sys.stdout.write("[b10-msgq] Listening\n")
-
+
self.runnable = True
def process_accept(self):
if fileno in self.sendbuffs:
amount_sent = 0
else:
- amount_sent = self.__send_data(sock, msg)
+ try:
+ amount_sent = self.__send_data(sock, msg)
+ except socket.error as sockerr:
+ # in the case the other side seems gone, kill the socket
+ # and drop the send action
+ if sockerr.errno == errno.EPIPE:
+ print("[b10-msgq] SIGPIPE on send, dropping message")
+ self.kill_socket(fileno, sock)
+ return
+ else:
+ raise
# Still something to send
if amount_sent < len(msg):
def run(self):
"""Process messages. Forever. Mostly."""
-
+
if self.poller:
self.run_poller()
else:
self.run_kqueue()
-
+
def run_poller(self):
while True:
try:
if __name__ == "__main__":
def check_port(option, opt_str, value, parser):
- """Function to insure that the port we are passed is actually
+ """Function to insure that the port we are passed is actually
a valid port number. Used by OptionParser() on startup."""
intval = int(value)
if (intval < 0) or (intval > 65535):