mirror of https://github.com/borgbase/vorta
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from PyQt6.QtCore import QTextStream, pyqtSignal
|
|
from PyQt6.QtNetwork import QLocalServer, QLocalSocket
|
|
from PyQt6.QtWidgets import QApplication
|
|
|
|
|
|
class QtSingleApplication(QApplication):
|
|
"""
|
|
Adapted from https://stackoverflow.com/a/12712362/11038610
|
|
|
|
Copyright 2013 Johan Råde
|
|
|
|
Redistribution and use in source and binary forms, with or without
|
|
modification, are permitted provided that the following conditions
|
|
are met:
|
|
|
|
1. Redistributions of source code must retain the above copyright
|
|
notice, this list of conditions and the following disclaimer.
|
|
|
|
2. Redistributions in binary form must reproduce the above
|
|
copyright notice, this list of conditions and the following
|
|
disclaimer in the documentation and/or other materials provided
|
|
with the distribution.
|
|
|
|
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
|
|
CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
|
|
INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
|
|
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
|
|
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS
|
|
BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
|
|
EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
|
|
TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
|
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
|
|
ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR
|
|
TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
|
|
THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
|
|
SUCH DAMAGE.
|
|
|
|
"""
|
|
|
|
message_received_event = pyqtSignal(str)
|
|
|
|
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)
|
|
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)
|
|
|
|
def isRunning(self):
|
|
return self._isRunning
|
|
|
|
def id(self):
|
|
return self._id
|
|
|
|
def sendMessage(self, msg):
|
|
if not self._outStream:
|
|
return False
|
|
self._outStream << msg << '\n'
|
|
self._outStream.flush()
|
|
return self._outSocket.waitForBytesWritten()
|
|
|
|
def _onNewConnection(self):
|
|
if self._inSocket:
|
|
self._inSocket.readyRead.disconnect(self._onReadyRead)
|
|
self._inSocket = self._server.nextPendingConnection()
|
|
if not self._inSocket:
|
|
return
|
|
self._inStream = QTextStream(self._inSocket)
|
|
self._inSocket.readyRead.connect(self._onReadyRead)
|
|
|
|
def _onReadyRead(self):
|
|
while True:
|
|
msg = self._inStream.readLine()
|
|
if not msg:
|
|
break
|
|
self.message_received_event.emit(msg)
|