A chat server using fibra.

The following code is a very simple chat server implemented using cooperative threads in the fibra 0.01 framework. Explanation follows below.

import fibra
import fibra.plugins.network as network
import fibra.plugins.tasks as tasks


class Chatter(object):
    def __init__(self, address):
        self.address = address
        self.members = {}

    def listener(self):
        while True:
            conn = (yield network.listen(self.address))
            yield tasks.spawn(self.login(conn))

    def login(self, conn):
        handle = None
        while handle is None:
            yield network.send(conn, 'What is your handle?')
            handle = (yield network.receive(conn))
            if handle in self.members:
                yield network.send(conn, 'That handle is already taken.')
                handle = None
        self.members[handle] = conn
        yield tasks.spawn(self.chat(conn, handle))

    def lost_conn(self, conn, handle):
        if handle in self.members:
            self.members.pop(handle)
        yield self.broadcast('%s has left the chat.' % handle)

    def broadcast(self, text):
        for member, conn in self.members.items():
            try:
                yield network.send(conn, text)
            except network.NetworkError:
                yield tasks.on_finish(self.lost_conn(conn, member))

    def closed_socket(self, conn, handle):
        yield network.on_lost_connection(conn)
        yield self.lost_conn(conn, handle)

    def chat(self, conn, handle):
        yield tasks.on_finish(self.lost_conn(conn, handle))
        yield tasks.spawn(self.closed_socket(conn, handle))
        yield self.broadcast('%s has joined the chat.' % handle)
        data = ''
        while True:
            try:
                data = yield network.receive(conn)
            except network.NetworkError:
                break
            if data == '/quit': break
            yield self.broadcast('%s says: %s' % (handle, data))
        yield network.close(conn)

if __name__ == "__main__":
    chatter = Chatter(('localhost',1980))
    s = fibra.Schedule()
    s.register_plugin(network.NetworkPlugin())
    s.install(chatter.listener())
    while s.tick(): pass

The Chatter class has 6 methods, which are all python generators, which in the context of fibra, I call tasklets.

At the bottom of the code, a fibra Schedule is created, and the NetworkPlugin is registered. The NetworkPlugin allows tasklets to yield certain values which perform network related operations. The main tasklet, chatter.listener, is installed into the scheduler, then the schedule is continually ticked in a while loop. The while loop will finish when there are no more tasklets to run.

So, what does the listener method do? It creates a tasklet, and yields the network.listen object, which will return a new connection when someone connects to the address passed into the network.listen call. This is a non-blocking operation. When the connection object is returned, the listener method spawns a new tasklet (self.login) with the connection, then goes back to listening for another new connection. The self.login tasklet will continue to run concurrently while the listen method is waiting.

The login method sends a prompt to the new connection, asking for a handle to identify the user. If the handle has not already been used, it spawns a chat tasklet and then exits.

The first line of the chat tasklet schedules another tasklet (self.lost_conn) which will be run when the chat tasklet finishes. The second line spawns a tasklet (self.closed_socket) which waits for the socket to close unexpectedly. The chat tasklet then broadcasts a message to any users who are already logged in. It then loops, sending chat messages to all users as they are received. Finally, if the tasklet receives a '/quit' line, it breaks out of the loop and closes the socket and finishes. At this point, the self.lost_conn task is awakened and runs.

If you want to test this code yourself, start the server, and telnet to localhost 1980.