Блог

надеюсь, вы сможете выяснить, почему мой реактор не останавливается? В моем конструкторе потоков я попробовал reactor.startRunning(False) и reactor.run(). В команде run для этого потока я периодически использую reactor.iterate(), а затем попробовал reactor.stop() и reactor.callFromThread(reactor.stop), но реактор не останавливается ни разу. Любая обратная связь по этому поводу была бы потрясающей 🙂

В качестве альтернативы, чем еще я могу заменить это, что совместимо с использованием twisted.internet server amp; resource для настройки / прослушивания порта? Или, может быть, что-то, что может заменить как reactor, так и twisted.internet?

     class DatePage(resource.Resource):
        def __init__(self, date_string):
            resource.Resource.__init__(self)
            self.date_str = str(date_string)

        def render_GET(self, request):
            request.responseHeaders.addRawHeader(b"content-type", b"application/json")
            return get_json().encode('utf-8')

    class PageFactory(resource.Resource):
      def getChild(self, date_string, request):
          return DatePage(date_string)

    class ControlThread(threading.Thread):
        def __init__(self, control_queue, stop_event):
            threading.Thread.__init__(self)
            self.control_queue = control_queue
            self.stop_event = stop_event
            self.site = server.Site(PageFactory())
            self.port = reactor.listenTCP(host_port, self.site)
            reactor.startRunning(False)

        def run(self):
            while not self.stop_event.is_set():
                try:
                    self.process_command(self.control_queue.get_nowait())
                    reactor.iterate()
                except queue.Empty:
                    pass
                time.sleep(0.001)
                reactor.iterate()
            sys.stderr.write("reactor.stop()n")
            self.port.stopListening()
            reactor.callFromThread(reactor.stop)
            # reactor_thread.join()
            while reactor.running:
                print("running")
                time.sleep(0.1)
            if not reactor.running:
                print("STOPPED!!")
  

Редактировать: Добавлен пример. Проблема в ControlThread, другие классы предназначены для фона.

Похоже, вы пытаетесь запустить Twisted reactor в неосновном потоке, а затем отправлять ему сообщения (сообщение о остановке и, похоже, некоторые другие сообщения).

У Twisted уже есть потокобезопасный API для передачи сообщений — и вы даже используете его: reactor.callFromThread . Вы должны просто использовать это для связи с реактором. Это устраняет необходимость вызывать неинтерфейсные методы startRunning и iterate и использовать запутанный и сложный running атрибут (всего этого вам следует избегать в любой написанной вами программе на основе Twisted).

Итак:

 class ReactorThread(Thread):
    def __init__(self, reactor):
        Thread.__init__(self)
        self.reactor = reactor

    def listen(self):
        self.reactor.callFromThread(self._listen)

    def _listen(self):
        self.site = server.Site(PageFactory())
        self.port = reactor.listenTCP(host_port, self.site)

    def run(self):
        self.reactor.run(installSignalHandlers=False)

# later ...
from twisted.internet import reactor
t = ReactorThread(reactor)
t.start()
# ...
t.reactor.callFromThread(t.reactor.stop)
t.join()
  

Вы также можете использовать t.reactor.callFromThread для отправки любых других сообщений, которые вам нравятся, в виде функции, которая будет вызываться в потоке reactor.

Вы также можете посмотреть наhttps://crochet.readthedocs.io/en/stable / для управления Twisted в потоке.

В моем приложении я могу закрыть приложение с помощью этого кода

 from twisted.internet import reactor

# execute this in my fucntion
reactor.stop()
  

Это будет эквивалентно завершению приложения с кодом 0.
Я все еще ищу альтернативу, потому что мне нужно остановить мое приложение с ошибкой, чтобы docker compose автоматически перезапустил мой контейнер.