Use a separate IPCConnection class

2 files changed, 156 insertions, 166 deletions

diff --git a/.mypy.ini b/.mypy.ini
index 81f69a09e..eef89ae20 100644
--- a/.mypy.ini
+++ b/.mypy.ini
@@ -240,9 +240,6 @@ disallow_untyped_defs = False
 [mypy-qutebrowser.misc.httpclient]
 disallow_untyped_defs = False
 
-[mypy-qutebrowser.misc.ipc]
-disallow_untyped_defs = False
-
 [mypy-qutebrowser.misc.keyhintwidget]
 disallow_untyped_defs = False
 
@@ -267,6 +264,7 @@ disallow_untyped_defs = False
 [mypy-qutebrowser.misc.split]
 disallow_untyped_defs = False
 
+
 [mypy-qutebrowser.qutebrowser]
 disallow_untyped_defs = False
 
diff --git a/qutebrowser/misc/ipc.py b/qutebrowser/misc/ipc.py
index 72159eab7..9342ab55b 100644
--- a/qutebrowser/misc/ipc.py
+++ b/qutebrowser/misc/ipc.py
@@ -10,7 +10,9 @@ import json
 import getpass
 import binascii
 import hashlib
-from typing import Optional
+import itertools
+import argparse
+from typing import Optional, List
 
 from qutebrowser.qt.core import pyqtSignal, pyqtSlot, QObject, Qt
 from qutebrowser.qt.network import QLocalSocket, QLocalServer, QAbstractSocket
@@ -31,7 +33,7 @@ PROTOCOL_VERSION = 1
 server: Optional["IPCServer"] = None
 
 
-def _get_socketname_windows(basedir):
+def _get_socketname_windows(basedir: Optional[str]) -> str:
     """Get a socketname to use for Windows."""
     try:
         username = getpass.getuser()
@@ -52,7 +54,7 @@ def _get_socketname_windows(basedir):
     return '-'.join(parts)
 
 
-def _get_socketname(basedir):
+def _get_socketname(basedir: Optional[str]) -> str:
     """Get a socketname to use."""
     if utils.is_windows:  # pragma: no cover
         return _get_socketname_windows(basedir)
@@ -84,7 +86,7 @@ class SocketError(Error):
         action: The action which was taken when the error happened.
     """
 
-    def __init__(self, action, socket):
+    def __init__(self, action: str, socket: QLocalSocket) -> None:
         """Constructor.
 
         Args:
@@ -96,7 +98,7 @@ class SocketError(Error):
         self.code: QLocalSocket.LocalSocketError = socket.error()
         self.message: str = socket.errorString()
 
-    def __str__(self):
+    def __str__(self) -> str:
         return "Error while {}: {} ({})".format(
             self.action, self.message, debug.qenum_key(QLocalSocket, self.code))
 
@@ -110,7 +112,7 @@ class ListenError(Error):
         message: The error message.
     """
 
-    def __init__(self, local_server):
+    def __init__(self, local_server: QLocalServer) -> None:
         """Constructor.
 
         Args:
@@ -120,7 +122,7 @@ class ListenError(Error):
         self.code: QAbstractSocket.SocketError = local_server.serverError()
         self.message: str = local_server.errorString()
 
-    def __str__(self):
+    def __str__(self) -> str:
         return "Error while listening to IPC server: {} ({})".format(
             self.message, debug.qenum_key(QAbstractSocket, self.code))
 
@@ -130,6 +132,114 @@ class AddressInUseError(ListenError):
     """Emitted when the server address is already in use."""
 
 
+class IPCConnection(QObject):
+    """A connection to an IPC socket.
+
+    Multiple connections might be active in parallel.
+
+    Attributes:
+        _socket: The QLocalSocket to use.
+
+    Signals:
+        got_raw: Emitted with the connection ID and raw data from the socket.
+    """
+
+    got_raw = pyqtSignal(int, bytes)
+    id_gen = itertools.count()
+
+    def __init__(self, socket: QLocalSocket, parent: Optional[QObject] = None):
+        super().__init__(parent)
+        self.conn_id = next(self.id_gen)
+        log.ipc.debug("Client connected (socket {}).".format(self.conn_id))
+
+        self._timer = usertypes.Timer(self, "ipc-timeout")
+        self._timer.setInterval(READ_TIMEOUT)
+        self._timer.timeout.connect(self.on_timeout)
+        self._timer.start()
+
+        self._socket: Optional[QLocalSocket] = socket
+        self._socket.readyRead.connect(self.on_ready_read)
+
+        if socket.canReadLine():
+            log.ipc.debug("We can read a line immediately.")
+            self.on_ready_read()
+
+        socket.errorOccurred.connect(self.on_error)
+
+        # FIXME:v4 Ignore needed due to overloaded signal/method in Qt 5
+        socket_error = socket.error()  # type: ignore[operator,unused-ignore]
+        if socket_error not in [
+            QLocalSocket.LocalSocketError.UnknownSocketError,
+            QLocalSocket.LocalSocketError.PeerClosedError,
+        ]:
+            log.ipc.debug("We got an error immediately.")
+            self.on_error(socket_error)
+
+        socket.disconnected.connect(self.on_disconnected)
+        if socket.state() == QLocalSocket.LocalSocketState.UnconnectedState:
+            log.ipc.debug("Socket was disconnected immediately.")
+            self.on_disconnected()
+
+    @pyqtSlot("QLocalSocket::LocalSocketError")
+    def on_error(self, err: QLocalSocket.LocalSocketError) -> None:
+        """Raise SocketError on fatal errors."""
+        if self._socket is None:
+            # Sometimes this gets called from stale sockets.
+            log.ipc.debug("In on_error with None socket!")
+            return
+        self._timer.stop()
+        log.ipc.debug(
+            "Socket {}: error {}: {}".format(
+                self.conn_id, self._socket.error(), self._socket.errorString()
+            )
+        )
+        if err != QLocalSocket.LocalSocketError.PeerClosedError:
+            raise SocketError(f"handling IPC connection {self.conn_id}", self._socket)
+
+    @pyqtSlot()
+    def on_disconnected(self) -> None:
+        """Clean up socket when the client disconnected."""
+        assert self._socket is not None
+        log.ipc.debug(f"Client disconnected from socket {self.conn_id}.")
+        self._timer.stop()
+        self._socket.deleteLater()
+        self._socket = None
+        self.deleteLater()
+
+    @pyqtSlot()
+    def on_ready_read(self) -> None:
+        """Read json data from the client."""
+        self._timer.stop()
+
+        while self._socket is not None and self._socket.canReadLine():
+            data = self._socket.readLine().data()
+            log.ipc.debug("Read from socket {}: {!r}".format(self.conn_id, data))
+            self.got_raw.emit(self.conn_id, data)
+
+        if self._socket is not None:
+            self._timer.start()
+
+    @pyqtSlot()
+    def on_timeout(self) -> None:
+        """Cancel the current connection if it was idle for too long."""
+        assert self._socket is not None
+        log.ipc.error(f"IPC connection timed out (socket {self.conn_id}).")
+        self._socket.disconnectFromServer()
+        if self._socket is not None:  # pragma: no cover
+            # on_disconnected sets it to None
+            self._socket.waitForDisconnected(CONNECT_TIMEOUT)
+        if self._socket is not None:  # pragma: no cover
+            # on_disconnected sets it to None
+            self._socket.abort()
+
+    @pyqtSlot(int)
+    def on_invalid_data(self, conn_id: int) -> None:
+        if conn_id != self.conn_id:
+            return
+        assert self._socket is not None
+        self._socket.disconnectFromServer()
+
+
 class IPCServer(QObject):
 
     """IPC server to which clients connect to.
@@ -138,23 +248,23 @@ class IPCServer(QObject):
         ignored: Whether requests are ignored (in exception hook).
         _timer: A timer to handle timeouts.
         _server: A QLocalServer to accept new connections.
-        _socket: The QLocalSocket we're currently connected to.
-        _socket_id: An unique, incrementing ID for every socket.
         _socketname: The socketname to use.
         _atime_timer: Timer to update the atime of the socket regularly.
 
     Signals:
         got_args: Emitted when there was an IPC connection and arguments were
                   passed.
-        got_args: Emitted with the raw data an IPC connection got.
+        got_raw: Emitted with the raw data an IPC connection got.
         got_invalid_data: Emitted when there was invalid incoming data.
+        shutting_down: IPC is shutting down.
     """
 
     got_args = pyqtSignal(list, str, str)
     got_raw = pyqtSignal(bytes)
-    got_invalid_data = pyqtSignal()
+    got_invalid_data = pyqtSignal(int)
+    shutting_down = pyqtSignal()
 
-    def __init__(self, socketname, parent=None):
+    def __init__(self, socketname: str, parent: QObject = None) -> None:
         """Start the IPC server and listen to commands.
 
         Args:
@@ -165,10 +275,6 @@ class IPCServer(QObject):
         self.ignored = False
         self._socketname = socketname
 
-        self._timer = usertypes.Timer(self, 'ipc-timeout')
-        self._timer.setInterval(READ_TIMEOUT)
-        self._timer.timeout.connect(self.on_timeout)
-
         if utils.is_windows:  # pragma: no cover
             self._atime_timer = None
         else:
@@ -180,10 +286,6 @@ class IPCServer(QObject):
         self._server: Optional[QLocalServer] = QLocalServer(self)
         self._server.newConnection.connect(self.handle_connection)
 
-        self._socket = None
-        self._socket_id = 0
-        self._old_socket = None
-
         if utils.is_windows:  # pragma: no cover
             # As a WORKAROUND for a Qt bug, we can't use UserAccessOption on Unix. If we
             # do, we don't get an AddressInUseError anymore:
@@ -196,14 +298,14 @@ class IPCServer(QObject):
         else:  # pragma: no cover
             log.ipc.debug("Not calling setSocketOptions")
 
-    def _remove_server(self):
+    def _remove_server(self) -> None:
         """Remove an existing server."""
         ok = QLocalServer.removeServer(self._socketname)
         if not ok:
             raise Error("Error while removing server {}!".format(
                 self._socketname))
 
-    def listen(self):
+    def listen(self) -> None:
         """Start listening on self._socketname."""
         assert self._server is not None
         log.ipc.debug("Listening as {}".format(self._socketname))
@@ -226,90 +328,30 @@ class IPCServer(QObject):
                 # True, so report this as an error.
                 raise ListenError(self._server)
 
-    @pyqtSlot('QLocalSocket::LocalSocketError')
-    def on_error(self, err):
-        """Raise SocketError on fatal errors."""
-        if self._socket is None:
-            # Sometimes this gets called from stale sockets.
-            log.ipc.debug("In on_error with None socket!")
-            return
-        self._timer.stop()
-        log.ipc.debug("Socket {}: error {}: {}".format(
-            self._socket_id, self._socket.error(),
-            self._socket.errorString()))
-        if err != QLocalSocket.LocalSocketError.PeerClosedError:
-            raise SocketError(
-                f"handling IPC connection {self._socket_id}", self._socket
-            )
-
     @pyqtSlot()
-    def handle_connection(self):
+    def handle_connection(self) -> None:
         """Handle a new connection to the server."""
         if self.ignored or self._server is None:
             return
-        if self._socket is not None:
-            log.ipc.debug("Got new connection but ignoring it because we're "
-                          "still handling another one ({}).".format(
-                              self._socket_id))
-            return
+
         socket = qtutils.add_optional(self._server.nextPendingConnection())
         if socket is None:
             log.ipc.debug("No new connection to handle.")
             return
 
-        self._socket_id += 1
-        log.ipc.debug("Client connected (socket {}).".format(self._socket_id))
-        self._socket = socket
-        self._timer.start()
-        socket.readyRead.connect(self.on_ready_read)
-        if socket.canReadLine():
-            log.ipc.debug("We can read a line immediately.")
-            self.on_ready_read()
-
-        socket.errorOccurred.connect(self.on_error)
-
-        # FIXME:v4 Ignore needed due to overloaded signal/method in Qt 5
-        socket_error = socket.error()  # type: ignore[operator,unused-ignore]
-        if socket_error not in [
-            QLocalSocket.LocalSocketError.UnknownSocketError,
-            QLocalSocket.LocalSocketError.PeerClosedError
-        ]:
-            log.ipc.debug("We got an error immediately.")
-            self.on_error(socket_error)
-
-        socket.disconnected.connect(self.on_disconnected)
-        if socket.state() == QLocalSocket.LocalSocketState.UnconnectedState:
-            log.ipc.debug("Socket was disconnected immediately.")
-            self.on_disconnected()
+        conn = IPCConnection(socket, parent=self)
+        conn.got_raw.connect(self.handle_data)
+        self.got_invalid_data.connect(conn.on_invalid_data)
+        self.shutting_down.connect(conn.on_disconnected)
 
-    @pyqtSlot()
-    def on_disconnected(self):
-        """Clean up socket when the client disconnected."""
-        log.ipc.debug("Client disconnected from socket {}.".format(self._socket_id))
-        self._timer.stop()
-        if self._old_socket is not None:
-            self._old_socket.deleteLater()
-        self._old_socket = self._socket
-        self._socket = None
-        # Maybe another connection is waiting.
-        self.handle_connection()
-
-    def _handle_invalid_data(self):
-        """Handle invalid data we got from a QLocalSocket."""
-        assert self._socket is not None
-        log.ipc.error("Ignoring invalid IPC data from socket {}.".format(
-            self._socket_id))
-        self.got_invalid_data.emit()
-        self._socket.errorOccurred.connect(self.on_error)
-        self._socket.disconnectFromServer()
-
-    def _handle_data(self, data):
-        """Handle data (as bytes) we got from on_ready_read."""
+    @pyqtSlot(int, bytes)
+    def handle_data(self, conn_id: int, data: bytes) -> None:
+        """Handle data we got from a connection."""
         try:
             decoded = data.decode('utf-8')
         except UnicodeDecodeError:
             log.ipc.error("invalid utf-8: {!r}".format(binascii.hexlify(data)))
-            self._handle_invalid_data()
+            self._handle_invalid_data(conn_id)
             return
 
         log.ipc.debug("Processing: {}".format(decoded))
@@ -317,26 +359,26 @@ class IPCServer(QObject):
             json_data = json.loads(decoded)
         except ValueError:
             log.ipc.error("invalid json: {}".format(decoded.strip()))
-            self._handle_invalid_data()
+            self._handle_invalid_data(conn_id)
             return
 
         for name in ['args', 'target_arg']:
             if name not in json_data:
                 log.ipc.error("Missing {}: {}".format(name, decoded.strip()))
-                self._handle_invalid_data()
+                self._handle_invalid_data(conn_id)
                 return
 
         try:
             protocol_version = int(json_data['protocol_version'])
         except (KeyError, ValueError):
             log.ipc.error("invalid version: {}".format(decoded.strip()))
-            self._handle_invalid_data()
+            self._handle_invalid_data(conn_id)
             return
 
         if protocol_version != PROTOCOL_VERSION:
             log.ipc.error("incompatible version: expected {}, got {}".format(
                 PROTOCOL_VERSION, protocol_version))
-            self._handle_invalid_data()
+            self._handle_invalid_data(conn_id)
             return
 
         args = json_data['args']
@@ -351,65 +393,13 @@ class IPCServer(QObject):
 
         self.got_args.emit(args, target_arg, cwd)
 
-    def _get_socket(self, warn=True):
-        """Get the current socket and ID for on_ready_read.
-
-        Arguments:
-            warn: Whether to warn if no socket was found.
-        """
-        if self._socket is None:  # pragma: no cover
-            # This happens when doing a connection while another one is already
-            # active for some reason.
-            if self._old_socket is None:
-                if warn:
-                    log.ipc.warning("In _get_socket with None socket and old_socket!")
-                return None, None
-            log.ipc.debug("In _get_socket with None socket!")
-            socket = self._old_socket
-            # hopefully accurate guess, but at least we have a debug log
-            socket_id = self._socket_id - 1
-        else:
-            socket = self._socket
-            socket_id = self._socket_id
-
-        if sip.isdeleted(socket):  # pragma: no cover
-            log.ipc.warning("Ignoring deleted IPC socket")
-            return None, None
-
-        return socket, socket_id
-
-    @pyqtSlot()
-    def on_ready_read(self):
-        """Read json data from the client."""
-        self._timer.stop()
-
-        socket, socket_id = self._get_socket()
-        while socket is not None and socket.canReadLine():
-            data = bytes(socket.readLine())
-            self.got_raw.emit(data)
-            log.ipc.debug("Read from socket {}: {!r}".format(socket_id, data))
-            self._handle_data(data)
-            socket, socket_id = self._get_socket(warn=False)
-
-        if self._socket is not None:
-            self._timer.start()
-
-    @pyqtSlot()
-    def on_timeout(self):
-        """Cancel the current connection if it was idle for too long."""
-        assert self._socket is not None
-        log.ipc.error("IPC connection timed out "
-                      "(socket {}).".format(self._socket_id))
-        self._socket.disconnectFromServer()
-        if self._socket is not None:  # pragma: no cover
-            # on_socket_disconnected sets it to None
-            self._socket.waitForDisconnected(CONNECT_TIMEOUT)
-        if self._socket is not None:  # pragma: no cover
-            # on_socket_disconnected sets it to None
-            self._socket.abort()
+    def _handle_invalid_data(self, conn_id: int) -> None:
+        """Handle invalid data we got from a QLocalSocket."""
+        log.ipc.error(f"Ignoring invalid IPC data from socket {conn_id}.")
+        self.got_invalid_data.emit(conn_id)
 
     @pyqtSlot()
-    def update_atime(self):
+    def update_atime(self) -> None:
         """Update the atime of the socket file all few hours.
 
         From the XDG basedir spec:
@@ -435,7 +425,7 @@ class IPCServer(QObject):
             self.listen()
 
     @pyqtSlot()
-    def shutdown(self):
+    def shutdown(self) -> None:
         """Shut down the IPC server cleanly."""
         if self._server is None:
             # We can get called twice when using :restart -- there, IPC is shut down
@@ -443,13 +433,9 @@ class IPCServer(QObject):
             # we get called again when the application is about to quit.
             return
 
-        log.ipc.debug("Shutting down IPC (socket {})".format(self._socket_id))
-
-        if self._socket is not None:
-            self._socket.deleteLater()
-            self._socket = None
+        log.ipc.debug("Shutting down IPC")
+        self.shutting_down.emit()
 
-        self._timer.stop()
         if self._atime_timer is not None:  # pragma: no branch
             self._atime_timer.stop()
             try:
@@ -463,7 +449,13 @@ class IPCServer(QObject):
         self._server = None
 
 
-def send_to_running_instance(socketname, command, target_arg, *, socket=None):
+def send_to_running_instance(
+    socketname: int,
+    command: List[str],
+    target_arg: str,
+    *,
+    socket: Optional[QLocalSocket] = None,
+) -> None:
     """Try to send a commandline to a running instance.
 
     Blocks for CONNECT_TIMEOUT ms.
@@ -515,14 +507,14 @@ def send_to_running_instance(socketname, command, target_arg, *, socket=None):
         return False
 
 
-def display_error(exc, args):
+def display_error(exc: Exception, args: argparse.Namespace) -> None:
     """Display a message box with an IPC error."""
     error.handle_fatal_exc(
         exc, "Error while connecting to running instance!",
         no_err_windows=args.no_err_windows)
 
 
-def send_or_listen(args):
+def send_or_listen(args: argparse.Namespace) -> None:
     """Send the args to a running instance or start a new IPCServer.
 
     Args: