Jak obsługiwać asyncore w klasie w python, bez blokowania czegokolwiek?

Question

Potrzebuję utworzyć klasę, która może odbierać i przechowywać wiadomości SMTP, tj. Wiadomości e-mail. Aby to zrobić, używam asyncore zgodnie z przykładem opublikowanym here. Jednak asyncore.loop() jest blokowany, więc nie mogę zrobić nic więcej w kodzie.

Więc pomyślałem o użyciu wątków. Oto przykładowy kod, który pokazuje, co mam na myśli:

class MyServer(smtpd.SMTPServer): 
    # derive from the python server class 

    def process_message(..): 
     # overwrite a smtpd.SMTPServer method to be able to handle the received messages 
     ... 
     self.list_emails.append(this_email) 

    def get_number_received_emails(self): 
     """Return the current number of stored emails""" 
     return len(self.list_emails) 


    def start_receiving(self): 
     """Start the actual server to listen on port 25""" 

     self.thread = threading.Thread(target=asyncore.loop) 
     self.thread.start()  

    def stop(self): 
     """Stop listening now to port 25""" 
     # close the SMTPserver from itself 
     self.close() 
     self.thread.join() 

Mam nadzieję, że dostaniesz zdjęcie. Klasa MyServer powinna mieć możliwość uruchamiania i zatrzymywania odsłuchiwania portu 25 w sposób nieblokujący, który może być sprawdzany w poszukiwaniu wiadomości podczas słuchania (lub nie). Metoda start uruchamia detektor asyncore.loop(), który po odebraniu wiadomości e-mail dołącza się do listy wewnętrznej. Podobnie, metoda stop powinna być w stanie zatrzymać ten serwer, zgodnie z sugestią here.

Pomimo faktu, kod ten nie działa jak oczekuję (asyncore wydaje się działać zawsze, nawet zadzwonię powyższy stop metody. Na error podniosę jest przechwycony w ciągu stop, ale nie wewnątrz funkcji target zawierającej asyncore.loop()) Nie jestem pewien, czy moje podejście do problemu jest sensowne. Sugerowane są wszelkie sugestie dotyczące naprawienia powyższego kodu lub zaproponowania bardziej solidnej implementacji (bez przy użyciu oprogramowania innej firmy).

Answer 1

Dostarczone rozwiązanie może nie być najbardziej wyrafinowanym rozwiązaniem, ale działa poprawnie i zostało przetestowane.

Przede wszystkim chodzi o asyncore.loop() jest to, że blokuje aż wszystkie asyncore kanały są zamknięte, a użytkownik Wessie wskazał w komentarzu wcześniej. Odwołując się do wspomnianego wcześniej smtp example, okazuje się, że smtpd.SMTPServer dziedziczy po asyncore.dispatcher (jak opisano w smtpd documentation), który odpowiada na pytanie, który kanał należy zamknąć.

Dlatego oryginalne pytanie można odpowiedzieć w następujący zaktualizowanego kodu przykładu:

class CustomSMTPServer(smtpd.SMTPServer): 
    # store the emails in any form inside the custom SMTP server 
    emails = [] 
    # overwrite the method that is used to process the received 
    # emails, putting them into self.emails for example 
    def process_message(self, peer, mailfrom, rcpttos, data): 
     # email processing 


class MyReceiver(object): 
    def start(self): 
     """Start the listening service""" 
     # here I create an instance of the SMTP server, derived from asyncore.dispatcher 
     self.smtp = CustomSMTPServer(('0.0.0.0', 25), None) 
     # and here I also start the asyncore loop, listening for SMTP connection, within a thread 
     # timeout parameter is important, otherwise code will block 30 seconds after the smtp channel has been closed 
     self.thread = threading.Thread(target=asyncore.loop,kwargs = {'timeout':1}) 
     self.thread.start()  

    def stop(self): 
     """Stop listening now to port 25""" 
     # close the SMTPserver to ensure no channels connect to asyncore 
     self.smtp.close() 
     # now it is save to wait for the thread to finish, i.e. for asyncore.loop() to exit 
     self.thread.join() 

    # now it finally it is possible to use an instance of this class to check for emails or whatever in a non-blocking way 
    def count(self): 
     """Return the number of emails received""" 
     return len(self.smtp.emails)   
    def get(self): 
     """Return all emails received so far""" 
     return self.smtp.emails 
    .... 

Więc w końcu, mam start oraz sposobu stop zacząć i przestać słuchać na porcie 25 w nieprzekraczalnym środowisko blokujące.

Answer 2

Zamiast tego należy rozważyć użycie Twisted. http://twistedmatrix.com/trac/browser/trunk/doc/mail/examples/emailserver.tac demonstruje sposób konfigurowania serwera SMTP z dostosowywanym hakiem dostarczania.

Answer 3

pochodzących z innych kwestii asyncore.loop doesn't terminate when there are no more connections

myślę, że jesteś nieco ponad myślenia wątków. Używając kodu z drugim pytaniem, można rozpocząć nowy wątek, który uruchamia asyncore.loop przez poniższym fragmencie kodu:

import threading 

loop_thread = threading.Thread(target=asyncore.loop, name="Asyncore Loop") 
# If you want to make the thread a daemon 
# loop_thread.daemon = True 
loop_thread.start() 

ta zostanie uruchomiona w nowym wątku i będzie będzie aż wszystkie asyncore kanały są zamknięte.