Extending Asterisk with HAppS

I just glued two of my favorite technologies together, Asterisk (the opensource PBX/VoIP/etc system) and HAppS (the Haskell Application Server framework).

If you have heard of HAppS, but never used it, you may have the impression that HAppS is a web development platform -- but that is not quite correct. HAppS is actually a collection of several different server components which can be combined together or used separately.

In this post, I will show how to build as simple FastAGI server on top of the HAppS-State component. We will not be using the web component (HAppS-Server), which is the part that provides HTTP, templating, cookies, etc. This post assumes no prior knowledge of HAppS or AGI.

What You Will Need

If you want to build this demo you will need:

• the latest version of HAppS-State and it's dependencies.
  
• Asterisk (I think any version later than 1.0 should work. I use 1.4.17 from Ubuntu Hardy)
  
• The haskell AGI library
  
• darcs get http://src.seereason.com/fastagi-hitcounter
  

Run make HitCounter.hs to produce a nice, clean
.hs file from the .lhs file.

FastAGI

The asterisk server can be extended by using the Asterisk Gateway Interface (AGI). AGI provides the functionality you need to do stuff like "Please enter your 16-digit account number."

An AGI script is a standalone program you write (in a language of your choice). Asterisk communications with your AGI script by running it directly, and writing to its stdin and reading from its stdout. The AGI protocol consists of simple commands and
responses which are human readable text.

Asterisk also has the option of communicating with your AGI script remotely via TCP instead of directly running a local program. This feature is known as FastAGI. The commands and responses are identical to normal AGI, the only differences are:

• The communication channel is setup via TCP instead of forking off a local process
  
• Some extra AGI variables are passed in by FastAGI
  

There is one additional importing difference, which is more of a side-effect. When using plain-old AGI, a new process will be spawned for each call. When using FastAGI, a new TCP connection will be opened -- typically to a single, long running server process. So, with AGI you will need to worry about how to provide communication and synchronization between multiple processes, but with FastAGI, you can just use threads.

HAppS-State

The HAppS State component provides in-memory state with ACID guarantees. It uses write ahead logging and checkpointing to ensure the state can be restored from disk in the event of a power outage, and also provides multimaster replication. Unlike a traditional relational database, HAppS-State works directly with native, arbitrary Haskell data types. This means you don't have to figure out how to get your beautiful data structures wedged into a relational database just to get ACID guarantees and replication.

Example Application

The remainder of the post is a simple example which implements a hit-counter. When you call the phone number, it tells you what caller number you are. I won't go into too much detail about the HAppS State portion, since this post is supposed to show how to integrate AGI, not how to use HAppS State.

> {-# LANGUAGE TemplateHaskell, UndecidableInstances, TypeFamilies, 
>  TypeSynonymInstances, FlexibleInstances, DeriveDataTypeable, 
>  MultiParamTypeClasses, TypeOperators, GeneralizedNewtypeDeriving #-}

> module Main where
>
> import Control.Concurrent
> import Control.Monad
> import Control.Monad.Reader
> import Control.Monad.State
> import Control.Monad.Trans
> import HAppS.Data
> import HAppS.State
> import Network
> import Network.AGI
> import System.Random
> import System.Posix.Unistd

The first thing we do is define the type we will use to store our
persistent state (aka, our "database schema"). The deriveAll is
similar to deriving (Eq, Ord, Read, Show, Num, Enum, Default,
Data, Typeable). Since there is no way to extend
deriving, we have to use Template Haskell to add support
for deriving Default.

>
> $(deriveAll [''Eq,''Ord,''Read,''Show,''Num,''Enum,''Default]
>  [d|
>      newtype HitCounter = HitCounter { hits :: Integer }
>   |])

deriveSerialize is part of the magic that allows HitCounter
to be serialized to disk or replicated between servers.

>
> $(deriveSerialize ''HitCounter)

The Version instance is used to migrate the
old data, if we modify the HitCounter data structure. That
is a subject for a different tutorial.

> instance Version HitCounter

Next we define a function which modifies the global state
(HitCounter). This function while be run
atomically. This means that there is no race condition between the
get and the put. The get and
put functions come from
Control.Monad.State.

>
> -- |increment hit counter, and return new value
> incHits :: Update HitCounter Integer
> incHits = 
>     do hc <- fmap succ get
>        put hc
>        return (hits hc)

This is the magic which converts the incHits function
into an atomic action for updating the global state.

>
> $(mkMethods ''HitCounter
>    [ 'incHits
>    ])

Next we define our top-level component which uses the global state. A more complex application might use a bunch of independent components similar to HitCounter. This allows us to easily build things like session support, user accounts, etc, in third party reusable libraries. I believe it also makes atomic actions finer grained and makes it possible to support shards.

>
> instance Component HitCounter where
>     type Dependencies HitCounter = End
>     initialValue = HitCounter 0

> entryPoint :: Proxy HitCounter
> entryPoint = Proxy

The main function starts up the state engine, forks off the fastAGI server, waits for a shutdown signal (for example, ^C), and then cleanly shuts down the state engine. The fastAGI function comes from the Haskell AGI library, and is in no way HAppS specific.

>
> main :: IO ()
> main =
>     do control <- startSystemState entryPoint
>        tid <- forkIO $ fastAGI Nothing agiMain
>        putStrLn "running..."
>        waitForTermination
>        killThread tid
>        shutdownSystem control

Here is our simple AGI application. It

1. answers the call
   
2. waits a second to give the caller time to finish setting up their end of the call
   
3. increments the hit counter
   
4. plays a pre-recorded file which says, "You are currently caller number"
   
5. says the caller number
   
6. plays a pre-recorded file which says "Goodbye."
   
7. hangs up
   

The functions answer, streamFile,
sayNumber, and hangUp come from the AGI
library.

The update IncHits call is our database query. Note
that we don't call the incHits function
directly. Instead we call update and pass it the value
IncHits. The IncHits type was created for
us automatically by the call to mkMethods we made
earlier.

> 
> agiMain :: HostName -> PortNumber -> AGI ()
> agiMain hostname portNum =
>     do answer
>        liftIO $ sleep 1 -- give the caller time to get their end of the call setup
>        h <- update IncHits
>        streamFile "queue-thereare" [] Nothing
>        sayNumber h []
>        streamFile "vm-goodbye" [] Nothing
>        hangUp Nothing
>        return ()

Hooking it up

To test the application, first we need to update the asterisk dialplan to call our AGI application. Something like this should do the trick (be sure to reload the dialplan after modifying extensions.conf):

[default]
exten => 31415,1,AGI(agi://127.0.0.1)

Next we start our AGI application server:

 $ runhaskell HitCounter.lhs

And finally, we dial 31415 and hope the magic happens.

Summary

The above code is a good starting template for a more interesting AGI application. Note that caller number is a bit fuzzy. The caller number is determined by who gets to the update function first -- which could be different from who actually connected to the asterisk server first.

Also, when calling a FastAGI application, it is possible to pass in a PATH and query string. The Haskell AGI library makes this information available, but does not provide any special mechanisms for doing something useful with it. This is likely to change in the future.