Adds channel implementation user STM.
Adds join/leave/tell support for channels.pull/1/head
parent
478a91d93f
commit
affa18de55
|
|
@ -1,9 +1,9 @@
|
|||
module Link.Client where
|
||||
|
||||
import Control.Concurrent.STM (STM, writeTChan, readTChan, atomically, orElse)
|
||||
import Control.Concurrent.STM
|
||||
import Control.Concurrent hiding (forkFinally)
|
||||
import Control.Exception hiding (handle)
|
||||
import Control.Monad (void, forever)
|
||||
import Control.Monad (void, forever, when, unless, forM_)
|
||||
import Data.Time (getCurrentTime, diffUTCTime)
|
||||
import System.IO (hGetLine)
|
||||
import System.Timeout (timeout)
|
||||
|
|
@ -27,6 +27,9 @@ sendMessage Client {..} = writeTChan clientChan
|
|||
sendMessageIO :: Client -> Message -> IO ()
|
||||
sendMessageIO client = atomically . sendMessage client
|
||||
|
||||
tellMessage :: Channel -> Message -> STM ()
|
||||
tellMessage Channel {..} = writeTChan channelChan
|
||||
|
||||
sendResponse :: Client -> Message -> IO ()
|
||||
sendResponse Client {..} = printToHandle clientHandle . formatMessage
|
||||
|
||||
|
|
@ -35,7 +38,12 @@ runClient Server {..} client@Client {..} = do
|
|||
clientAlive <- newMVar True
|
||||
pingThread <- forkIO $ ping clientAlive `finally` killClient clientAlive
|
||||
commandThread <- forkIO $ readCommands `finally` killClient clientAlive
|
||||
run clientAlive `finally` (killThread pingThread >> killThread commandThread)
|
||||
run clientAlive `finally` do
|
||||
killThread pingThread
|
||||
killThread commandThread
|
||||
clientChannelMap <- readTVarIO clientChannelChans
|
||||
forM_ (Map.keys clientChannelMap) $ \channelName ->
|
||||
handleMessage (Leave channelName) clientAlive
|
||||
where
|
||||
pingDelay = 120
|
||||
pingDelayMicros = pingDelay * 1000 * 1000
|
||||
|
|
@ -56,7 +64,9 @@ runClient Server {..} client@Client {..} = do
|
|||
if not alive
|
||||
then printf "Closing connection: %s\n" (userName clientUser)
|
||||
else do
|
||||
r <- try . timeout pingDelayMicros . atomically . readTChan $ clientChan
|
||||
r <- try . timeout pingDelayMicros . atomically $ do
|
||||
clientChannelMap <- readTVar clientChannelChans
|
||||
foldr (orElse . readTChan) retry $ clientChan : Map.elems clientChannelMap
|
||||
case r of
|
||||
Left (e :: SomeException) -> printf "Exception: %s\n" (show e)
|
||||
Right g -> do
|
||||
|
|
@ -77,27 +87,52 @@ runClient Server {..} client@Client {..} = do
|
|||
case Map.lookup user clientMap of
|
||||
Nothing -> sendResponse client $ NoSuchUser (userName user)
|
||||
Just client' -> sendMessageIO client' $ MsgReply clientUser msg
|
||||
|
||||
handleMessage Pong _ = do
|
||||
now <- getCurrentTime
|
||||
void $ swapMVar clientPongTime now
|
||||
|
||||
handleMessage Quit clientAlive = killClient clientAlive
|
||||
handleMessage (Join channelName) _ = do
|
||||
modifyMVar_ serverChannels $ \channelMap -> do
|
||||
case Map.lookup channelName channelMap of
|
||||
Just (Channel {channelUsers}) -> do
|
||||
modifyMVar_ channelUsers $ return . Set.insert clientUser
|
||||
return channelMap
|
||||
Nothing -> do
|
||||
|
||||
handleMessage (Join channelName) _ = atomically $ do
|
||||
clientChannelMap <- readTVar clientChannelChans
|
||||
unless (Map.member channelName clientChannelMap) $ do
|
||||
channelMap <- readTVar serverChannels
|
||||
channel@Channel {channelChan} <- case Map.lookup channelName channelMap of
|
||||
Just (channel@Channel {channelUsers}) -> do
|
||||
modifyTVar' channelUsers $ Set.insert clientUser
|
||||
return channel
|
||||
Nothing -> do
|
||||
channel <- newChannel channelName $ Set.singleton clientUser
|
||||
return $ Map.insert channelName channel channelMap
|
||||
handleMessage (Leave channelName) _ = do
|
||||
modifyMVar_ serverChannels $ \channelMap -> do
|
||||
case Map.lookup channelName channelMap of
|
||||
Just (Channel {channelUsers}) -> do
|
||||
modifyMVar_ channelUsers $ return . Set.delete clientUser
|
||||
users <- readMVar channelUsers
|
||||
return $ if Set.null users
|
||||
then Map.delete channelName channelMap
|
||||
else channelMap
|
||||
Nothing -> return channelMap
|
||||
modifyTVar' serverChannels $ Map.insert channelName channel
|
||||
return channel
|
||||
clientChannelChan <- dupTChan channelChan
|
||||
modifyTVar' clientChannelChans $ Map.insert channelName clientChannelChan
|
||||
tellMessage channel $ Joined channelName clientUser
|
||||
|
||||
handleMessage (Leave channelName) _ = atomically $ do
|
||||
channelMap <- readTVar serverChannels
|
||||
case Map.lookup channelName channelMap of
|
||||
Just (channel@Channel {channelUsers}) -> do
|
||||
modifyTVar' channelUsers $ Set.delete clientUser
|
||||
users <- readTVar channelUsers
|
||||
when (Set.null users) $
|
||||
modifyTVar' serverChannels $ Map.delete channelName
|
||||
modifyTVar' clientChannelChans $ Map.delete channelName
|
||||
tellMessage channel $ Leaved channelName clientUser
|
||||
Nothing -> return ()
|
||||
|
||||
handleMessage (Names channelName) _ = atomically $ do
|
||||
channelMap <- readTVar serverChannels
|
||||
users <- case Map.lookup channelName channelMap of
|
||||
Just (Channel {channelUsers}) -> readTVar channelUsers
|
||||
Nothing -> return Set.empty
|
||||
sendMessage client $ NamesReply channelName users
|
||||
|
||||
handleMessage (Tell channelName msg) _ = atomically $ do
|
||||
channelMap <- readTVar serverChannels
|
||||
case Map.lookup channelName channelMap of
|
||||
Just channel -> tellMessage channel $ TellReply channelName clientUser msg
|
||||
Nothing -> return ()
|
||||
|
||||
handleMessage message _ = sendResponse client message
|
||||
|
|
|
|||
|
|
@ -2,22 +2,30 @@ module Link.Protocol where
|
|||
|
||||
import Text.Printf (printf)
|
||||
|
||||
import qualified Data.Set as Set
|
||||
|
||||
import Link.Types
|
||||
|
||||
parseCommand :: String -> Maybe Message
|
||||
parseCommand command = case words command of
|
||||
["PONG"] -> Just Pong
|
||||
"MSG" : userName : msg -> Just $ Msg (User userName) (unwords msg)
|
||||
["QUIT"] -> Just $ Quit
|
||||
"JOIN" : channelName -> Just $ Join (unwords channelName)
|
||||
"LEAVE" : channelName -> Just $ Leave (unwords channelName)
|
||||
_ -> Nothing
|
||||
["PONG"] -> Just Pong
|
||||
"MSG" : userName : msg -> Just $ Msg (User userName) (unwords msg)
|
||||
"TELL" : channelName : msg -> Just $ Tell channelName (unwords msg)
|
||||
["QUIT"] -> Just Quit
|
||||
"JOIN" : channelName -> Just $ Join (unwords channelName)
|
||||
"LEAVE" : channelName -> Just $ Leave (unwords channelName)
|
||||
"NAMES" : channelName -> Just $ Names (unwords channelName)
|
||||
_ -> Nothing
|
||||
|
||||
formatMessage :: Message -> String
|
||||
formatMessage (MsgReply user msg) = printf "MSG %s %s" (userName user) msg
|
||||
formatMessage (TellReply channelName user msg) =
|
||||
printf "TELL %s %s %s" channelName (userName user) msg
|
||||
formatMessage (NameInUse name) = printf "NAMEINUSE %s" name
|
||||
formatMessage (Connected name) = printf "CONNECTED %s" name
|
||||
formatMessage Ping = "PING"
|
||||
formatMessage (NoSuchUser name) = printf "NOSUCHUSER %s" name
|
||||
formatMessage (Joined channelName user) = printf "JOINED %s %s" channelName (userName user)
|
||||
formatMessage (Leaved channelName user) = printf "LEFT %s %s" channelName (userName user)
|
||||
formatMessage (NamesReply channelName users) =
|
||||
printf "NAMES %s %s" channelName . unwords . map userName . Set.toList $ users
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
module Link.Types where
|
||||
|
||||
import Control.Concurrent
|
||||
import Control.Concurrent.STM (TVar, TChan, newTVarIO, newTChanIO, newBroadcastTChanIO)
|
||||
import Control.Concurrent.STM (STM, TVar, TChan, newTVar, newTVarIO, newTChanIO, newBroadcastTChan)
|
||||
import Data.Time (UTCTime, getCurrentTime)
|
||||
import System.IO (Handle)
|
||||
|
||||
|
|
@ -15,52 +15,58 @@ data User = User { userName :: !UserName }
|
|||
deriving (Show, Eq, Ord)
|
||||
|
||||
data Client = Client {
|
||||
clientUser :: !User
|
||||
, clientHandle :: !Handle
|
||||
, clientChan :: TChan Message
|
||||
, clientPongTime :: MVar UTCTime
|
||||
clientUser :: !User
|
||||
, clientHandle :: !Handle
|
||||
, clientChan :: TChan Message
|
||||
, clientPongTime :: MVar UTCTime
|
||||
, clientChannelChans :: TVar (Map.Map ChannelName (TChan Message))
|
||||
}
|
||||
|
||||
newClient :: User -> Handle -> IO Client
|
||||
newClient user handle = do
|
||||
clientChan <- newTChanIO
|
||||
now <- getCurrentTime
|
||||
clientPongTime <- newMVar now
|
||||
return $ Client user handle clientChan clientPongTime
|
||||
clientChan <- newTChanIO
|
||||
now <- getCurrentTime
|
||||
clientPongTime <- newMVar now
|
||||
clientChannelChans <- newTVarIO Map.empty
|
||||
return $ Client user handle clientChan clientPongTime clientChannelChans
|
||||
|
||||
data Channel = Channel {
|
||||
channelName :: !ChannelName
|
||||
, channelUsers :: MVar (Set.Set User)
|
||||
, channelUsers :: TVar (Set.Set User)
|
||||
, channelChan :: TChan Message
|
||||
}
|
||||
|
||||
newChannel :: ChannelName -> Set.Set User -> IO Channel
|
||||
newChannel :: ChannelName -> Set.Set User -> STM Channel
|
||||
newChannel channelName users = do
|
||||
channelUsers <- newMVar users
|
||||
channelChan <- newBroadcastTChanIO
|
||||
channelUsers <- newTVar users
|
||||
channelChan <- newBroadcastTChan
|
||||
return $ Channel channelName channelUsers channelChan
|
||||
|
||||
data Server = Server {
|
||||
serverUsers :: MVar (Map.Map User Client)
|
||||
, serverChannels :: MVar (Map.Map ChannelName Channel)
|
||||
, serverChannels :: TVar (Map.Map ChannelName Channel)
|
||||
}
|
||||
|
||||
newServer :: IO Server
|
||||
newServer = do
|
||||
serverUsers <- newMVar Map.empty
|
||||
serverChannels <- newMVar Map.empty
|
||||
serverChannels <- newTVarIO Map.empty
|
||||
return $ Server serverUsers serverChannels
|
||||
|
||||
data Message = NameInUse UserName
|
||||
| Connected UserName
|
||||
| Ping
|
||||
| MsgReply User String
|
||||
| TellReply ChannelName User String
|
||||
| NoSuchUser UserName
|
||||
| Joined ChannelName User
|
||||
| Leaved ChannelName User
|
||||
| NamesReply ChannelName (Set.Set User)
|
||||
| Pong
|
||||
| Msg User String
|
||||
| Tell ChannelName String
|
||||
| Join ChannelName
|
||||
| Leave ChannelName
|
||||
| Names ChannelName
|
||||
| Quit
|
||||
deriving (Show, Eq)
|
||||
|
|
|
|||
Loading…
Reference in New Issue