Examples

A collection examples on how to use the Concurrence Framework

A Simple Chat Server

In an example of how Tasks and Messages are typically used together. This example implements a simple multi-user chat server:

from concurrence import dispatch, Tasklet, Message
from concurrence.io import BufferedStream, Socket, Server

class MSG_WRITE_LINE(Message): pass
class MSG_QUIT(Message): pass
class MSG_LINE_READ(Message): pass

connected_clients = set() #set of currently connected clients (tasks)
        
def handle(client_socket):
    """handles a single client connected to the chat server"""
    stream = BufferedStream(client_socket)

    client_task = Tasklet.current() #this is the current task as started by server
    connected_clients.add(client_task)
    
    def writer():
        for msg, args, kwargs in Tasklet.receive():
            if msg.match(MSG_WRITE_LINE):
                stream.writer.write_bytes(args[0] + '\n')
                stream.writer.flush()
            
    def reader():
        for line in stream.reader.read_lines():
            line = line.strip()
            if line == 'quit': 
                MSG_QUIT.send(client_task)()
            else:
                MSG_LINE_READ.send(client_task)(line)
    
    reader_task = Tasklet.new(reader)()
    writer_task = Tasklet.new(writer)()

    MSG_WRITE_LINE.send(writer_task)("type 'quit' to exit..")
    
    for msg, args, kwargs in Tasklet.receive():
        if msg.match(MSG_QUIT):
            break
        elif msg.match(MSG_LINE_READ):
            #a line was recv from our client, multicast it to the other clients
            for task in connected_clients:
                if task != client_task: #don't echo the line back to myself
                    MSG_WRITE_LINE.send(task)(args[0])
        elif msg.match(MSG_WRITE_LINE):
            MSG_WRITE_LINE.send(writer_task)(args[0])
        
    connected_clients.remove(client_task)
    reader_task.kill()
    writer_task.kill()
    client_socket.close()
           
def server():
    """accepts connections on a socket, and dispatches
    new tasks for handling the incoming requests"""
    print 'listening for connections on port 9010'
    Server.serve(('localhost', 9010), handle)

if __name__ == '__main__':
    dispatch(server)

You can start it with stackless chat.py and then start 1 or more sessions using telnet localhost 9010. If you type some message in one session, it will be multi-cast to all other currently connected sessions. A quick overview of the code:

• First a new tcp ‘server’ is started.
• For each incoming connection a new client task is created that will execute the ‘handle’ function.
• The client task will in turn start 2 child tasks; ‘reader’ and ‘writer’ that are responsible for reading and writing lines from/to the corresponding client.
• The client task then will enter a ‘receive’ loop and repond to messages until it receives the quit message.
• Coorperation between the tasks is handled by the 3 messages defined at the top
• Incoming lines are read by ‘reader’, it messages the client_task, which in turn multi-casts the line to all other connected client_tasks.