How to handle asyncore within a class in python, w

问题:

I need to create a class that can receive and store SMTP messages, i.e. E-Mails. To do so, I am using asyncore according to an example posted here. However, asyncore.loop() is blocking so I cannot do anything else in the code.

So I thought of using threads. Here is an example-code that shows what I have in mind:

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()

I hope you get the picture. The class MyServer should be able to start and stop listening to port 25 in a non-blocking way, able to be queried for messages while listening (or not). The start method starts the asyncore.loop() listener, which, when a reception of an email occurs, append to an internal list. Similar, the stop method should be able to stop this server, as suggested here.

Despite the fact this code does not work as I expect to (asyncore seems to run forever, even I call the above stop method. The error I raise is catched within stop, but not within the target function containing asyncore.loop()), I am not sure if my approach to the problem is senseful. Any suggestions for fixing the above code or proposing a more solid implementation (without using third party software), are appreciated.

回答1:

The solution provided might not be the most sophisticated solution, but it works reasonable and has been tested.

First of all, the matter with asyncore.loop() is that it blocks until all asyncore channels are closed, as user Wessie pointed out in a comment before. Referring to the smtp example mentioned earlier, it turns out that smtpd.SMTPServer inherits from asyncore.dispatcher (as described on the smtpd documentation), which answers the question of which channel to be closed.

Therefore, the original question can be answered with the following updated example code:

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
    ....

So in the end, I have a start and a stop method to start and stop listening on port 25 within a non-blocking environment.

回答2:

Coming from the other question asyncore.loop doesn't terminate when there are no more connections

I think you are slightly over thinking the threading. Using the code from the other question, you can start a new thread that runs the asyncore.loop by the following code snippet:

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()

This will run it in a new thread and will keep going till all asyncore channels are closed.

回答3:

You should consider using Twisted, instead. http://twistedmatrix.com/trac/browser/trunk/doc/mail/examples/emailserver.tac demonstrates how to set up an SMTP server with a customizable on-delivery hook.

回答4:

Alex answer is the best but was incomplete for my use case. I wanted to test SMTP as part of a unit test which meant building the fake SMTP server inside my test objects and the server would not terminate the asyncio thread so I had to add a line to set it to a daemon thread to allow the rest of the unit test to complete without blocking waiting for that asyncio thread to join. I also added in complete logging of all email data so that I could assert anything sent through the SMTP.

Here is my fake SMTP class:

class TestingSMTP(smtpd.SMTPServer):
    def __init__(self, *args, **kwargs):
        super(TestingSMTP, self).__init__(*args, **kwargs)
        self.emails = []

    def process_message(self, peer, mailfrom, rcpttos, data, **kwargs):
        msg = {'peer': peer,
               'mailfrom': mailfrom,
               'rcpttos': rcpttos,
               'data': data}
        msg.update(kwargs)
        self.emails.append(msg)


class TestingSMTP_Server(object):

    def __init__(self):
        self.smtp = TestingSMTP(('0.0.0.0', 25), None)
        self.thread = threading.Thread()

    def start(self):
        self.thread = threading.Thread(target=asyncore.loop, kwargs={'timeout': 1})
        self.thread.daemon = True
        self.thread.start()

    def stop(self):
        self.smtp.close()
        self.thread.join()

    def count(self):
        return len(self.smtp.emails)

    def get(self):
        return self.smtp.emails

And here is how it is called by the unittest classes:

smtp_server = TestingSMTP_Server()
smtp_server.start()

# send some emails

assertTrue(smtp_server.count() == 1) # or however many you intended to send
assertEqual(self.smtp_server.get()[0]['mailfrom'], '[e-mail]')

# stop it when done testing
smtp_server.stop()

回答5:

In case anyone else needs this fleshed out a bit more, here's what I ended up using. This uses smtpd for the email server and smtpblib for the email client, with Flask as a http server [gist]:

app.py

from flask import Flask, render_template
from smtp_client import send_email
from smtp_server import SMTPServer

app = Flask(__name__)

@app.route('/send_email')
def email():
  server = SMTPServer()
  server.start()
  try:
    send_email()
  finally:
    server.stop()
  return 'OK'

@app.route('/')
def index():
  return 'Woohoo'

if __name__ == '__main__':
  app.run(debug=True, host='0.0.0.0')

smtp_server.py

# smtp_server.py
import smtpd
import asyncore
import threading

class CustomSMTPServer(smtpd.SMTPServer):
  def process_message(self, peer, mailfrom, rcpttos, data):
    print('Receiving message from:', peer)
    print('Message addressed from:', mailfrom)
    print('Message addressed to:', rcpttos)
    print('Message length:', len(data))
    return

class SMTPServer():
  def __init__(self):
    self.port = 1025

  def start(self):
    '''Start listening on self.port'''
    # create an instance of the SMTP server, derived from  asyncore.dispatcher
    self.smtp = CustomSMTPServer(('0.0.0.0', self.port), None)
    # 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
    kwargs = {'timeout':1, 'use_poll': True}
    self.thread = threading.Thread(target=asyncore.loop, kwargs=kwargs)
    self.thread.start()

  def stop(self):
    '''Stop listening to self.port'''
    # close the SMTPserver to ensure no channels connect to asyncore
    self.smtp.close()
    # now it is safe to wait for asyncore.loop() to exit
    self.thread.join()

  # check for emails in a non-blocking way
  def get(self):
    '''Return all emails received so far'''
    return self.smtp.emails

if __name__ == '__main__':
  server = CustomSMTPServer(('0.0.0.0', 1025), None)
  asyncore.loop()

smtp_client.py

import smtplib
import email.utils
from email.mime.text import MIMEText

def send_email():
  sender='[e-mail]'
  recipient='[e-mail]'

  msg = MIMEText('This is the body of the message.')
  msg['To'] = email.utils.formataddr(('Recipient', recipient))
  msg['From'] = email.utils.formataddr(('Author', '[e-mail]'))
  msg['Subject'] = 'Simple test message'

  client = smtplib.SMTP('127.0.0.1', 1025)
  client.set_debuglevel(True) # show communication with the server
  try:
    client.sendmail('[e-mail]', [recipient], msg.as_string())
  finally:
    client.quit()

Then start the server with python app.py and in another request simulate a request to /send_email with curl localhost:5000/send_email. Note that to actually send the email (or sms) you'll need to jump through other hoops detailed here: https://blog.codinghorror.com/so-youd-like-to-send-some-email-through-code/.