How to handle asyncore within a class in python, without blocking anything?

Question

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.

Buyer

I feel some confusion there. What is the problem with asyncore.loop() blocking ? Do you understand why you call the loop function and what it does ? – Hickie

@mmgp: The problem with asyncore.loop() is that it is blocking. I want to be able to use the class at any time within some other code. On the other side, I am not an expert on asyncore.loop(), but AFAIK it handles internal the select.select, which is looking e.g. for incoming SMTP messages on port 25, in this case. – Buyer

have you used GUI toolkits ? Basically all of them are based on event loops. You have to arrange things such that they produce events to be handled by the "event loop". The confusion that I mentioned is because you seem unaware of how to use an event loop, is that the case ? – Hickie

@Hickie Yes, I am quite unaware of how I use an event loop. Therefore I asked this question to get a meaningful and helpful answer to my problem in case someone knows about event loops and can provide a solution for my problem. – Buyer

@Buyer - It's unfortunate that you seem to have the same problem I do with people thinking that because things work one way for them that that's the way it should work for everybody (i.e. 'just use the event loop' or 'using something extra won't hurt')... – Julissajulita

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

Buyer

@Buyer is there any chance you could elaborate this to show how to use this in conjunction with the smtplib to send an email? I'd be grateful if you could... – Borscht

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

Anticipant

Ok, I was assuming my approach was a bit too large. But to followup, in order to end this thread I need to close all asyncore channels. How to do that? How can I 'stop' asyncore? How can I stop this thread? (As part of my actual question) – Buyer

@Buyer The call to asyncore.loop will unblock when all asyncore channels are closed. Channels in this context referring to any subclasses of asyncore.dispatcher or asynchat.async_chat. In your case you simply need to call server.close() at the time you want to exit. – Anticipant

I tried that in the above example code and put in the following in the stop method: self.close(); self._thread.join(). It does not work. It hangs at thread.join which implies that there is another (?) channel open? Which one? How to find them? – Buyer

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

Asparagus

sorry, I do not want to use something extra. I should have been more explicit. – Buyer

It's not likely that using something "extra" will actually hurt your project. It's much more likely it will make it a lot better. You're already using Python, that's "extra", too. It's okay, people can install software. You can even make super slick packages to install it for them. By limiting yourself to just what's in the Python standard library, you're cutting off a massive amount of incredibly useful functionality, Twisted being just one small example of it. In the end, you're wasting your own time (and the time of people who choose to answer your questions) and hurting your project. – Asparagus

Answer 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'], '[email protected]')

# stop it when done testing
smtp_server.stop()

Recrement

Answer 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='[email protected]'
  recipient='[email protected]'

  msg = MIMEText('This is the body of the message.')
  msg['To'] = email.utils.formataddr(('Recipient', recipient))
  msg['From'] = email.utils.formataddr(('Author', '[email protected]'))
  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('[email protected]', [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/.

Borscht