Idiomatic asynchronous design

Question

Are there any sorts of useful idioms I can make use of when writing an API that is asynchronous? I would like to standardize on something as I seem to be using a few different styles throughout. It seems hard to make asynchronous code simple; I suppose this is because asynchronous operations are anything but.

At the most basic level, the user of the API must be able to:

1. Have data pushed to them as it becomes available
2. Check the status of the asynchronous operation
3. Be notified of errors that occur
4. Wait for completion (converting the asynchronous operation to a synchronous one)

My classes support several asynchronous operations. I have been putting some of the status/error callbacks in the class around it, but the class is becoming gunked up with a lot of incidental fields, as well as getting too large. I am curious if anyone has used an asynchronous API they found to be well-organized. I have looked at .NET's Begin/EndAsyncOperation + AsyncResult design, as well as some classes in Java (e.g. Future).

This is being written in Python, so it remains very flexible. There is a caveat: some of these asynchronous operations are being marshaled to a remote machine and executed over there. Thus, not every operation necessarily executes in a separate thread.

Answer 1

You may want to look at Python Twisted. It is a nice Reactor based API that supports asynchronous operations. Proactor is the common term for asynchronous completion handler like frameworks.

Answer 2

Also have a look at the Asynchronous Completion Token and ActiveObject patterns.

Answer 3

This sounds like the Observer design pattern. link.

Your client object is an Observer. Your API belongs to an object that's Observable.

Each client (in Java parlance) implements the Observer interface. In Python, it's a matter of each client offering a number of methods that your observable will use.

class SomeClientInterface( object ):
    def update( self, source, data ):
        # handle data being pushed from Observable source
    def error( self, from, status ):
        # handle error in Observable source

Your Observable object has a way for Observers to register and do other things.

class Observable( object ):
    def __init__( self ):
        self.clients= set()
    def register( self, observer ):
        self.clients.add( observer )
    def whenSomethingHappens( self ):
        # doing work
        if itAllWentToHell:
            for c in self.clients:
                c.error( self, "some status object" )
        else:
            for c in self.clients:
                c.update( self, the pushed data )
    def waitFor( self ):
        # observers are waiting...
        return theData
    def status( self ):
        return self.currentState