packages feed

periodic-client (empty) → 1.1.7.1

raw patch · 12 files changed

+766/−0 lines, 12 filesdep +basedep +binarydep +byteable

Dependencies added: base, binary, byteable, bytestring, hslogger, metro, metro-socket, mtl, periodic-common, resource-pool, transformers, unliftio

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Li Meng Jun (c) 2017++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Li Meng Jun nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,3 @@+# periodic-client++Periodic task system haskell client.
+ periodic-client.cabal view
@@ -0,0 +1,47 @@+name:                periodic-client+version:             1.1.7.1+synopsis:            Periodic task system haskell client.+description:         Periodic task system haskell client library.+homepage:            https://github.com/Lupino/haskell-periodic/tree/master/periodic-client#readme+license:             BSD3+license-file:        LICENSE+author:              Li Meng Jun+maintainer:          lmjubuntu@gmail.com+copyright:           MIT+category:            value+build-type:          Simple+extra-source-files:  README.md+cabal-version:       >=1.10++library+  hs-source-dirs:      src+  exposed-modules:     Periodic.Trans.BaseClient+                     , Periodic.Trans.Client+                     , Periodic.Trans.ClientPool+                     , Periodic.Trans.Job+                     , Periodic.Trans.Worker+                     , Periodic.Client+                     , Periodic.ClientPool+                     , Periodic.Job+                     , Periodic.Worker++  build-depends:       base >= 4.7 && < 5+                     , periodic-common+                     , bytestring+                     , hslogger+                     , byteable+                     , transformers++                     , resource-pool++                     , unliftio++                     , metro+                     , metro-socket+                     , binary+                     , mtl+  default-language:    Haskell2010++source-repository head+  type:     git+  location: https://github.com/Lupino/haskell-periodic
+ src/Periodic/Client.hs view
@@ -0,0 +1,36 @@+module Periodic.Client+  ( ClientM+  , ClientEnv+  , open+  , close+  , runClientM++  -- re-exports+  , ping+  , submitJob_+  , submitJob+  , runJob_+  , runJob+  , removeJob+  , dropFunc+  , status+  , configGet+  , configSet+  , load+  , dump+  , shutdown+  ) where++import           Metro.TP.Socket       (Socket, socket)+import           Periodic.Trans.Client hiding (ClientEnv, open)+import qualified Periodic.Trans.Client as C (ClientEnv, open)++type ClientM = ClientT Socket IO++type ClientEnv = C.ClientEnv Socket++open :: String -> IO ClientEnv+open = C.open . socket++runClientM :: ClientEnv -> ClientM a -> IO a+runClientM = runClientT
+ src/Periodic/ClientPool.hs view
@@ -0,0 +1,19 @@+module Periodic.ClientPool+  ( module Periodic.Client+  , ClientPoolEnv+  , runClientPoolM+  , openPool+  ) where++import           Metro.TP.Socket           (Socket, socket)+import           Periodic.Client           hiding (close)+import           Periodic.Trans.ClientPool (runClientPoolT)+import qualified Periodic.Trans.ClientPool as P (ClientPoolEnv, openPool)++type ClientPoolEnv = P.ClientPoolEnv Socket++runClientPoolM :: ClientPoolEnv -> ClientM a -> IO a+runClientPoolM = runClientPoolT++openPool :: String -> Int -> IO ClientPoolEnv+openPool h = P.openPool (socket h)
+ src/Periodic/Job.hs view
@@ -0,0 +1,10 @@+module Periodic.Job+  ( JobM+  , module Periodic.Trans.Job+  ) where++import           Metro.TP.Socket    (Socket)+import           Periodic.Trans.Job hiding (JobT)+import qualified Periodic.Trans.Job as J (JobT)++type JobM = J.JobT Socket IO
+ src/Periodic/Trans/BaseClient.hs view
@@ -0,0 +1,75 @@+{-# LANGUAGE OverloadedStrings #-}++module Periodic.Trans.BaseClient+  ( BaseClientT+  , BaseClientEnv+  , getClientEnv+  , close+  , runBaseClientT++  , ping+  , submitJob_+  , submitJob+  , runJob_+  , runJob+  , checkHealth+  ) where++import           Control.Monad                (unless)+import           Data.ByteString              (ByteString)+import           Data.Int                     (Int64)+import           Data.Maybe                   (fromMaybe)+import           Metro.Class                  (Transport)+import           Metro.Node                   (getEnv1, request, stopNodeT)+import           Metro.Utils                  (getEpochTime)+import           Periodic.Node+import           Periodic.Types               (getResult, packetREQ)+import           Periodic.Types.ClientCommand+import           Periodic.Types.Job+import           Periodic.Types.ServerCommand+import           UnliftIO++type BaseClientEnv u = NodeEnv u ServerCommand+type BaseClientT u = NodeT u ServerCommand++runBaseClientT :: Monad m => BaseClientEnv u tp -> BaseClientT u tp m a -> m a+runBaseClientT = runNodeT++close :: (MonadUnliftIO m, Transport tp) => BaseClientT u tp m ()+close = stopNodeT++ping :: (MonadUnliftIO m, Transport tp) => BaseClientT u tp m Bool+ping = getResult False isPong <$> request Nothing (packetREQ Ping)++submitJob_ :: (MonadUnliftIO m, Transport tp) => Job -> BaseClientT u tp m Bool+submitJob_ j = getResult False isSuccess <$> request Nothing (packetREQ (SubmitJob j))++submitJob+  :: (MonadUnliftIO m, Transport tp)+  => FuncName -> JobName -> Maybe Workload -> Maybe Int64 -> BaseClientT u tp m Bool+submitJob fn jn w later = do+  schedAt <- (+fromMaybe 0 later) <$> getEpochTime+  submitJob_ $ setSchedAt schedAt $ setWorkload (fromMaybe "" w) $ initJob fn jn++runJob_ :: (MonadUnliftIO m, Transport tp) => Job -> BaseClientT u tp m (Maybe ByteString)+runJob_ j =  getResult Nothing getData <$> request Nothing (packetREQ . RunJob $ setSchedAt 0 j)+  where getData :: ServerCommand -> Maybe ByteString+        getData (Data bs) = Just bs+        getData _         = Nothing++runJob+  :: (MonadUnliftIO m, Transport tp)+  => FuncName -> JobName -> Maybe Workload -> BaseClientT u tp m (Maybe ByteString)+runJob fn jn w = do+  schedAt <- getEpochTime+  runJob_ $ setSchedAt schedAt $ setWorkload (fromMaybe "" w) $ initJob fn jn++checkHealth :: (MonadUnliftIO m, Transport tp) => BaseClientT u tp m ()+checkHealth = do+  ret <- timeout 10000000 ping+  case ret of+    Nothing -> close+    Just r  -> unless r close++getClientEnv :: (Monad m, Transport tp) => BaseClientT u tp m (BaseClientEnv u tp)+getClientEnv = getEnv1
+ src/Periodic/Trans/Client.hs view
@@ -0,0 +1,125 @@+{-# LANGUAGE OverloadedStrings #-}++module Periodic.Trans.Client+  ( ClientT+  , ClientEnv+  , open+  , close+  , runClientT++  , ping+  , submitJob_+  , submitJob+  , runJob_+  , runJob+  , removeJob+  , dropFunc+  , status+  , configGet+  , configSet+  , load+  , dump+  , shutdown+  ) where++import           Control.Monad                (forever, void)+import           Data.Binary                  (decode)+import           Data.ByteString              (ByteString)+import           Data.ByteString.Lazy         (fromStrict)+import           Metro.Class                  (Transport, TransportConfig)+import           Metro.Conn                   (initConnEnv, runConnT)+import qualified Metro.Conn                   as Conn+import           Metro.Node                   (NodeMode (..), SessionMode (..),+                                               initEnv1, request,+                                               setDefaultSessionTimeout,+                                               setNodeMode, setSessionMode,+                                               startNodeT, withSessionT)+import           Metro.Session                (send)+import           Periodic.Node+import           Periodic.Trans.BaseClient+import           Periodic.Types               (ClientType (TypeClient),+                                               getClientType, getResult,+                                               packetREQ, regPacketREQ)+import           Periodic.Types.ClientCommand+import           Periodic.Types.Internal      (ConfigKey (..))+import           Periodic.Types.Job+import           Periodic.Types.ServerCommand+import           UnliftIO+import           UnliftIO.Concurrent          (threadDelay)++type ClientEnv = BaseClientEnv ()+type ClientT = BaseClientT ()++runClientT :: Monad m => ClientEnv tp -> ClientT tp m a -> m a+runClientT = runNodeT++open+  :: (MonadUnliftIO m, Transport tp)+  => TransportConfig tp -> m (ClientEnv tp)+open config = do+  connEnv <- initConnEnv config+  r <- runConnT connEnv $ do+    Conn.send $ regPacketREQ TypeClient+    Conn.receive++  let nid = case getClientType r of+              Data v -> v+              _      -> ""++  clientEnv <- initEnv1 mapEnv connEnv () (Nid nid) sessionGen++  runClientT clientEnv $ do+    void . async $ forever $ do+      threadDelay $ 100 * 1000 * 1000+      checkHealth++    void $ async $ startNodeT defaultSessionHandler+  return clientEnv++  where mapEnv =+          setNodeMode Multi+          . setSessionMode SingleAction+          . setDefaultSessionTimeout 100++dropFunc+  :: (MonadUnliftIO m, Transport tp)+  => FuncName -> BaseClientT u tp m Bool+dropFunc func = getResult False isSuccess <$> request Nothing (packetREQ (DropFunc func))++removeJob+  :: (MonadUnliftIO m, Transport tp)+  => FuncName -> JobName -> BaseClientT u tp m Bool+removeJob f n = getResult False isSuccess <$> request Nothing (packetREQ (RemoveJob f n))++status+  :: (MonadUnliftIO m, Transport tp)+  => BaseClientT u tp m ByteString+status = getResult "" getRaw <$> request Nothing (packetREQ Status)+  where getRaw :: ServerCommand -> ByteString+        getRaw (Data bs) = bs+        getRaw _         = ""++configGet+  :: (MonadUnliftIO m, Transport tp)+  => String -> BaseClientT u tp m Int+configGet k = getResult 0 getV <$> request Nothing (packetREQ (ConfigGet (ConfigKey k)))+  where getV :: ServerCommand -> Int+        getV (Config v) = v+        getV _          = 0++configSet+  :: (MonadUnliftIO m, Transport tp)+  => String -> Int -> BaseClientT u tp m Bool+configSet k v = getResult False isSuccess <$> request Nothing (packetREQ (ConfigSet (ConfigKey k) v))++load :: (MonadUnliftIO m, Transport tp) => [Job] -> BaseClientT u tp m Bool+load jobs = getResult False isSuccess <$> request Nothing (packetREQ (Load jobs))++dump :: (MonadUnliftIO m, Transport tp) => BaseClientT u tp m [Job]+dump = getResult [] getV <$> request Nothing (packetREQ Dump)+  where getV :: ServerCommand -> [Job]+        getV (Data bs) = decode $ fromStrict bs+        getV _         = []++shutdown :: (MonadUnliftIO m, Transport tp) => BaseClientT u tp m ()+shutdown = withSessionT Nothing $ send $ packetREQ Shutdown
+ src/Periodic/Trans/ClientPool.hs view
@@ -0,0 +1,19 @@+module Periodic.Trans.ClientPool+  ( module Periodic.Trans.Client+  , ClientPoolEnv+  , runClientPoolT+  , openPool+  ) where++import           Data.Pool             (Pool, createPool, withResource)+import           Metro.Class           (Transport, TransportConfig)+import           Periodic.Trans.Client hiding (close)+import qualified Periodic.Trans.Client as C (close)++type ClientPoolEnv tp = Pool (ClientEnv tp)++runClientPoolT :: ClientPoolEnv tp -> ClientT tp IO a -> IO a+runClientPoolT pool m = withResource pool $ flip runClientT m++openPool :: Transport tp => TransportConfig tp -> Int -> IO (ClientPoolEnv tp)+openPool config = createPool (open config) (`runClientT` C.close) 1 5000
+ src/Periodic/Trans/Job.hs view
@@ -0,0 +1,137 @@+module Periodic.Trans.Job+  ( JobT+  , JobEnv+  , name+  , func+  , workload+  , name_+  , func_+  , workload_+  , count+  , timeout++  , workDone+  , workDone_+  , workFail+  , schedLater+  , schedLater'++  , acquireLock+  , releaseLock++  , withLock+  , withLock_+  , module Periodic.Trans.BaseClient+  ) where++import           Control.Monad                (when)+import           Data.ByteString              (ByteString, empty)+import           Data.Int                     (Int64)+import           Data.Maybe                   (fromJust)+import           Metro.Class                  (Transport)+import           Metro.Node                   (env, request, withSessionT)+import           Metro.Session                (send)+import           Periodic.Node+import           Periodic.Trans.BaseClient    (BaseClientT, runJob, runJob_,+                                               submitJob, submitJob_)+import           Periodic.Types               (FromBS (..), LockName, getResult,+                                               packetREQ)+import           Periodic.Types.Job+import           Periodic.Types.ServerCommand (ServerCommand (Acquired))+import           Periodic.Types.WorkerCommand+import           UnliftIO                     hiding (timeout)++type JobT = BaseClientT (Maybe Job)+type JobEnv = SessionEnv (Maybe Job) ServerCommand++job :: Monad m => JobT tp m Job+job = fromJust <$> env++name :: (FromBS a, Show a, Monad m) => JobT tp m a+name = fromBS . unJN <$> name_++name_ :: Monad m => JobT tp m JobName+name_ = getName <$> job++func :: (FromBS a, Show a, Monad m) => JobT tp m a+func = fromBS . unFN <$> func_++func_ :: Monad m => JobT tp m FuncName+func_ = getFuncName <$> job++workload :: (FromBS a, Show a, Monad m) => JobT tp m a+workload = fromBS . unWL <$> workload_++workload_ :: Monad m => JobT tp m Workload+workload_ = getWorkload <$> job++count :: Monad m => JobT tp m Int+count = getCount <$> job++timeout :: Monad m => JobT tp m Int+timeout = getTimeout <$> job++workDone+  :: (MonadUnliftIO m, Transport tp)+  => JobT tp m ()+workDone = workDone_ empty++workDone_+  :: (MonadUnliftIO m, Transport tp)+  => ByteString -> JobT tp m ()+workDone_ w = do+  h <- getHandle <$> job+  withSessionT Nothing $ send (packetREQ $ WorkDone h w)++workFail+  :: (MonadUnliftIO m, Transport tp)+  =>  JobT tp m ()+workFail = do+  h <- getHandle <$> job+  withSessionT Nothing $ send (packetREQ $ WorkFail h)++schedLater+  :: (MonadUnliftIO m, Transport tp)+  =>  Int64 -> JobT tp m ()+schedLater later = do+  h <- getHandle <$> job+  withSessionT Nothing $ send (packetREQ $ SchedLater h later 0)++schedLater'+  :: (MonadUnliftIO m, Transport tp)+  =>  Int64 -> Int -> JobT tp m ()+schedLater' later step = do+  h <- getHandle <$> job+  withSessionT Nothing $ send (packetREQ $ SchedLater h later step)++acquireLock+  :: (MonadUnliftIO m, Transport tp)+  => LockName -> Int -> JobT tp m Bool+acquireLock n maxCount = do+  h <- getHandle <$> job+  getResult False getAcq <$> request Nothing (packetREQ (Acquire n maxCount h))++  where getAcq :: ServerCommand -> Bool+        getAcq (Acquired v) = v+        getAcq _            = False++releaseLock+  :: (MonadUnliftIO m, Transport tp)+  => LockName -> JobT tp m ()+releaseLock n = do+  h <- getHandle <$> job+  withSessionT Nothing $ send (packetREQ $ Release n h)++withLock_+  :: (MonadUnliftIO m, Transport tp)+  => LockName -> Int -> JobT tp m () -> JobT tp m ()+withLock_ n maxCount j = do+  acquired <- acquireLock n maxCount+  when acquired j++withLock+  :: (MonadUnliftIO m, Transport tp)+  => LockName -> Int -> JobT tp m () -> JobT tp m ()+withLock n maxCount j = do+  withLock_ n maxCount j+  releaseLock n
+ src/Periodic/Trans/Worker.hs view
@@ -0,0 +1,247 @@+{-# LANGUAGE FlexibleContexts           #-}+{-# LANGUAGE FlexibleInstances          #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses      #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE RankNTypes                 #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE TypeFamilies               #-}+{-# LANGUAGE UndecidableInstances       #-}++module Periodic.Trans.Worker+  ( WorkerT+  , startWorkerT+  , ping+  , addFunc+  , broadcast+  , removeFunc+  , work+  , close+  , runJobT+  , getClientEnv+  ) where++import           Control.Monad                (forever, replicateM, void, when)+import           Control.Monad.Reader.Class   (MonadReader, asks)+import           Control.Monad.Trans.Class    (MonadTrans, lift)+import           Control.Monad.Trans.Reader   (ReaderT (..), runReaderT)+import           Metro.Class                  (Transport, TransportConfig,+                                               getPacketId)+import           Metro.Conn                   (initConnEnv, runConnT)+import qualified Metro.Conn                   as Conn+import           Metro.IOHashMap              (IOHashMap, newIOHashMap)+import qualified Metro.IOHashMap              as HM (delete, insert, lookup)+import           Metro.Node                   (NodeMode (..), SessionMode (..),+                                               initEnv1, newSessionEnv,+                                               nextSessionId, runSessionT_,+                                               setDefaultSessionTimeout,+                                               setNodeMode, setSessionMode,+                                               startNodeT_, withEnv,+                                               withSessionT)+import           Metro.Session                (send)+import           Periodic.IOList              (IOList, newIOList)+import qualified Periodic.IOList              as IL (append, clearSTM,+                                                     toListSTM)+import           Periodic.Node+import qualified Periodic.Trans.BaseClient    as BT (BaseClientEnv, checkHealth,+                                                     close, getClientEnv, ping)+import           Periodic.Trans.Job           (JobT, func_, name, workFail)+import           Periodic.Types               (ClientType (TypeWorker), Msgid,+                                               Packet, getClientType, getResult,+                                               packetREQ, regPacketREQ)+import           Periodic.Types.Job+import           Periodic.Types.ServerCommand+import           Periodic.Types.WorkerCommand+import           System.Log.Logger            (errorM)+import           UnliftIO+import           UnliftIO.Concurrent          (threadDelay)++type TaskList tp m = IOHashMap FuncName (JobT tp m ())++type JobList = IOList (Msgid, Job)++data WorkerEnv tp m = WorkerEnv+    { taskList :: TaskList tp m+    , jobList  :: JobList+    , taskSize :: TVar Int+    }++newtype WorkerT tp m a = WorkerT {unWorkerT :: ReaderT (WorkerEnv tp m) (JobT tp m) a}+  deriving+    ( Functor+    , Applicative+    , Monad+    , MonadIO+    , MonadReader (WorkerEnv tp m)+    )++instance MonadUnliftIO m => MonadUnliftIO (WorkerT tp m) where+  withRunInIO inner = WorkerT $+    ReaderT $ \r ->+      withRunInIO $ \run ->+        inner (run . runWorkerT r)++instance MonadTrans (WorkerT tp) where+  lift = WorkerT . lift . lift++runWorkerT :: WorkerEnv tp m -> WorkerT tp m a -> JobT tp m a+runWorkerT workerEnv = flip runReaderT workerEnv . unWorkerT++startWorkerT+  :: (MonadUnliftIO m, Transport tp)+  => TransportConfig tp -> WorkerT tp m () -> m ()+startWorkerT config m = do+  connEnv <- initConnEnv config+  r <- runConnT connEnv $ do+    Conn.send $ regPacketREQ TypeWorker+    Conn.receive++  let nid = case getClientType r of+              Data v -> v+              _      -> ""++  taskList <- newIOHashMap+  jobList <- newIOList+  taskSize <- newTVarIO 0++  jobEnv1 <- initEnv1 mapEnv connEnv Nothing (Nid nid) sessionGen++  let wEnv = WorkerEnv {..}++  runNodeT jobEnv1 $ do++    void $ async $ startNodeT_ (filterPacketM jobList) defaultSessionHandler+    runWorkerT wEnv $ do+      void . async $ forever $ do+        threadDelay $ 100 * 1000 * 1000+        checkHealth+      void . async . runJobT $ processJobQueue wEnv+      m++  where mapEnv =+          setNodeMode Multi+          . setSessionMode SingleAction+          . setDefaultSessionTimeout 100++filterPacketM :: MonadIO m => JobList -> Packet ServerCommand -> m Bool+filterPacketM jl rpkt = do+  case getResult Nothing getAssignJob (Just rpkt) of+    Nothing  -> return True+    Just job -> do+      IL.append jl (getPacketId rpkt, job)+      return False+++runJobT :: Monad m => JobT tp m a -> WorkerT tp m a+runJobT = WorkerT . lift++getClientEnv+  :: (Monad m, Transport tp)+  => WorkerT tp m (BT.BaseClientEnv (Maybe Job) tp)+getClientEnv = runJobT BT.getClientEnv++close :: (MonadUnliftIO m, Transport tp) => WorkerT tp m ()+close = runJobT BT.close++ping :: (MonadUnliftIO m, Transport tp) => WorkerT tp m Bool+ping = runJobT BT.ping++addFunc+  :: (MonadUnliftIO m, Transport tp)+  => FuncName -> JobT tp m () -> WorkerT tp m ()+addFunc f j = do+  runJobT $ withSessionT Nothing $ send (packetREQ $ CanDo f)+  ref <- asks taskList+  HM.insert ref f j++broadcast+  :: (MonadUnliftIO m, Transport tp)+  => FuncName -> JobT tp m () -> WorkerT tp m ()+broadcast f j = do+  runJobT $ withSessionT Nothing $ send (packetREQ $ Broadcast f)+  ref <- asks taskList+  HM.insert ref f j++removeFunc+  :: (MonadUnliftIO m, Transport tp)+  => FuncName -> WorkerT tp m ()+removeFunc f = do+  tskList <- asks taskList+  runJobT $ removeFunc_ tskList f++removeFunc_+  :: (MonadUnliftIO m, Transport tp)+  => TaskList tp m -> FuncName -> JobT tp m ()+removeFunc_ ref f = do+  withSessionT Nothing $ send (packetREQ $ CantDo f)+  HM.delete ref f++getAssignJob :: ServerCommand -> Maybe Job+getAssignJob (JobAssign job) = Just job+getAssignJob _               = Nothing++work+  :: (MonadUnliftIO m, Transport tp)+  => Int -> WorkerT tp m ()+work size = do+  tskSize <- asks taskSize+  runJobT $ do+    envs <- mapM (newSessionEnv (Just (-1))) =<< replicateM size nextSessionId+    forever $ do+      atomically $ do+        s <- readTVar tskSize+        when (s >= size) retrySTM++      mapM_ (`runSessionT_` (send $ packetREQ GrabJob)) envs+      threadDelay 10000000 -- 10s+++processJob :: (MonadUnliftIO m, Transport tp) => WorkerEnv tp m -> (Msgid, Job) -> JobT tp m ()+processJob WorkerEnv{..} (sid, job) = do+  atomically $ do+    s <- readTVar taskSize+    writeTVar taskSize (s + 1)+  withEnv (Just job) $ do+    f <- func_+    task <- HM.lookup taskList f+    case task of+      Nothing -> do+        removeFunc_ taskList f+        workFail+      Just task' ->+        catchAny task' $ \e -> do+          n <- name+          liftIO $ errorM "Periodic.Trans.Worker"+                 $ concat [ "Failing on running job { name = "+                          , n+                          , ", "+                          , show f+                          , " }"+                          , "\nError: "+                          , show e+                          ]+          workFail++  atomically $ do+    s <- readTVar taskSize+    writeTVar taskSize (s - 1)++  jobEnv <- newSessionEnv (Just (-1)) sid+  runSessionT_ jobEnv $ send (packetREQ GrabJob)++checkHealth+  :: (MonadUnliftIO m, Transport tp)+  => WorkerT tp m ()+checkHealth = runJobT BT.checkHealth++processJobQueue :: (MonadUnliftIO m, Transport tp) => WorkerEnv tp m -> JobT tp m ()+processJobQueue wEnv@WorkerEnv {..} = forever $ do+  jobs <- atomically $ do+    v <- IL.toListSTM jobList+    if null v then retrySTM+              else do+                IL.clearSTM jobList+                return v++  mapM_ (async . processJob wEnv) jobs
+ src/Periodic/Worker.hs view
@@ -0,0 +1,18 @@+module Periodic.Worker+  ( WorkerM+  , startWorkerM+  , ping+  , addFunc+  , broadcast+  , removeFunc+  , work+  , close+  ) where++import           Metro.TP.Socket       (Socket, socket)+import           Periodic.Trans.Worker++type WorkerM = WorkerT Socket IO++startWorkerM :: String -> WorkerM () -> IO ()+startWorkerM h = startWorkerT (socket h)