NetworkReplyTimeout.py

# Copyright (c) 2022 Aldo Hoeben / fieldOfView
# NetworkReplyTimeout is released under the terms of the AGPLv3 or higher.

try:
    from cura.ApplicationMetadata import CuraSDKVersion
except ImportError: # Cura <= 3.6
    CuraSDKVersion = "6.0.0"
if CuraSDKVersion >= "8.0.0":
    from PyQt6.QtCore import QObject, QTimer
    from PyQt6.QtNetwork import QNetworkReply
else:
    from PyQt5.QtCore import QObject, QTimer
    from PyQt5.QtNetwork import QNetworkReply

from UM.Signal import Signal

from typing import Optional, Callable

#
# A timer that is started when a QNetworkRequest returns a QNetworkReply, which closes the
# QNetworkReply does not reply in a timely manner
#
class NetworkReplyTimeout(QObject):
    timeout = Signal()

    def __init__(
        self,
        reply: QNetworkReply,
        timeout: int,
        callback: Optional[Callable[[QNetworkReply], None]] = None,
    ) -> None:
        super().__init__()

        self._reply = reply
        self._callback = callback

        self._timer = QTimer()
        self._timer.setInterval(timeout)
        self._timer.setSingleShot(True)
        self._timer.timeout.connect(self._onTimeout)

        self._timer.start()

    def _onTimeout(self):
        if self._reply.isRunning():
            self._reply.abort()
            if self._callback:
                self._callback(self._reply)
            self.timeout.emit(self._reply)