Python源码示例：PyQt5.QtNetwork.QLocalSocket()
示例1
def __init__(self, id, *argv):
super().__init__(*argv)
self._id = id
# Is there another instance running?
self._outSocket = QLocalSocket()
self._outSocket.connectToServer(self._id)
self._isRunning = self._outSocket.waitForConnected()
if self._isRunning:
# Yes, there is.
self._outStream = QTextStream(self._outSocket)
self._outStream.setCodec('UTF-8')
else:
# No, there isn't.
self._outSocket = None
self._outStream = None
self._inSocket = None
self._inStream = None
self._server = QLocalServer()
self._server.removeServer(self._id)
self._server.listen(self._id)
self._server.newConnection.connect(self._onNewConnection)
示例2
def __init__(self, options, argv):
"""
Initialize Eddy.
:type options: Namespace
:type argv: list
"""
super().__init__(argv)
self.server = None
self.socket = QtNetwork.QLocalSocket()
self.socket.connectToServer(APPID)
self.running = self.socket.waitForConnected()
self.sessions = DistinctList()
self.welcome = None
if not self.isRunning() or options.tests:
self.server = QtNetwork.QLocalServer()
self.server.listen(APPID)
self.socket = None
connect(self.sgnCreateSession, self.doCreateSession)
#############################################
# INTERFACE
#################################
示例3
def __init__(self, *args, **kwargs):
super(QSingleApplication, self).__init__(*args, **kwargs)
appid = QApplication.applicationFilePath().lower().split("/")[-1]
self._socketName = "qtsingleapp-" + appid
# print("socketName", self._socketName)
self._activationWindow = None
self._activateOnMessage = False
self._socketServer = None
self._socketIn = None
self._socketOut = None
self._running = False
# 先尝试连接
self._socketOut = QLocalSocket(self)
self._socketOut.connectToServer(self._socketName)
self._socketOut.error.connect(self.handleError)
self._running = self._socketOut.waitForConnected()
if not self._running: # 程序未运行
self._socketOut.close()
del self._socketOut
self._socketServer = QLocalServer(self)
self._socketServer.listen(self._socketName)
self._socketServer.newConnection.connect(self._onNewConnection)
self.aboutToQuit.connect(self.removeServer)
示例4
def __init__(self, *args, **kwargs):
super(QSingleApplication, self).__init__(*args, **kwargs)
appid = QApplication.applicationFilePath().lower().split("/")[-1]
self._socketName = "qtsingleapp-" + appid
print("socketName", self._socketName)
self._activationWindow = None
self._activateOnMessage = False
self._socketServer = None
self._socketIn = None
self._socketOut = None
self._running = False
# 先尝试连接
self._socketOut = QLocalSocket(self)
self._socketOut.connectToServer(self._socketName)
self._socketOut.error.connect(self.handleError)
self._running = self._socketOut.waitForConnected()
if not self._running: # 程序未运行
self._socketOut.close()
del self._socketOut
self._socketServer = QLocalServer(self)
self._socketServer.listen(self._socketName)
self._socketServer.newConnection.connect(self._onNewConnection)
self.aboutToQuit.connect(self.removeServer)
示例5
def __init__(self, id, *argv):
super(QSingleApplication, self).__init__(*argv)
self._id = id
self._activationWindow = None
self._activateOnMessage = False
self._server = None
# Is there another instance running?
self._outSocket = QLocalSocket()
self._outSocket.connectToServer(self._id)
self._outSocket.error.connect(self.handleError)
self._isRunning = self._outSocket.waitForConnected()
if self._isRunning:
# Yes, there is.
self._outStream = QTextStream(self._outSocket)
self._outStream.setCodec('UTF-8')
else:
# No, there isn't.
self._outSocket = None
self._outStream = None
self._inSocket = None
self._inStream = None
self._server = QLocalServer()
self._server.listen(self._id)
self._server.newConnection.connect(self._onNewConnection)
self.aboutToQuit.connect(self.removeServer)
示例6
def main():
global win
signal.signal(signal.SIGINT, exit)
args = parse_arguments()
appKey = "scudcloud.pid"
socket = QLocalSocket()
socket.connectToServer(appKey)
if socket.isOpen():
socket.close()
socket.deleteLater()
return 0
socket.deleteLater()
app = QtWidgets.QApplication(sys.argv)
app.setApplicationName(Resources.APP_NAME+' Slack')
app.setWindowIcon(QtGui.QIcon(Resources.get_path('scudcloud.png')))
try:
settings_path, cache_path = load_settings(args.confdir, args.cachedir)
except:
print("Data directories "+args.confdir+" and "+args.cachedir+" could not be created! Exiting...")
raise SystemExit()
minimized = True if args.minimized is True else None
urgent_hint = True if args.urgent_hint is True else None
# Let's move the CSS to cachedir to enable additional actions
copyfile(Resources.get_path('resources.css'), os.path.join(cache_path, 'resources.css'))
# If there is an qt4 config and not a qt5, let's copy the old one
qt4_config = os.path.join(settings_path, 'scudcloud.cfg')
qt5_config = os.path.join(settings_path, 'scudcloud_qt5.cfg')
if os.path.exists(qt4_config) and not os.path.exists(qt5_config):
copyfile(qt4_config, qt5_config)
win = sca.ScudCloud(
debug=args.debug,
minimized=minimized,
urgent_hint=urgent_hint,
settings_path=settings_path,
cache_path=cache_path
)
app.commitDataRequest.connect(win.setForceClose, type=QtCore.Qt.DirectConnection)
server = QLocalServer()
server.newConnection.connect(restore)
server.listen(appKey)
win.restore()
if win.minimized is None:
win.show()
sys.exit(app.exec_())
示例7
def __init__(self, _id,_viewer_id, *argv):
if sys.platform.startswith("darwin") and mp.current_process().name == "WebLCDs":
import AppKit
info = AppKit.NSBundle.mainBundle().infoDictionary() # @UndefinedVariable
info["LSBackgroundOnly"] = "1"
super(QtSingleApplication, self).__init__(*argv)
self._id = _id
self._viewer_id = _viewer_id
self._activationWindow = None
self._activateOnMessage = False
self._outSocket = None
self._isRunning = False
self._server = None
# we exclude the WebLCDs parallel process from participating any Artisan inter-app communication
if mp.current_process().name != "WebLCDs":
# Is there another instance running?
self._outSocket = QLocalSocket()
self._outSocket.connectToServer(self._id)
self._isRunning = self._outSocket.waitForConnected(-1)
if self._isRunning:
# Yes, there is.
self._outStream = QTextStream(self._outSocket)
self._outStream.setCodec('UTF-8')
# Is there another viewer running?
self._outSocketViewer = QLocalSocket()
self._outSocketViewer.connectToServer(self._viewer_id)
self._isRunningViewer = self._outSocketViewer.waitForConnected(-1)
if self._isRunningViewer:
self._outStreamViewer = QTextStream(self._outSocketViewer)
self._outStreamViewer.setCodec('UTF-8')
else:
# app is running, we announce us as viewer app
# First we remove existing servers of that name that might not have been properly closed as the server died
QLocalServer.removeServer(self._viewer_id)
self._outSocketViewer = None
self._outStreamViewer = None
self._inSocket = None
self._inStream = None
self._server = QLocalServer()
self._server.listen(self._viewer_id)
self._server.newConnection.connect(self._onNewConnection)
else:
self._isRunningViewer = False
# No, there isn't.
# First we remove existing servers of that name that might not have been properly closed as the server died
QLocalServer.removeServer(self._id)
self._outSocket = None
self._outStream = None
self._inSocket = None
self._inStream = None
self._server = QLocalServer()
self._server.listen(self._id)
self._server.newConnection.connect(self._onNewConnection)
