moduleNetwork.Server.ScalableServer(-- * Introduction-- $introrunServer,RequestPipeline(..),RequestCreator,RequestProcessor)whereimportNetwork.SocketimportNetwork.Socket.Enumerator(enumSocket)importqualifiedNetwork.Socket.ByteStringasBinSockimportNetwork.BSDimportControl.Exception(finally)importControl.Monad(forever,liftM,replicateM,void)importControl.Monad.Trans(liftIO)importControl.Concurrent(forkIO,threadDelay)importControl.Concurrent.STM(newTChanIO,writeTChan,readTChan,atomically,TChan)importData.Enumerator(($$),run_)importqualifiedData.EnumeratorasEimportData.Attoparsec.Enumerator(iterParser)importBlaze.ByteString.Builder(toByteStringIO,Builder)importData.Enumerator(yield,continue,Iteratee,Stream(..))importqualifiedData.AttoparsecasAttoimportqualifiedData.Attoparsec.Char8asAttoCimportqualifiedData.ByteString.Char8asSimportqualifiedData.ByteStringasWSimportqualifiedData.ByteString.Lazy.Char8asB-- $intro---- 'ScalableServer' is a library that attempts to capture current best-- practices for writing fast/scalable socket servers in Haskell.---- Currently, that involves providing the right glue for hooking up-- to enumerator/attoparsec-enumerator/blaze-builder and network-bytestring---- It provides a relatively simple parse/generate toolchain for-- plugging into these engines---- Servers written using this library support "pipelining"; that is, a client-- can issue many requests serially before the server has responded to the-- first---- Server written using this library also can be invoked with +RTS -NX-- invocation for multicore support-- |The 'RequestPipeline' acts as a specification for your service,-- indicating both a parser/request object generator, the RequestCreator,-- and the processor of these requests, one that ultimately generates a-- response expressed by a blaze 'Builder'dataRequestPipelinea=RequestPipeline(RequestCreatora)(RequestProcessora)-- |The RequestCreator is an Attoparsec parser that yields some request-- object 'a'typeRequestCreatora=Atto.Parsera-- |The RequestProcessor is a function in the IO monad (for DB access, etc)-- that returns a builder that can generate the responsetypeRequestProcessora=a->IOBuildertypeRequestChannela=TChan(Maybea)-- |Given a pipeline specification and a port, run TCP traffic using the-- pipeline for parsing, processing and response.---- Note: there is currently no way for a server to specific the socket-- should be disconnectedrunServer::RequestPipelinea->PortNumber->IO()runServerpipeport=doproto<-getProtocolNumber"tcp"s<-socketAF_INETStreamprotosetSocketOptionsReuseAddr1bindSockets(SockAddrInetportiNADDR_ANY)serverListenLooppipesserverListenLoop::RequestPipelinea->Socket->IO()serverListenLooppipes=dolistens100forever$do(c,_)<-acceptsforkIO$connHandlerpipecconnHandler::RequestPipelinea->Socket->IO()connHandler(RequestPipelinereqParsereqProc)s=dochan<-newTChanIO(doletenum=enumSocket4096sletparser=iterParserreqParsevoid$forkIO$processRequestschanreqProcsvoid$run_(enum$$E.sequenceparser$$requestHandlerchans))`finally`((atomically$writeTChanchanNothing)>>sCloses)requestHandler::RequestChannela->Socket->IterateeaIO()requestHandlerchans=docontinuerequestConsumewhererequestConsume(Chunksmrs)=doliftIO$mapM_(\m->atomically$writeTChanchan$Justm)mrscontinuerequestConsumerequestConsumeEOF=doyield()EOFprocessRequests::RequestChannela->RequestProcessora->Socket->IO()processRequestschanprocs=donext<-atomically$readTChanchancasenextofJusta->doresp<-proca-- XXX handle exceptions?-- XXX eventually, pipeline to enumerator?toByteStringIO(BinSock.sendAlls)$respprocessRequestschanprocsNothing->return()