##// END OF EJS Templates
Use default OS shell to run system commands...
Use default OS shell to run system commands Instead of using os.system which uses /bin/sh, this uses subprocess.call (the replacement of os.system) to run the command using the default shell of the OS. With this, one can use more advanced commands for bash, zsh, ksh, ... I also edited the docstring, added comments and fixed the handling of return codes.

File last commit:

r11130:de5468b5
r12366:47045d65
Show More
heartbeat.py
56 lines | 2.0 KiB | text/x-python | PythonLexer
"""The client and server for a basic ping-pong style heartbeat.
"""
#-----------------------------------------------------------------------------
# Copyright (C) 2008-2011 The IPython Development Team
#
# Distributed under the terms of the BSD License. The full license is in
# the file COPYING, distributed as part of this software.
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
import os
import socket
from threading import Thread
import zmq
from IPython.utils.localinterfaces import LOCALHOST
#-----------------------------------------------------------------------------
# Code
#-----------------------------------------------------------------------------
class Heartbeat(Thread):
"A simple ping-pong style heartbeat that runs in a thread."
def __init__(self, context, addr=('tcp', LOCALHOST, 0)):
Thread.__init__(self)
self.context = context
self.transport, self.ip, self.port = addr
if self.port == 0:
if addr[0] == 'tcp':
s = socket.socket()
# '*' means all interfaces to 0MQ, which is '' to socket.socket
s.bind(('' if self.ip == '*' else self.ip, 0))
self.port = s.getsockname()[1]
s.close()
elif addr[0] == 'ipc':
self.port = 1
while os.path.exists("%s-%s" % (self.ip, self.port)):
self.port = self.port + 1
else:
raise ValueError("Unrecognized zmq transport: %s" % addr[0])
self.addr = (self.ip, self.port)
self.daemon = True
def run(self):
self.socket = self.context.socket(zmq.REP)
c = ':' if self.transport == 'tcp' else '-'
self.socket.bind('%s://%s' % (self.transport, self.ip) + c + str(self.port))
zmq.device(zmq.FORWARDER, self.socket, self.socket)