metro (empty) → 0.1.0.0
raw patch · 16 files changed
+1411/−0 lines, 16 filesdep +basedep +binarydep +bytestringsetup-changed
Dependencies added: base, binary, bytestring, hashable, hslogger, mtl, transformers, unix-time, unliftio, unordered-containers
Files
- ChangeLog.md +3/−0
- LICENSE +21/−0
- README.md +55/−0
- Setup.hs +2/−0
- metro.cabal +58/−0
- src/Metro.hs +12/−0
- src/Metro/Class.hs +62/−0
- src/Metro/Conn.hs +94/−0
- src/Metro/IOHashMap.hs +86/−0
- src/Metro/Lock.hs +17/−0
- src/Metro/Node.hs +362/−0
- src/Metro/Server.hs +292/−0
- src/Metro/Session.hs +168/−0
- src/Metro/TP/BS.hs +75/−0
- src/Metro/TP/Debug.hs +46/−0
- src/Metro/Utils.hs +58/−0
+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for metro++## Unreleased changes
+ LICENSE view
@@ -0,0 +1,21 @@+The MIT License (MIT)++Copyright (c) 2019 Li Meng Jun <lmjubuntu@gmail.com>++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN+THE SOFTWARE.
+ README.md view
@@ -0,0 +1,55 @@+# metro++a simple tcp and udp socket server framework++## Quick start with example++```haskell+import Metro.Class+import Metro.Node+import Metro.TCP+import Metro.Servable+import Metro.Session (SessionT, makeResponse_)++data CustomPacket = CustomPacket { ... }+type CustomPacketId = ...++instance RecvPacket CustomPacket where+ recvPacket recv = ...+instance SendPacket CustomPacket where+ sendPacket pkt send = ...++instance GetPacketId CustomPacketId where+ getPacketId = ...+instance SetPacketId CustomPacketId where+ setPacketId k pkt = ...++type NodeId = ...+data CustomEnv = CustomEnv { ... }+type DeviceT = NodeT CustomEnv NodeId CustomPacketId CustomPacket+type DeviceEnv = NodeEnv1 CustomEnv NodeId CustomPacketId CustomPacket++sessionHandler = makeResponse_ $ \pkt -> ...++sessionGen :: IO CustomPacketId+sessionGen = ..++prepare :: Socket -> ConnEnv tp -> IO (Maybe (NodeId, CustomEnv))+prepare sock connEnv = Just ...++keepalive = 300++bind_port = "tcp://:8080"++startExampleServer = do+ sEnv <- initServerEnv "Example" (tcpConfig "tcp://:8080") sessionGen rawSocket prepare+ void $ forkIO $ startServer sEnv sessionHandler+```++more see [metro-example/src/Metro/Example.hs](https://github.com/Lupino/metro/tree/master/metro-example/src/Metro/Example.hs)++## Projects use metro++- [haskell-hole](https://github.com/Lupino/haskell-hole) A hole to pass through the gateway. haskell version+- [metro-example](https://github.com/Lupino/metro/tree/master/metro-example) An example use metro+- [haskell-periodic](https://github.com/Lupino/haskell-periodic) Periodic task system haskell client and server
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ metro.cabal view
@@ -0,0 +1,58 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: fc11297ad715dd46f77689e3e7b67ebbba22e21be3d166fc4e27a28c686e4730++name: metro+version: 0.1.0.0+synopsis: A simple tcp and udp socket server framework+description: Please see the README on GitHub at <https://github.com/Lupino/metro#readme>+category: Network,Framework+homepage: https://github.com/Lupino/metro#readme+bug-reports: https://github.com/Lupino/metro/issues+author: Lupino+maintainer: lmjubuntu@gmail.com+copyright: MIT+license: BSD3+license-file: LICENSE+build-type: Simple+extra-source-files:+ README.md+ ChangeLog.md++source-repository head+ type: git+ location: https://github.com/Lupino/metro++library+ exposed-modules:+ Metro+ Metro.Class+ Metro.Conn+ Metro.IOHashMap+ Metro.Lock+ Metro.Node+ Metro.Server+ Metro.Session+ Metro.TP.BS+ Metro.TP.Debug+ Metro.Utils+ other-modules:+ Paths_metro+ hs-source-dirs:+ src+ build-depends:+ base >=4.7 && <5+ , binary+ , bytestring+ , hashable+ , hslogger+ , mtl+ , transformers+ , unix-time+ , unliftio+ , unordered-containers+ default-language: Haskell2010
+ src/Metro.hs view
@@ -0,0 +1,12 @@+module Metro+ ( module X+ ) where++import Metro.Class as X+import Metro.Conn as X (ConnEnv, ConnT, FromConn (..), initConnEnv,+ runConnT)+import Metro.Node as X hiding (setDefaultSessionTimeout,+ setNodeMode, setSessionMode)+import Metro.Server as X+import Metro.Session as X (SessionT, makeResponse, makeResponse_,+ receive, send, sessionState)
+ src/Metro/Class.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++module Metro.Class+ ( Transport (..)+ , TransportError (..)+ , Servable (..)+ , RecvPacket (..)+ , SendPacket (..)+ , sendBinary+ , SetPacketId (..)+ , GetPacketId (..)+ ) where++import Control.Exception (Exception)+import Data.Binary (Binary, encode)+import Data.ByteString (ByteString)+import Data.ByteString.Lazy (toStrict)+import UnliftIO (MonadIO, MonadUnliftIO)++data TransportError = TransportClosed+ deriving (Show, Eq, Ord)++instance Exception TransportError++class Transport transport where+ data TransportConfig transport+ newTransport :: TransportConfig transport -> IO transport+ recvData :: transport -> Int -> IO ByteString+ sendData :: transport -> ByteString -> IO ()+ closeTransport :: transport -> IO ()++class Servable serv where+ data ServerConfig serv+ type SID serv+ type STP serv+ newServer :: MonadIO m => ServerConfig serv -> m serv+ servOnce :: MonadUnliftIO m => serv -> (Maybe (SID serv, TransportConfig (STP serv)) -> m ()) -> m ()+ onConnEnter :: MonadIO m => serv -> SID serv -> m ()+ onConnLeave :: MonadIO m => serv -> SID serv -> m ()+ servClose :: MonadIO m => serv -> m ()++class RecvPacket rpkt where+ recvPacket :: MonadIO m => (Int -> m ByteString) -> m rpkt++class SendPacket spkt where+ sendPacket :: MonadIO m => spkt -> (ByteString -> m ()) -> m ()+ default sendPacket :: (MonadIO m, Binary spkt) => spkt -> (ByteString -> m ()) -> m ()+ sendPacket = sendBinary++sendBinary :: (MonadIO m, Binary spkt) => spkt -> (ByteString -> m ()) -> m ()+sendBinary spkt send = send . toStrict $ encode spkt++class SetPacketId k pkt where+ setPacketId :: k -> pkt -> pkt++class GetPacketId k pkt where+ getPacketId :: pkt -> k
+ src/Metro/Conn.hs view
@@ -0,0 +1,94 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++module Metro.Conn+ ( ConnEnv+ , ConnT+ , FromConn (..)+ , runConnT+ , initConnEnv+ , receive+ , send+ , close+ , statusTVar+ ) where++import Control.Monad.Reader.Class (MonadReader (ask))+import Control.Monad.Trans.Class (MonadTrans, lift)+import Control.Monad.Trans.Reader (ReaderT (..), runReaderT)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B (empty)+import Metro.Class+import qualified Metro.Lock as L (Lock, new, with)+import Metro.Utils (recvEnough)+import UnliftIO++data ConnEnv tp = ConnEnv+ { transport :: tp+ , readLock :: L.Lock+ , writeLock :: L.Lock+ , buffer :: TVar ByteString+ , status :: TVar Bool+ }++newtype ConnT tp m a = ConnT { unConnT :: ReaderT (ConnEnv tp) m a }+ deriving+ ( Functor+ , Applicative+ , Monad+ , MonadTrans+ , MonadIO+ , MonadReader (ConnEnv tp)+ )++instance MonadUnliftIO m => MonadUnliftIO (ConnT tp m) where+ askUnliftIO = ConnT $+ ReaderT $ \r ->+ withUnliftIO $ \u ->+ return (UnliftIO (unliftIO u . runConnT r))+ withRunInIO inner = ConnT $+ ReaderT $ \r ->+ withRunInIO $ \run ->+ inner (run . runConnT r)++class FromConn m where+ fromConn :: Monad n => ConnT tp n a -> m tp n a++instance FromConn ConnT where+ fromConn = id++runConnT :: ConnEnv tp -> ConnT tp m a -> m a+runConnT connEnv = flip runReaderT connEnv . unConnT++initConnEnv :: (MonadIO m, Transport tp) => TransportConfig tp -> m (ConnEnv tp)+initConnEnv config = do+ readLock <- L.new+ writeLock <- L.new+ status <- newTVarIO True+ buffer <- newTVarIO B.empty+ transport <- liftIO $ newTransport config+ return ConnEnv{..}++receive :: (MonadUnliftIO m, Transport tp, RecvPacket pkt) => ConnT tp m pkt+receive = do+ ConnEnv{..} <- ask+ L.with readLock $ lift $ recvPacket (recvEnough buffer transport)++send :: (MonadUnliftIO m, Transport tp, SendPacket pkt) => pkt -> ConnT tp m ()+send pkt = do+ ConnEnv{..} <- ask+ L.with writeLock $ lift $ sendPacket pkt (liftIO . sendData transport)++close :: (MonadIO m, Transport tp) => ConnT tp m ()+close = do+ ConnEnv{..} <- ask+ atomically $ writeTVar status False+ liftIO $ closeTransport transport++statusTVar :: Monad m => ConnT tp m (TVar Bool)+statusTVar = status <$> ask
+ src/Metro/IOHashMap.hs view
@@ -0,0 +1,86 @@+module Metro.IOHashMap+ ( IOHashMap+ , newIOHashMap+ , insert+ , delete+ , lookup+ , update+ , adjust+ , alter+ , null+ , size+ , member+ , keys+ , elems+ , clear+ , toList++ , insertSTM+ , lookupSTM+ , foldrWithKeySTM+ , deleteSTM+ ) where++import Data.Hashable+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HM+import Prelude hiding (lookup, null)+import UnliftIO (MonadIO (..), STM, TVar, atomically,+ modifyTVar', newTVarIO, readTVar,+ readTVarIO)++newtype IOHashMap a b = IOHashMap (TVar (HashMap a b))++newIOHashMap :: MonadIO m => m (IOHashMap a b)+newIOHashMap = IOHashMap <$> newTVarIO HM.empty++insert :: (Eq a, Hashable a, MonadIO m) => IOHashMap a b -> a -> b -> m ()+insert (IOHashMap h) k v = atomically . modifyTVar' h $ HM.insert k v++delete :: (Eq a, Hashable a, MonadIO m) => IOHashMap a b -> a -> m ()+delete (IOHashMap h) k = atomically . modifyTVar' h $ HM.delete k++lookup :: (Eq a, Hashable a, MonadIO m) => IOHashMap a b -> a -> m (Maybe b)+lookup (IOHashMap h) k = HM.lookup k <$> readTVarIO h++adjust :: (Eq a, Hashable a, MonadIO m) => IOHashMap a b -> (b -> b) -> a -> m ()+adjust (IOHashMap h) f k = atomically . modifyTVar' h $ HM.adjust f k++update :: (Eq a, Hashable a, MonadIO m) => IOHashMap a b -> (b -> Maybe b) -> a -> m ()+update (IOHashMap h) f k = atomically . modifyTVar' h $ HM.update f k++alter :: (Eq a, Hashable a, MonadIO m) => IOHashMap a b -> (Maybe b -> Maybe b) -> a -> m ()+alter (IOHashMap h) f k = atomically . modifyTVar' h $ HM.alter f k++null :: MonadIO m => IOHashMap a b -> m Bool+null (IOHashMap h) = HM.null <$> readTVarIO h++size :: MonadIO m => IOHashMap a b -> m Int+size (IOHashMap h) = HM.size <$> readTVarIO h++member :: (Eq a, Hashable a, MonadIO m) => IOHashMap a b -> a -> m Bool+member (IOHashMap h) k = HM.member k <$> readTVarIO h++keys :: MonadIO m => IOHashMap a b -> m [a]+keys (IOHashMap h) = HM.keys <$> readTVarIO h++elems :: MonadIO m => IOHashMap a b -> m [b]+elems (IOHashMap h) = HM.elems <$> readTVarIO h++clear :: MonadIO m => IOHashMap a b -> m ()+clear (IOHashMap h) = atomically . modifyTVar' h $ const HM.empty++toList :: MonadIO m => IOHashMap a b -> m [(a, b)]+toList (IOHashMap h) = HM.toList <$> readTVarIO h++insertSTM :: (Eq a, Hashable a) => IOHashMap a b -> a -> b -> STM ()+insertSTM (IOHashMap h) k v = modifyTVar' h $ HM.insert k v++lookupSTM :: (Eq a, Hashable a) => IOHashMap a b -> a -> STM (Maybe b)+lookupSTM (IOHashMap h) k = HM.lookup k <$> readTVar h++foldrWithKeySTM :: IOHashMap a b -> (a -> b -> c -> c) -> c -> STM c+foldrWithKeySTM (IOHashMap h) f acc = HM.foldrWithKey f acc <$> readTVar h++deleteSTM :: (Eq a, Hashable a) => IOHashMap a b -> a -> STM ()+deleteSTM (IOHashMap h) k = modifyTVar' h $ HM.delete k
+ src/Metro/Lock.hs view
@@ -0,0 +1,17 @@+{-# LANGUAGE RecordWildCards #-}+module Metro.Lock+ ( Lock+ , new+ , with+ ) where+++import UnliftIO (MVar, MonadIO, MonadUnliftIO, newMVar, withMVar)++newtype Lock = Lock { un :: MVar () }++new :: MonadIO m => m Lock+new = Lock <$> newMVar ()++with :: MonadUnliftIO m => Lock -> m a -> m a+with Lock{..} = withMVar un . const
+ src/Metro/Node.hs view
@@ -0,0 +1,362 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE UndecidableInstances #-}+{-# OPTIONS_GHC -Wno-name-shadowing #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Metro.Node+ ( NodeEnv+ , NodeMode (..)+ , SessionMode (..)+ , NodeT+ , initEnv+ , withEnv++ , setNodeMode+ , setSessionMode+ , setDefaultSessionTimeout++ , runNodeT+ , startNodeT+ , startNodeT_+ , withSessionT+ , nodeState+ , stopNodeT+ , env+ , request+ , requestAndRetry++ , newSessionEnv+ , nextSessionId+ , runSessionT_++ , busy++ -- combine node env and conn env+ , NodeEnv1 (..)+ , initEnv1+ , runNodeT1+ , getEnv1++ , getTimer+ , getNodeId++ , getSessionSize+ , getSessionSize1+ ) where++import Control.Monad (forM, forever, mzero, void, when)+import Control.Monad.Reader.Class (MonadReader (ask), asks)+import Control.Monad.Trans.Class (MonadTrans (..))+import Control.Monad.Trans.Maybe (runMaybeT)+import Control.Monad.Trans.Reader (ReaderT (..), runReaderT)+import Data.Hashable+import Data.Int (Int64)+import Data.Maybe (fromMaybe, isJust)+import Metro.Class (GetPacketId, RecvPacket,+ SendPacket, SetPacketId, Transport,+ getPacketId)+import Metro.Conn (ConnEnv, ConnT, FromConn (..),+ close, receive, runConnT)+import Metro.IOHashMap (IOHashMap, newIOHashMap)+import qualified Metro.IOHashMap as HM (delete, elems, insert,+ lookup, size)+import Metro.Session (SessionEnv (sessionId), SessionT,+ feed, isTimeout, runSessionT)+import qualified Metro.Session as S (newSessionEnv, receive, send)+import Metro.Utils (getEpochTime)+import System.Log.Logger (errorM)+import UnliftIO+import UnliftIO.Concurrent (threadDelay)++data NodeMode = Single+ | Multi+ deriving (Show, Eq)++data SessionMode = SingleAction+ | MultiAction+ deriving (Show, Eq)+++data NodeEnv u nid k rpkt = NodeEnv+ { uEnv :: u+ , nodeStatus :: TVar Bool+ , nodeMode :: NodeMode+ , sessionMode :: SessionMode+ , nodeSession :: TVar (Maybe (SessionEnv u nid k rpkt))+ , sessionList :: IOHashMap k (SessionEnv u nid k rpkt)+ , sessionGen :: IO k+ , nodeTimer :: TVar Int64+ , nodeId :: nid+ , sessTimeout :: Int64+ , onNodeLeave :: TVar (Maybe (u -> IO ()))+ }++data NodeEnv1 u nid k rpkt tp = NodeEnv1+ { nodeEnv :: NodeEnv u nid k rpkt+ , connEnv :: ConnEnv tp+ }++newtype NodeT u nid k rpkt tp m a = NodeT { unNodeT :: ReaderT (NodeEnv u nid k rpkt) (ConnT tp m) a }+ deriving+ ( Functor+ , Applicative+ , Monad+ , MonadIO+ , MonadReader (NodeEnv u nid k rpkt)+ )++instance MonadUnliftIO m => MonadUnliftIO (NodeT u nid k rpkt tp m) where+ askUnliftIO = NodeT $+ ReaderT $ \r ->+ withUnliftIO $ \u ->+ return (UnliftIO (unliftIO u . runNodeT r))+ withRunInIO inner = NodeT $+ ReaderT $ \r ->+ withRunInIO $ \run ->+ inner (run . runNodeT r)++instance MonadTrans (NodeT u nid k rpkt tp) where+ lift = NodeT . lift . lift++instance FromConn (NodeT u nid k rpkt) where+ fromConn = NodeT . lift++runNodeT :: NodeEnv u nid k rpkt -> NodeT u nid k rpkt tp m a -> ConnT tp m a+runNodeT nEnv = flip runReaderT nEnv . unNodeT++runNodeT1 :: NodeEnv1 u nid k rpkt tp -> NodeT u nid k rpkt tp m a -> m a+runNodeT1 NodeEnv1 {..} = runConnT connEnv . runNodeT nodeEnv++initEnv :: MonadIO m => u -> nid -> IO k -> m (NodeEnv u nid k rpkt)+initEnv uEnv nodeId sessionGen = do+ nodeStatus <- newTVarIO True+ nodeSession <- newTVarIO Nothing+ sessionList <- newIOHashMap+ nodeTimer <- newTVarIO =<< getEpochTime+ onNodeLeave <- newTVarIO Nothing+ pure NodeEnv+ { nodeMode = Multi+ , sessionMode = SingleAction+ , sessTimeout = 300+ , ..+ }++withEnv :: (Monad m) => u -> NodeT u nid k rpkt tp m a -> NodeT u nid k rpkt tp m a+withEnv u m = do+ env0 <- ask+ fromConn $ runNodeT (env0 {uEnv=u}) m++setNodeMode :: NodeMode -> NodeEnv u nid k rpkt -> NodeEnv u nid k rpkt+setNodeMode mode nodeEnv = nodeEnv {nodeMode = mode}++setSessionMode :: SessionMode -> NodeEnv u nid k rpkt -> NodeEnv u nid k rpkt+setSessionMode mode nodeEnv = nodeEnv {sessionMode = mode}++setDefaultSessionTimeout :: Int64 -> NodeEnv u nid k rpkt -> NodeEnv u nid k rpkt+setDefaultSessionTimeout t nodeEnv = nodeEnv { sessTimeout = t }++initEnv1+ :: MonadIO m+ => (NodeEnv u nid k rpkt -> NodeEnv u nid k rpkt)+ -> ConnEnv tp -> u -> nid -> IO k -> m (NodeEnv1 u nid k rpkt tp)+initEnv1 mapEnv connEnv uEnv nid gen = do+ nodeEnv <- mapEnv <$> initEnv uEnv nid gen+ return NodeEnv1 {..}++getEnv1+ :: (Monad m, Transport tp)+ => NodeT u nid k rpkt tp m (NodeEnv1 u nid k rpkt tp)+getEnv1 = do+ connEnv <- fromConn ask+ nodeEnv <- ask+ return NodeEnv1 {..}++runSessionT_ :: Monad m => SessionEnv u nid k rpkt -> SessionT u nid k rpkt tp m a -> NodeT u nid k rpkt tp m a+runSessionT_ aEnv = fromConn . runSessionT aEnv++withSessionT+ :: (MonadUnliftIO m, Eq k, Hashable k)+ => Maybe Int64 -> SessionT u nid k rpkt tp m a -> NodeT u nid k rpkt tp m a+withSessionT sTout sessionT =+ bracket nextSessionId removeSession $ \sid -> do+ aEnv <- newSessionEnv sTout sid+ runSessionT_ aEnv sessionT++newSessionEnv :: (MonadIO m, Eq k, Hashable k) => Maybe Int64 -> k -> NodeT u nid k rpkt tp m (SessionEnv u nid k rpkt)+newSessionEnv sTout sid = do+ NodeEnv{..} <- ask+ sEnv <- S.newSessionEnv uEnv nodeId sid (fromMaybe sessTimeout sTout) []+ case nodeMode of+ Single -> atomically $ do+ sess <- readTVar nodeSession+ case sess of+ Nothing -> writeTVar nodeSession $ Just sEnv+ Just _ -> do+ state <- readTVar nodeStatus+ when state retrySTM+ Multi -> HM.insert sessionList sid sEnv+ return sEnv++nextSessionId :: MonadIO m => NodeT u nid k rpkt tp m k+nextSessionId = liftIO =<< asks sessionGen++removeSession :: (MonadIO m, Eq k, Hashable k) => k -> NodeT u nid k rpkt tp m ()+removeSession mid = do+ NodeEnv{..} <- ask+ case nodeMode of+ Single -> atomically $ writeTVar nodeSession Nothing+ Multi -> HM.delete sessionList mid++busy :: MonadIO m => NodeT u nid k rpkt tp m Bool+busy = do+ NodeEnv{..} <- ask+ case nodeMode of+ Single -> isJust <$> readTVarIO nodeSession+ Multi -> return False++tryMainLoop+ :: (MonadUnliftIO m, Transport tp, RecvPacket rpkt, GetPacketId k rpkt, Eq k, Hashable k)+ => (rpkt -> m Bool) -> SessionT u nid k rpkt tp m () -> NodeT u nid k rpkt tp m ()+tryMainLoop preprocess sessionHandler = do+ r <- tryAny $ mainLoop preprocess sessionHandler+ case r of+ Left _ -> stopNodeT+ Right _ -> pure ()++mainLoop+ :: (MonadUnliftIO m, Transport tp, RecvPacket rpkt, GetPacketId k rpkt, Eq k, Hashable k)+ => (rpkt -> m Bool) -> SessionT u nid k rpkt tp m () -> NodeT u nid k rpkt tp m ()+mainLoop preprocess sessionHandler = do+ NodeEnv{..} <- ask+ rpkt <- fromConn receive+ setTimer =<< getEpochTime+ r <- lift $ preprocess rpkt+ when r $ void . async $ tryDoFeed rpkt sessionHandler++tryDoFeed+ :: (MonadUnliftIO m, Transport tp, GetPacketId k rpkt, Eq k, Hashable k)+ => rpkt -> SessionT u nid k rpkt tp m () -> NodeT u nid k rpkt tp m ()+tryDoFeed rpkt sessionHandler = do+ r <- tryAny $ doFeed rpkt sessionHandler+ case r of+ Left e -> liftIO $ errorM "Metro.Node" $ "DoFeed Error: " ++ show e+ Right _ -> pure ()++doFeed+ :: (MonadUnliftIO m, GetPacketId k rpkt, Eq k, Hashable k)+ => rpkt -> SessionT u nid k rpkt tp m () -> NodeT u nid k rpkt tp m ()+doFeed rpkt sessionHandler = do+ NodeEnv{..} <- ask+ v <- case nodeMode of+ Single -> readTVarIO nodeSession+ Multi -> HM.lookup sessionList $ getPacketId rpkt+ case v of+ Just aEnv ->+ runSessionT_ aEnv $ feed $ Just rpkt+ Nothing -> do+ let sid = getPacketId rpkt+ sEnv <- S.newSessionEnv uEnv nodeId sid sessTimeout [Just rpkt]+ when (sessionMode == MultiAction) $+ case nodeMode of+ Single -> atomically $ writeTVar nodeSession $ Just sEnv+ Multi -> HM.insert sessionList sid sEnv+ bracket (return sid) removeSession $ \_ ->+ runSessionT_ sEnv sessionHandler++startNodeT+ :: (MonadUnliftIO m, Transport tp, RecvPacket rpkt, GetPacketId k rpkt, Eq k, Hashable k)+ => SessionT u nid k rpkt tp m () -> NodeT u nid k rpkt tp m ()+startNodeT = startNodeT_ (const $ return True)++startNodeT_+ :: (MonadUnliftIO m, Transport tp, RecvPacket rpkt, GetPacketId k rpkt, Eq k, Hashable k)+ => (rpkt -> m Bool) -> SessionT u nid k rpkt tp m () -> NodeT u nid k rpkt tp m ()+startNodeT_ preprocess sessionHandler = do+ sess <- runCheckSessionState+ void . runMaybeT . forever $ do+ alive <- lift nodeState+ if alive then lift $ tryMainLoop preprocess sessionHandler+ else mzero++ cancel sess+ doFeedError++nodeState :: MonadIO m => NodeT u nid k rpkt tp m Bool+nodeState = readTVarIO =<< asks nodeStatus++doFeedError :: MonadIO m => NodeT u nid k rpkt tp m ()+doFeedError =+ asks sessionList >>= HM.elems >>= mapM_ go+ where go :: MonadIO m => SessionEnv u nid k rpkt -> NodeT u nid k rpkt tp m ()+ go aEnv = runSessionT_ aEnv $ feed Nothing++stopNodeT :: (MonadIO m, Transport tp) => NodeT u nid k rpkt tp m ()+stopNodeT = do+ st <- asks nodeStatus+ atomically $ writeTVar st False+ fromConn close++env :: Monad m => NodeT u nid k rpkt tp m u+env = asks uEnv++request+ :: (MonadUnliftIO m, Transport tp, SendPacket spkt, SetPacketId k spkt, Eq k, Hashable k)+ => Maybe Int64 -> spkt -> NodeT u nid k rpkt tp m (Maybe rpkt)+request sTout = requestAndRetry sTout Nothing++requestAndRetry+ :: (MonadUnliftIO m, Transport tp, SendPacket spkt, SetPacketId k spkt, Eq k, Hashable k)+ => Maybe Int64 -> Maybe Int -> spkt -> NodeT u nid k rpkt tp m (Maybe rpkt)+requestAndRetry sTout retryTout spkt = do+ alive <- nodeState+ if alive then+ withSessionT sTout $ do+ S.send spkt+ t <- forM retryTout $ \tout ->+ async $ forever $ do+ threadDelay $ tout * 1000 * 1000+ S.send spkt+ ret <- S.receive+ mapM_ cancel t+ return ret+++ else return Nothing++getTimer :: MonadIO m => NodeT u nid k rpkt tp m Int64+getTimer = readTVarIO =<< asks nodeTimer++setTimer :: MonadIO m => Int64 -> NodeT u nid k rpkt tp m ()+setTimer t = do+ v <- asks nodeTimer+ atomically $ writeTVar v t++getNodeId :: Monad m => NodeT n nid k rpkt tp m nid+getNodeId = asks nodeId++runCheckSessionState :: (MonadUnliftIO m, Eq k, Hashable k) => NodeT u nid k rpkt tp m (Async ())+runCheckSessionState = do+ sessList <- asks sessionList+ async . forever $ do+ threadDelay $ 1000 * 1000 * 10 -- 10 seconds+ mapM_ (checkAlive sessList) =<< HM.elems sessList++ where checkAlive+ :: (MonadUnliftIO m, Eq k, Hashable k)+ => IOHashMap k (SessionEnv u nid k rpkt) -> SessionEnv u nid k rpkt -> NodeT u nid k rpkt tp m ()+ checkAlive sessList sessEnv =+ runSessionT_ sessEnv $ do+ to <- isTimeout+ when to $ do+ feed Nothing+ HM.delete sessList (sessionId sessEnv)++getSessionSize :: MonadIO m => NodeEnv u nid k rpkt -> m Int+getSessionSize NodeEnv {..} = HM.size sessionList++getSessionSize1 :: MonadIO m => NodeEnv1 u nid k rpkt tp -> m Int+getSessionSize1 NodeEnv1 {..} = getSessionSize nodeEnv
+ src/Metro/Server.hs view
@@ -0,0 +1,292 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++module Metro.Server+ ( startServer+ , startServer_+ , ServerEnv+ , ServerT+ , Servable (..)+ , getNodeEnvList+ , getServ+ , serverEnv+ , initServerEnv++ -- server env action+ , setServerName+ , setNodeMode+ , setSessionMode+ , setDefaultSessionTimeout+ , setKeepalive++ , setOnNodeLeave++ , runServerT+ , stopServerT+ , handleConn+ ) where++import Control.Monad (forM_, forever, mzero, unless,+ void, when)+import Control.Monad.Reader.Class (MonadReader (ask), asks)+import Control.Monad.Trans.Class (MonadTrans, lift)+import Control.Monad.Trans.Maybe (runMaybeT)+import Control.Monad.Trans.Reader (ReaderT (..), runReaderT)+import Data.Either (isLeft)+import Data.Hashable+import Data.Int (Int64)+import Metro.Class (GetPacketId, RecvPacket,+ Servable (..), Transport,+ TransportConfig)+import Metro.Conn hiding (close)+import Metro.IOHashMap (IOHashMap, newIOHashMap)+import qualified Metro.IOHashMap as HM (delete, elems, insertSTM,+ lookupSTM)+import Metro.Node (NodeEnv1, NodeMode (..),+ SessionMode (..), getNodeId,+ getTimer, initEnv1, runNodeT1,+ startNodeT_, stopNodeT)+import qualified Metro.Node as Node+import Metro.Session (SessionT)+import Metro.Utils (getEpochTime)+import System.Log.Logger (errorM, infoM)+import UnliftIO+import UnliftIO.Concurrent (threadDelay)++data ServerEnv serv u nid k rpkt tp = ServerEnv+ { serveServ :: serv+ , serveState :: TVar Bool+ , nodeEnvList :: IOHashMap nid (NodeEnv1 u nid k rpkt tp)+ , prepare :: SID serv -> ConnEnv tp -> IO (Maybe (nid, u))+ , gen :: IO k+ , keepalive :: Int64+ , defSessTout :: Int64+ , nodeMode :: NodeMode+ , sessionMode :: SessionMode+ , serveName :: String+ , onNodeLeave :: TVar (Maybe (nid -> u -> IO ()))+ , mapTransport :: TransportConfig (STP serv) -> TransportConfig tp+ }+++newtype ServerT serv u nid k rpkt tp m a = ServerT {unServerT :: ReaderT (ServerEnv serv u nid k rpkt tp) m a}+ deriving+ ( Functor+ , Applicative+ , Monad+ , MonadIO+ , MonadReader (ServerEnv serv u nid k rpkt tp)+ )++instance MonadTrans (ServerT serv u nid k rpkt tp) where+ lift = ServerT . lift++instance MonadUnliftIO m => MonadUnliftIO (ServerT serv u nid k rpkt tp m) where+ askUnliftIO = ServerT $+ ReaderT $ \r ->+ withUnliftIO $ \u ->+ return (UnliftIO (unliftIO u . runServerT r))+ withRunInIO inner = ServerT $+ ReaderT $ \r ->+ withRunInIO $ \run ->+ inner (run . runServerT r)++runServerT :: ServerEnv serv u nid k rpkt tp -> ServerT serv u nid k rpkt tp m a -> m a+runServerT sEnv = flip runReaderT sEnv . unServerT++initServerEnv+ :: (MonadIO m, Servable serv)+ => ServerConfig serv -> IO k+ -> (TransportConfig (STP serv) -> TransportConfig tp)+ -> (SID serv -> ConnEnv tp -> IO (Maybe (nid, u)))+ -> m (ServerEnv serv u nid k rpkt tp)+initServerEnv sc gen mapTransport prepare = do+ serveServ <- newServer sc+ serveState <- newTVarIO True+ nodeEnvList <- newIOHashMap+ onNodeLeave <- newTVarIO Nothing+ pure ServerEnv+ { nodeMode = Multi+ , sessionMode = SingleAction+ , serveName = "Metro"+ , keepalive = 0+ , defSessTout = 300+ , ..+ }++setNodeMode+ :: NodeMode -> ServerEnv serv u nid k rpkt tp -> ServerEnv serv u nid k rpkt tp+setNodeMode mode sEnv = sEnv {nodeMode = mode}++setSessionMode+ :: SessionMode -> ServerEnv serv u nid k rpkt tp -> ServerEnv serv u nid k rpkt tp+setSessionMode mode sEnv = sEnv {sessionMode = mode}++setServerName+ :: String -> ServerEnv serv u nid k rpkt tp -> ServerEnv serv u nid k rpkt tp+setServerName n sEnv = sEnv {serveName = n}++setKeepalive+ :: Int64 -> ServerEnv serv u nid k rpkt tp -> ServerEnv serv u nid k rpkt tp+setKeepalive k sEnv = sEnv {keepalive = k}++setDefaultSessionTimeout+ :: Int64 -> ServerEnv serv u nid k rpkt tp -> ServerEnv serv u nid k rpkt tp+setDefaultSessionTimeout t sEnv = sEnv {defSessTout = t}++setOnNodeLeave :: MonadIO m => ServerEnv serv u nid k rpkt tp -> (nid -> u -> IO ()) -> m ()+setOnNodeLeave sEnv =+ atomically . writeTVar (onNodeLeave sEnv) . Just++serveForever+ :: (MonadUnliftIO m, Transport tp, Show nid, Eq nid, Hashable nid, Eq k, Hashable k, GetPacketId k rpkt, RecvPacket rpkt, Servable serv)+ => (rpkt -> m Bool)+ -> SessionT u nid k rpkt tp m ()+ -> ServerT serv u nid k rpkt tp m ()+serveForever preprocess sess = do+ name <- asks serveName+ liftIO $ infoM "Metro.Server" $ name ++ "Server started"+ state <- asks serveState+ void . runMaybeT . forever $ do+ e <- lift $ tryServeOnce preprocess sess+ when (isLeft e) mzero+ alive <- readTVarIO state+ unless alive mzero+ liftIO $ infoM "Metro.Server" $ name ++ "Server closed"++tryServeOnce+ :: (MonadUnliftIO m, Transport tp, Show nid, Eq nid, Hashable nid, Eq k, Hashable k, GetPacketId k rpkt, RecvPacket rpkt, Servable serv)+ => (rpkt -> m Bool)+ -> SessionT u nid k rpkt tp m ()+ -> ServerT serv u nid k rpkt tp m (Either SomeException ())+tryServeOnce preprocess sess = tryAny (serveOnce preprocess sess)++serveOnce+ :: ( MonadUnliftIO m+ , Transport tp+ , Show nid, Eq nid, Hashable nid+ , Eq k, Hashable k, GetPacketId k rpkt, RecvPacket rpkt+ , Servable serv)+ => (rpkt -> m Bool)+ -> SessionT u nid k rpkt tp m ()+ -> ServerT serv u nid k rpkt tp m ()+serveOnce preprocess sess = do+ ServerEnv {..} <- ask+ servOnce serveServ $ doServeOnce preprocess sess++doServeOnce+ :: ( MonadUnliftIO m+ , Transport tp+ , Show nid, Eq nid, Hashable nid+ , Eq k, Hashable k, GetPacketId k rpkt, RecvPacket rpkt+ , Servable serv)+ => (rpkt -> m Bool)+ -> SessionT u nid k rpkt tp m ()+ -> Maybe (SID serv, TransportConfig (STP serv))+ -> ServerT serv u nid k rpkt tp m ()+doServeOnce _ _ Nothing = return ()+doServeOnce preprocess sess (Just (servID, stp)) = do+ ServerEnv {..} <- ask+ connEnv <- initConnEnv $ mapTransport stp+ mnid <- liftIO $ prepare servID connEnv+ forM_ mnid $ \(nid, uEnv) -> do+ (_, io) <- handleConn "Client" servID connEnv nid uEnv preprocess sess+ r <- waitCatch io+ case r of+ Left e -> liftIO $ errorM "Metro.Server" $ "Handle connection error " ++ show e+ Right _ -> return ()++handleConn+ :: (MonadUnliftIO m, Transport tp, Show nid, Eq nid, Hashable nid, Eq k, Hashable k, GetPacketId k rpkt, RecvPacket rpkt, Servable serv)+ => String+ -> SID serv+ -> ConnEnv tp+ -> nid+ -> u+ -> (rpkt -> m Bool)+ -> SessionT u nid k rpkt tp m ()+ -> ServerT serv u nid k rpkt tp m (NodeEnv1 u nid k rpkt tp, Async ())+handleConn n servID connEnv nid uEnv preprocess sess = do+ ServerEnv {..} <- ask++ liftIO $ infoM "Metro.Server" (serveName ++ n ++ ": " ++ show nid ++ " connected")+ env0 <- initEnv1+ (Node.setNodeMode nodeMode+ . Node.setSessionMode sessionMode+ . Node.setDefaultSessionTimeout defSessTout) connEnv uEnv nid gen++ env1 <- atomically $ do+ v <- HM.lookupSTM nodeEnvList nid+ HM.insertSTM nodeEnvList nid env0+ pure v++ mapM_ (`runNodeT1` stopNodeT) env1++ io <- async $ do+ onConnEnter serveServ servID+ lift . runNodeT1 env0 $ startNodeT_ preprocess sess+ onConnLeave serveServ servID+ nodeLeave <- readTVarIO onNodeLeave+ case nodeLeave of+ Nothing -> pure ()+ Just f -> liftIO $ f nid uEnv+ liftIO $ infoM "Metro.Server" (serveName ++ n ++ ": " ++ show nid ++ " disconnected")++ return (env0, io)++startServer+ :: (MonadUnliftIO m, Transport tp, Show nid, Eq nid, Hashable nid, Eq k, Hashable k, GetPacketId k rpkt, RecvPacket rpkt, Servable serv)+ => ServerEnv serv u nid k rpkt tp+ -> SessionT u nid k rpkt tp m ()+ -> m ()+startServer sEnv = startServer_ sEnv (const $ return True)++startServer_+ :: (MonadUnliftIO m, Transport tp, Show nid, Eq nid, Hashable nid, Eq k, Hashable k, GetPacketId k rpkt, RecvPacket rpkt, Servable serv)+ => ServerEnv serv u nid k rpkt tp+ -> (rpkt -> m Bool)+ -> SessionT u nid k rpkt tp m ()+ -> m ()+startServer_ sEnv preprocess sess = do+ when (keepalive sEnv > 0) $ runCheckNodeState (keepalive sEnv) (nodeEnvList sEnv)+ runServerT sEnv $ serveForever preprocess sess+ liftIO $ servClose $ serveServ sEnv++stopServerT :: (MonadIO m, Servable serv) => ServerT serv u nid k rpkt tp m ()+stopServerT = do+ ServerEnv {..} <- ask+ atomically $ writeTVar serveState False+ liftIO $ servClose serveServ++runCheckNodeState+ :: (MonadUnliftIO m, Eq nid, Hashable nid, Transport tp)+ => Int64 -> IOHashMap nid (NodeEnv1 u nid k rpkt tp) -> m ()+runCheckNodeState alive envList = void . async . forever $ do+ threadDelay $ fromIntegral alive * 1000 * 1000+ mapM_ (checkAlive envList) =<< HM.elems envList++ where checkAlive+ :: (MonadUnliftIO m, Eq nid, Hashable nid, Transport tp)+ => IOHashMap nid (NodeEnv1 u nid k rpkt tp)+ -> NodeEnv1 u nid k rpkt tp -> m ()+ checkAlive ref env1 = runNodeT1 env1 $ do+ expiredAt <- (alive +) <$> getTimer+ now <- getEpochTime+ when (now > expiredAt) $ do+ nid <- getNodeId+ stopNodeT+ HM.delete ref nid++serverEnv :: Monad m => ServerT serv u nid k rpkt tp m (ServerEnv serv u nid k rpkt tp)+serverEnv = ask++getNodeEnvList :: ServerEnv serv u nid k rpkt tp -> IOHashMap nid (NodeEnv1 u nid k rpkt tp)+getNodeEnvList = nodeEnvList++getServ :: ServerEnv serv u nid k rpkt tp -> serv+getServ = serveServ
+ src/Metro/Session.hs view
@@ -0,0 +1,168 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++module Metro.Session+ ( SessionEnv (..)+ , SessionEnv1 (..)+ , newSessionEnv+ , SessionT+ , runSessionT+ , runSessionT1+ , send+ , sessionState+ , feed+ , receive+ , readerSize+ , getSessionId+ , getNodeId+ , getSessionEnv1+ , env+ , ident++ , isTimeout++ , makeResponse+ , makeResponse_+ ) where++import Control.Monad.Reader.Class (MonadReader, ask, asks)+import Control.Monad.Trans.Class (MonadTrans (..))+import Control.Monad.Trans.Reader (ReaderT (..), runReaderT)+import Data.Int (Int64)+import Metro.Class (SendPacket, SetPacketId, Transport,+ setPacketId)+import Metro.Conn (ConnEnv, ConnT, FromConn (..),+ runConnT, statusTVar)+import qualified Metro.Conn as Conn (send)+import Metro.Utils (getEpochTime)+import UnliftIO++data SessionEnv u nid k rpkt = SessionEnv+ { sessionData :: TVar [Maybe rpkt]+ , sessionNid :: nid+ , sessionId :: k+ , sessionUEnv :: u+ , sessionTimer :: TVar Int64+ , sessionTimeout :: Int64+ }++data SessionEnv1 u nid k rpkt tp = SessionEnv1+ { sessionEnv :: SessionEnv u nid k rpkt+ , connEnv :: ConnEnv tp+ }++newSessionEnv :: MonadIO m => u -> nid -> k -> Int64 -> [Maybe rpkt] -> m (SessionEnv u nid k rpkt)+newSessionEnv sessionUEnv sessionNid sessionId sessionTimeout rpkts = do+ sessionData <- newTVarIO rpkts+ sessionTimer <- newTVarIO =<< getEpochTime+ pure SessionEnv {..}++newtype SessionT u nid k rpkt tp m a = SessionT { unSessionT :: ReaderT (SessionEnv u nid k rpkt) (ConnT tp m) a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadReader (SessionEnv u nid k rpkt))++instance MonadTrans (SessionT u nid k rpkt tp) where+ lift = SessionT . lift . lift++instance MonadUnliftIO m => MonadUnliftIO (SessionT u nid k rpkt tp m) where+ askUnliftIO = SessionT $+ ReaderT $ \r ->+ withUnliftIO $ \u ->+ return (UnliftIO (unliftIO u . runSessionT r))+ withRunInIO inner = SessionT $+ ReaderT $ \r ->+ withRunInIO $ \run ->+ inner (run . runSessionT r)++instance FromConn (SessionT u nid k rpkt) where+ fromConn = SessionT . lift++runSessionT :: SessionEnv u nid k rpkt -> SessionT u nid k rpkt tp m a -> ConnT tp m a+runSessionT aEnv = flip runReaderT aEnv . unSessionT++runSessionT1 :: SessionEnv1 u nid k rpkt tp -> SessionT u nid k rpkt tp m a -> m a+runSessionT1 SessionEnv1 {..} = runConnT connEnv . runSessionT sessionEnv++sessionState :: MonadIO m => SessionT u nid k rpkt tp m Bool+sessionState = readTVarIO =<< fromConn statusTVar++send+ :: (MonadUnliftIO m, Transport tp, SendPacket spkt, SetPacketId k spkt)+ => spkt -> SessionT u nid k rpkt tp m ()+send rpkt = do+ mid <- getSessionId+ fromConn $ Conn.send $ setPacketId mid rpkt++feed :: (MonadIO m) => Maybe rpkt -> SessionT u nid k rpkt tp m ()+feed rpkt = do+ reader <- asks sessionData+ setTimer =<< getEpochTime+ atomically . modifyTVar' reader $ \v -> v ++ [rpkt]++receive :: (MonadIO m, Transport tp) => SessionT u nid k rpkt tp m (Maybe rpkt)+receive = do+ reader <- asks sessionData+ st <- fromConn statusTVar+ atomically $ do+ v <- readTVar reader+ if null v then do+ s <- readTVar st+ if s then retrySTM+ else pure Nothing+ else do+ writeTVar reader $! tail v+ pure $ head v++readerSize :: MonadIO m => SessionT u nid k rpkt tp m Int+readerSize = fmap length $ readTVarIO =<< asks sessionData++getSessionId :: Monad m => SessionT u nid k rpkt tp m k+getSessionId = asks sessionId++getNodeId :: Monad m => SessionT u nid k rpkt tp m nid+getNodeId = asks sessionNid++env :: Monad m => SessionT u nid k rpkt tp m u+env = asks sessionUEnv++-- makeResponse if Nothing ignore+makeResponse+ :: (MonadUnliftIO m, Transport tp, SendPacket spkt, SetPacketId k spkt)+ => (rpkt -> m (Maybe spkt)) -> SessionT u nid k rpkt tp m ()+makeResponse f = mapM_ doSend =<< receive++ where doSend spkt = mapM_ send =<< (lift . f) spkt++makeResponse_+ :: (MonadUnliftIO m, Transport tp, SendPacket spkt, SetPacketId k spkt)+ => (rpkt -> Maybe spkt) -> SessionT u nid k rpkt tp m ()+makeResponse_ f = makeResponse (pure . f)++getTimer :: MonadIO m => SessionT u nid k rpkt tp m Int64+getTimer = readTVarIO =<< asks sessionTimer++setTimer :: MonadIO m => Int64 -> SessionT u nid k rpkt tp m ()+setTimer t = do+ v <- asks sessionTimer+ atomically $ writeTVar v t++isTimeout :: MonadIO m => SessionT u nid k rpkt tp m Bool+isTimeout = do+ t <- getTimer+ tout <- asks sessionTimeout+ now <- getEpochTime+ if tout > 0 then return $ (t + tout) < now+ else return False++getSessionEnv1 :: (Monad m, Transport tp) => SessionT u nid k rpkt tp m (SessionEnv1 u nid k rpkt tp)+getSessionEnv1 = do+ connEnv <- fromConn ask+ sessionEnv <- ask+ pure SessionEnv1 {..}++ident :: SessionEnv1 u nid k rpkt tp -> (nid, k)+ident SessionEnv1 {..} = (sessionNid sessionEnv, sessionId sessionEnv)
+ src/Metro/TP/BS.hs view
@@ -0,0 +1,75 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeFamilies #-}++module Metro.TP.BS+ ( BSTransport+ , BSHandle+ , newBSHandle+ , newBSHandle_+ , feed+ , closeBSHandle+ , bsTransportConfig++ , makePipe+ ) where++import Control.Monad (when)+import Data.ByteString (ByteString, empty)+import qualified Data.ByteString as B (drop, length, take)+import Metro.Class (Transport (..))+import UnliftIO++data BSHandle = BSHandle Int (TVar Bool) (TVar ByteString)++newBSHandle :: MonadIO m => ByteString -> m BSHandle+newBSHandle = newBSHandle_ 41943040 -- 40M++newBSHandle_ :: MonadIO m => Int -> ByteString -> m BSHandle+newBSHandle_ size bs = do+ state <- newTVarIO True+ BSHandle size state <$> newTVarIO bs++feed :: MonadIO m => BSHandle -> ByteString -> m ()+feed (BSHandle size state h) bs = atomically $ do+ st <- readTVar state+ when st $ do+ bs0 <- readTVar h+ when (B.length bs0 > size) retrySTM+ writeTVar h $ bs0 <> bs++closeBSHandle :: MonadIO m => BSHandle -> m ()+closeBSHandle (BSHandle _ state _) = atomically $ writeTVar state False++data BSTransport = BS+ { bsHandle :: TVar ByteString+ , bsWriter :: ByteString -> IO ()+ , bsState :: TVar Bool+ }++instance Transport BSTransport where+ data TransportConfig BSTransport = BSConfig BSHandle (ByteString -> IO ())+ newTransport (BSConfig (BSHandle _ bsState bsHandle) bsWriter) =+ return BS {..}+ recvData BS {..} nbytes = atomically $ do+ bs <- readTVar bsHandle+ if bs == empty then do+ status <- readTVar bsState+ if status then retrySTM+ else return bs+ else do+ writeTVar bsHandle $ B.drop nbytes bs+ return $ B.take nbytes bs+ sendData BS {..} bs = do+ status <- readTVarIO bsState+ when status $ bsWriter bs+ closeTransport BS {..} = atomically $ writeTVar bsState False++bsTransportConfig :: BSHandle -> (ByteString -> IO ()) -> TransportConfig BSTransport+bsTransportConfig = BSConfig++makePipe :: MonadIO m => m (TransportConfig BSTransport, TransportConfig BSTransport)+makePipe = do+ rHandle <- newBSHandle empty+ wHandle <- newBSHandle empty++ return (bsTransportConfig rHandle (feed wHandle), bsTransportConfig wHandle (feed rHandle))
+ src/Metro/TP/Debug.hs view
@@ -0,0 +1,46 @@+{-# LANGUAGE TypeFamilies #-}++module Metro.TP.Debug+ ( Debug+ , DebugMode (..)+ , debugConfig+ ) where++import Data.ByteString (ByteString)+import Data.ByteString.Char8 (unpack)+import Metro.Class (Transport (..))+import System.Log.Logger (debugM)++hex :: ByteString -> String+hex = Prelude.concatMap w . unpack+ where w ch = let s = "0123456789ABCDEF"+ x = fromEnum ch+ in [s !! div x 16,s !! mod x 16]++data Debug tp = Debug String (ByteString -> String) tp++data DebugMode = Raw+ | Hex++instance Transport tp => Transport (Debug tp) where+ data TransportConfig (Debug tp) = DebugConfig String DebugMode (TransportConfig tp)+ newTransport (DebugConfig h mode config) = do+ tp <- newTransport config+ return $ Debug h f tp+ where f = case mode of+ Raw -> show+ Hex -> hex++ recvData (Debug h f tp) nbytes = do+ bs <- recvData tp nbytes+ debugM "Metro.Transport.Debug" $ h ++ " recv " ++ f bs+ return bs+ sendData (Debug h f tp) bs = do+ debugM "Metro.Transport.Debug" $ h ++ " send " ++ f bs+ sendData tp bs+ closeTransport (Debug h _ tp) = do+ debugM "Metro.Transport.Debug" $ h ++ " transport close"+ closeTransport tp++debugConfig :: String -> DebugMode -> TransportConfig tp -> TransportConfig (Debug tp)+debugConfig = DebugConfig
+ src/Metro/Utils.hs view
@@ -0,0 +1,58 @@+module Metro.Utils+ ( getEpochTime+ , setupLog+ , recvEnough+ ) where+++import Control.Monad (when)+import Data.ByteString (ByteString)+import qualified Data.ByteString as B (concat, drop, empty, length,+ null, take)+import Data.Int (Int64)+import Data.UnixTime (getUnixTime, toEpochTime)+import Foreign.C.Types (CTime (..))+import Metro.Class (Transport (..), TransportError (..))+import System.IO (stderr)+import System.Log.Formatter (simpleLogFormatter)+import System.Log.Handler (setFormatter)+import System.Log.Handler.Simple (streamHandler)+import System.Log.Logger+import UnliftIO (MonadIO (..), TVar, atomically,+ readTVar, throwIO, writeTVar)++-- utils+getEpochTime :: MonadIO m => m Int64+getEpochTime = liftIO $ un . toEpochTime <$> getUnixTime+ where un :: CTime -> Int64+ un (CTime t) = t++setupLog :: Priority -> IO ()+setupLog logLevel = do+ removeAllHandlers+ handle <- streamHandler stderr logLevel >>= \lh -> return $+ setFormatter lh (simpleLogFormatter "[$time : $loggername : $prio] $msg")+ updateGlobalLogger rootLoggerName (addHandler handle . setLevel logLevel)++recvEnough :: (MonadIO m, Transport tp) => TVar ByteString -> tp -> Int -> m ByteString+recvEnough buffer tp nbytes = do+ buf <- atomically $ do+ bf <- readTVar buffer+ writeTVar buffer $! B.drop nbytes bf+ return $! B.take nbytes bf+ if B.length buf == nbytes then return buf+ else do+ otherBuf <- liftIO $ readBuf (nbytes - B.length buf)+ let out = B.concat [ buf, otherBuf ]+ atomically . writeTVar buffer $! B.drop nbytes out+ return $! B.take nbytes out++ where readBuf :: Int -> IO ByteString+ readBuf 0 = return B.empty+ readBuf nb = do+ buf <- recvData tp $ max 4096 nb -- 4k+ when (B.null buf) $ throwIO TransportClosed+ if B.length buf >= nb then return buf+ else do+ otherBuf <- readBuf (nb - B.length buf)+ return $! B.concat [ buf, otherBuf ]