packages feed

jobqueue (empty) → 0.1.3

raw patch · 20 files changed

+1651/−0 lines, 20 filesdep +HDBCdep +HDBC-sqlite3dep +QuickChecksetup-changed

Dependencies added: HDBC, HDBC-sqlite3, QuickCheck, async, attoparsec, base, bytestring, containers, data-default, directory, fast-logger, hslogger, hspec, hzk, jobqueue, lifted-base, monad-control, monad-logger, mtl, network, split, stm, template-haskell, text, text-format, time, transformers-base

Files

+ LICENSE view
@@ -0,0 +1,22 @@+Copyright (c) 2013 GREE, Inc.++MIT License++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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ jobqueue.cabal view
@@ -0,0 +1,80 @@+Name:           jobqueue+Version:        0.1.3+Synopsis:       A job queue library+License:        MIT+License-File:   LICENSE+Author:         Kiyoshi Ikehara+Maintainer:     kiyoshi.ikehara at gree.net+Copyright:      GREE, Inc.+Build-Type:     Simple+Category:       Network, Client+Cabal-Version:  >=1.8+Description:+  Haskell JobQueue is a library used for building a job scheduler with priority queues.+  The state of jobs is stored in a backend database such as Apache Zookeeper or other +  highly reliable message queue systems.++Source-Repository head+  type:     git+  location: https://github.com/gree/haskell-jobqueue.git++Library+  Ghc-Options:     -Wall+  Build-Depends:   base                     >= 4 && < 5+                 , mtl                      > 2.2+                 , network                  >= 2.3.2+                 , hslogger+                 , text+                 , bytestring+                 , containers+                 , split+                 , time+                 , data-default+                 , stm                      >= 2.4+                 , hzk                      >= 2.0.0+                 , attoparsec+                 , data-default+                 , HDBC+                 , HDBC-sqlite3+                 , fast-logger+                 , monad-logger+                 , template-haskell+                 , text-format+                 , monad-control+                 , transformers-base+                 , lifted-base+  Hs-source-dirs:  src+  Exposed-modules: Network.JobQueue+                 , Network.JobQueue.Class+                 , Network.JobQueue.Types+                 , Network.JobQueue.Action+                 , Network.JobQueue.AuxClass+                 , Network.JobQueue.Job+                 , Network.JobQueue.Job.Internal+                 , Network.JobQueue.JobQueue+                 , Network.JobQueue.JobQueue.Internal+                 , Network.JobQueue.Backend+                 , Network.JobQueue.Backend.Class+                 , Network.JobQueue.Backend.Types+                 , Network.JobQueue.Backend.Zookeeper+                 , Network.JobQueue.Backend.Sqlite3+                 , Network.JobQueue.Logger+  Other-modules:   Network.JobQueue.Backend.Zookeeper.ZookeeperQueue+  Extensions:      DeriveDataTypeable++Test-Suite test+  Ghc-Options:     -threaded -O2+  Build-depends:   base+                 , bytestring+                 , network >= 2.3.2+                 , directory+                 , QuickCheck == 2.5.1.1+                 , hspec+                 , async+                 , jobqueue+                 , data-default+                 , stm+  Type:            exitcode-stdio-1.0+  Hs-source-dirs:  test+  Main-is:         Main.hs+
+ src/Network/JobQueue.hs view
@@ -0,0 +1,178 @@+{- |+  Module      : Network.JobQueue+  Copyright   : (c) Gree, Inc. 2013+  License     : MIT-style+  +  Maintainer  : Kiyoshi Ikehara <kiyoshi.ikehara@gree.net>+  Stability   : experimental+  Portability : portable++Haskell JobQueue is a library used for building a job scheduler with a priority queue.+The state of a job is stored in a backend database such as Apache Zookeeper or other +highly reliable mesage queue systems.++[@Unit@]++Unit represents each state in an entire state machine. Units are described as value+constructors in Haskell code.+Unit itself is not executable. To execute using job queue system, extra information such+as job identifier, scheduled time is needed. An instance of a unit is wrapped by a 'job'+and stored into the job queue with those information.++The code shown below describes how to define a Unit.++>  data JobUnit = HelloStep | WorldStep deriving (Show, Read)+>  +>  instance Unit JobUnit where++In this case, you define JobUnit type with 2 states, HelloStep and WorldStep.+This is the entire state machine of your job queue system.+You can define nested or child state machines by defining more complex data types as +long as they are serializable with read and show functions.++For more information, see "Network.JobQueue.Class".++[@Job@]++Each task executed by state machines (such as checking server state or repairing a+cluster) is called a 'job'.++A job is described as a particular state of a state machine. Each state only does one+thing (especially for modifying operations).+This prevents jobs ending in a failure state, which the state machine is unable to handle.++You don't have to know the internal data structure of a job, but need to understand+its when you write action code.++For more information, see "Network.JobQueue.Job".++[@Environment@]++Each unit can contain information used in the action of the state. But in many cases,+there is some information used by almost all states and it is convenient if there is +some kind of global data set that is accessible from all the state's actions.++For this reason, you can define global data structures called environment.+The enviroment can be retrieved using getEnv function in action monad.++>  env <- getEnv++For more information, see "Network.JobQueue.Class".++[@Action@]++An action is a function that is called with a unit. You can define actions with the +"process" function.++>    let withJobQueue = buildJobQueue loc name $ do+>          process $ \WorldStep -> commitIO (putStrLn "world") >> fin+>          process $ \HelloStep -> commitIO (putStr "hello, ") >> next WorldStep++In general, an action does the following things:++  * check if the precondition of the state is satisfied or not++  * do the action associated with the state++  * check the postcondition and return the next state.++For more information, see "Network.JobQueue.Action".++-}++{-# LANGUAGE ScopedTypeVariables #-}++module Network.JobQueue (+    buildJobQueue+  , runJobQueue+  , Job(..)+  , JobState(..)+  , Unit(..)+  , ActionM+  , JobM+  , JobActionState+  , process+  , createJob+  , fin+  , none+  , next+  , orNext+  , yield+  , fork+  , forkInTime+  , forkOnTime+  , abort+  , getEnv+  , param+  , commitIO+  , liftIO+  , module Network.JobQueue.Class+  , module Network.JobQueue.AuxClass+  , module Network.JobQueue.JobQueue+  , module Network.JobQueue.Logger+  ) where++import Prelude hiding (log)+import Control.Exception+import Control.Monad++import Network.JobQueue.Types+import Network.JobQueue.Class+import Network.JobQueue.AuxClass+import Network.JobQueue.JobQueue+import Network.JobQueue.Job+import Network.JobQueue.Logger++{- | Build a function that takes a function (('JobQueue' a -> 'IO' ()) -> IO ()) as its first parameter.++The following code executes jobs as long as the queue is not empty.++>  main' loc name = do+>    let withJobQueue = buildJobQueue loc name $ do+>          process $ \WorldStep -> commitIO (putStrLn "world") >> fin+>          process $ \HelloStep -> commitIO (putStr "hello, ") >> next WorldStep+>    withJobQueue $ loop (initJobEnv loc name [])+>    where+>      loop env jq = do+>        executeJob jq env+>        count <- countJobQueue jq+>        when (count > 0) $ loop env jq++The following code registers a job with initial state.++>  main' loc name = do+>    let withJobQueue = buildJobQueue loc name $ do+>          process $ \WorldStep -> commitIO (putStrLn "world") >> fin+>          process $ \HelloStep -> commitIO (putStr "hello, ") >> next WorldStep+>    withJobQueue $ \jq -> scheduleJob jq HelloStep++-}+buildJobQueue :: (Env e, Unit a) => String -- ^ locator (ex.\"zookeeper:\/\/192.168.0.1\/myapp\")+                 -> String          -- ^ queue name (ex. \"/jobqueue\")+                 -> JobM e a ()     -- ^ job construction function+                 -> ((JobQueue e a -> IO ()) -> IO ()) -- ^ job queue executor+buildJobQueue loc name jobm = \action -> do+  bracket (openSession loc) (closeSession) $ \session -> do+    jq <- openJobQueue session name jobm+    action jq+    closeJobQueue jq++{- | Run a job queue while there is at least one job in the queue.+-}+runJobQueue :: (Aux e, Env e, Unit a)+               => e+               -> String          -- ^ locator (ex.\"zookeeper:\/\/192.168.0.1\/myapp\")+               -> String          -- ^ queue name (ex. \"/jobqueue\")+               -> JobM e a ()     -- ^ job construction function+               -> IO ()+runJobQueue env loc name jobm = buildJobQueue loc name jobm loop+  where+    loop jq = do+      executeJob jq env+      count <- countJobQueue jq+      when (count > 0) $ loop jq++----------------------------------------------------------------------+-- Docs+----------------------------------------------------------------------+
+ src/Network/JobQueue/Action.hs view
@@ -0,0 +1,199 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}++module Network.JobQueue.Action (+    JobActionState+  , runActionState+  , runAction+  , getEnv+  , param+  , next+  , orNext+  , fin+  , none+  , fork+  , forkInTime+  , forkOnTime+  , abort+  , commitIO+  , liftIO+  , yield+  ) where++import Control.Applicative+import Control.Monad.Except+import Control.Monad.Reader+import Control.Monad.State+import Control.Exception (SomeException(..), toException)+import Control.Exception.Base (PatternMatchFail(..))+import Control.Monad.Logger (runLoggingT)+import Control.Exception.Lifted (catch)+import Control.Monad.Base ()++import Data.Maybe+import Data.Time.Clock+import Data.Default (Default, def)++import Network.JobQueue.Class+import Network.JobQueue.AuxClass+import Network.JobQueue.Types+import Network.JobQueue.Logger++runActionState :: (Env e, Unit a) => JobActionState e a -> ActionFn e a+runActionState (JobActionState { jobActions = actions } ) env ju = runActionState' actions+  where+    runActionState' actions' = case actions' of+      [] -> return $ Right Nothing+      (act:acts) -> do+        r <- act env ju+        case r of+          Right Nothing -> runActionState' acts+          _ -> return r++runAction :: (Aux e, Env e, Unit a) => +             e -> a -> ActionT e a IO () -> IO (Either Break (Maybe (RuntimeState a)))+runAction env ju action = do+  (e,r) <- flip runLoggingT (auxLogger env)+         $ flip runStateT Nothing+         $ flip runReaderT (ActionEnv env ju)+         $ runExceptT+         $ runAM $ do+             when (toBeLogged ju) $ $(logWarn) "{}" [desc ju]+             action `catch` handlePatternMatchFail `catch` handleSome+  return $ either Left (const $ Right r) e++handlePatternMatchFail :: (Aux e, Env e, Unit a) => PatternMatchFail -> ActionT e a IO ()+handlePatternMatchFail e = do+  s <- get+  if getCommits (fromMaybe def s) > 0+    then do+      ju <- getJobUnit <$> ask+      $(logError) "pattern match fail: ! ({})" [desc ju]+      throwError $ Unhandled (toException e)+    else none++handleSome :: (Aux e, Env e, Unit a) => SomeException -> ActionT e a IO b+handleSome e = do+  $(logError) "unhandled exception: {}" [show e]+  throwError $ Unhandled e++--------------------------------++{- | Get environment in action.+-}+getEnv :: (Env e, Unit a) => ActionM e a e+getEnv = getJobEnv <$> ask++{- | Get a parameter value with a key from the environment in action.+     This is a special function for ParamEnv.+-}+param :: (ParamEnv e, Unit a, Read b) => (String, String) -> ActionM e a b+param (key, defaultValue) = do+  env <- getEnv+  case maybeRead defaultValue of+    Nothing -> do+      $(logCritical) "internal error. no parse: " [show (key, defaultValue)]+      abort+    Just defaultValue' -> case lookup key (envParameters env) of+      Just value -> return (fromMaybe defaultValue' (maybeRead value))+      Nothing -> return (defaultValue')+  where+    maybeRead = fmap fst . listToMaybe . reads+      +----------------++{- | Do a dirty I/O action with a side effect to the external system.+     If it doesn't change the state of the external system, you should use liftIO instead.+-}+commitIO :: (Env e, Unit a) => IO b -> ActionM e a b+commitIO action = do+  do s <- get+     when (getCommits (fromMaybe def s) > 0) $ do+       ju <- getJobUnit <$> ask+       $(logWarn) "commitIO called twice! ({})" [desc ju]+  modify $ \s -> Just $ incrementCommits $ fromMaybe def s+  liftIO action++----------------++{- | Yield execution+-}+yield :: (Env e, Unit a) => ActionM e a ()+yield = do+  ju <- getJobUnit <$> ask+  forkWith ju Nothing++{- | Create a job with a unit and schedule it.+-}+fork :: (Env e, Unit a)+        => a -- ^ a unit+        -> ActionM e a ()+fork ju = forkWith ju Nothing++{- | Create a job with a unit and schedule it at a specific time.+-}+forkOnTime :: (Env e, Unit a)+              => UTCTime        -- ^ absolute time in UTC+              -> a              -- ^ a unit+              -> ActionM e a ()+forkOnTime t ju = forkWith ju (Just t)++{- | Create a job with a unit and schedule it after a few micro seconds.+-}+forkInTime :: (Env e, Unit a) => NominalDiffTime -> a -> ActionM e a ()+forkInTime tDiff ju = do+  currentTime <- liftIO $ getCurrentTime+  forkWith ju (Just (addUTCTime tDiff currentTime))++{- | Move to the next state immediately.+     After the execution of the action the job being processed will be+     moved to the given state. The next action will be invoked immediately+     and can continue to work without being interrupted by another job.+     NOTE: This overrides the next state if it is already set.+-}+next :: (Env e, Unit a)+        => a              -- ^ the next state+        -> ActionM e a ()+next ju = modify $ \s -> Just $ setNextJob ju $ fromMaybe def s++{- | Move to the next state immediately.+     This is different from "next" function because this doesn't override+     if the next job is already set.+-}+orNext :: (Env e, Unit a)+         => a              -- ^ the next state+         -> ActionM e a ()+orNext ju = modify $ \s -> Just $ setNextJobIfEmpty ju $ fromMaybe def s++{- | Finish a job.+-}+fin :: (Env e, Unit a) => ActionM e a ()+fin = modify $ \s -> Just $ emptyNextJob $ fromMaybe def s++{- | If the unit passed by the job queue system cannot be processed by the+     action function, the function should call this.+-}+none :: (Env e, Unit a) => ActionM e a ()+none = result Nothing++{- | Abort the execution of a state machine.+     If a critical problem is found and there is a need to switch to the failure state,+     call this function.+-}+abort :: (Env e, Unit a) => ActionM e a b+abort = do+  ju <- getJobUnit <$> ask+  throwError $ Failure ("aborted on " ++ desc ju)++---------------------------------------------------------------- PRIVATE++{- | Set the result of the action. (for internal use)+-}+result :: (Env e, Unit a) => Maybe (RuntimeState a) -> ActionM e a ()+result = modify . setResult++forkWith :: (Env e, Unit a) => a -> Maybe UTCTime -> ActionM e a ()+forkWith ju mt = modify $ \s -> Just $ addForkJob (ju, mt) $ fromMaybe def s
+ src/Network/JobQueue/AuxClass.hs view
@@ -0,0 +1,41 @@+-- Copyright (c) Gree, Inc. 2014+-- License: MIT-style++{-# LANGUAGE OverloadedStrings #-}++module Network.JobQueue.AuxClass where++import Control.Monad.Logger+import Control.Applicative+import System.Log.FastLogger+import System.Log.Logger+import System.Environment (getProgName)++import Network.JobQueue.Types+import Network.JobQueue.Job.Internal++import qualified Data.ByteString.Char8 as S8++class Aux a where+  auxLogger :: a -> Loc -> LogSource -> LogLevel -> LogStr -> IO ()+  auxLogger _ loc logsrc loglevel msg = do+    progName <- getProgName+    logFunc loglevel progName $ S8.unpack $ fromLogStr $ defaultLogStr loc logsrc loglevel msg+    where+      logFunc level = case level of+        LevelDebug -> debugM+        LevelInfo -> infoM+        LevelWarn -> warningM+        LevelError -> errorM+        LevelOther "notice" -> noticeM+        LevelOther "critical" -> criticalM+        LevelOther _ -> warningM++  auxHandleFailure :: (Unit b) => a -> Maybe (Job b) -> IO (Maybe (Job b))+  auxHandleFailure _ mjob = do+    case mjob of+      Just job -> Just <$> createJob Runnable (getRecovery (jobUnit job))+      Nothing -> return (Nothing)++  auxHandleAfterExecute :: (Unit b) => a -> Job b -> IO ()+  auxHandleAfterExecute _ _job = return ()
+ src/Network/JobQueue/Backend.hs view
@@ -0,0 +1,49 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE OverloadedStrings #-}++module Network.JobQueue.Backend (openBackend) where++import Prelude hiding (takeWhile)+import Control.Exception (throwIO)+import qualified Data.ByteString.Char8 as BS+import Network.JobQueue.Backend.Class ()+import Network.JobQueue.Backend.Types+import Network.JobQueue.Backend.Zookeeper+import Network.JobQueue.Backend.Sqlite3++import Data.Attoparsec.ByteString+import Control.Applicative++data Locator = Zookeeper String | Sqlite3 String+++{- | Open a backend database.+-}+openBackend :: String        -- ^ locator (eg. \"zookeeper://localhost:2181/myapp\", \"sqlite3://myapp.sqlite3\")+               -> IO Backend -- ^ backend+openBackend locator = case parseLocator locator of+  Just (Zookeeper connString) -> openZookeeperBackend connString+  Just (Sqlite3 localPath) -> openSqlite3Backend localPath+  _ -> throwIO $ userError "invalid locator"+++---------------------------------------------------------------- PRIVATE++parseLocator :: String -> Maybe Locator+parseLocator v = case parse locatorParser $ BS.pack v of+  Done _ locator -> Just locator+  Partial parse' -> case parse' "" of+    Done _ locator -> Just locator+    _ -> Nothing+  _ -> Nothing+  where+    locatorParser :: Parser (Locator)+    locatorParser = do+      scheme <- takeWhile (/= 58) <* string "://"+      case scheme of+        "zookeeper" -> Zookeeper <$> fmap BS.unpack (takeWhile1 (\_ -> True))+        "sqlite3" -> Sqlite3 <$> fmap BS.unpack (takeWhile1 (\_ -> True))+        _ -> fail "unknown scheme"+
+ src/Network/JobQueue/Backend/Class.hs view
@@ -0,0 +1,19 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++module Network.JobQueue.Backend.Class (BackendQueue(..)) where++import qualified Data.ByteString.Char8 as BS++class BackendQueue q where+  readQueue    :: q -> IO (Maybe (BS.ByteString, String))+  peekQueue    :: q -> IO (Maybe (BS.ByteString, String, String, Int))+  updateQueue  :: q -> String -> BS.ByteString -> Int -> IO (Bool)+  deleteQueue  :: q -> String -> IO (Bool)+  writeQueue   :: q -> BS.ByteString -> Int -> IO (String)+  listQueue    :: q -> IO ([BS.ByteString])+  itemsQueue   :: q -> IO ([String])+  countQueue   :: q -> IO (Int)+  closeQueue   :: q -> IO ()+  closeQueue _ = return ()+
+ src/Network/JobQueue/Backend/Sqlite3.hs view
@@ -0,0 +1,112 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Network.JobQueue.Backend.Sqlite3 (openSqlite3Backend, newSqlite3Backend) where++import qualified Data.ByteString.Char8 as BS+import Database.HDBC+import Database.HDBC.Sqlite3+import Network.JobQueue.Backend.Types+import Network.JobQueue.Backend.Class+import Control.Concurrent.MVar+import Control.Exception++data Sqlite3Queue = Sqlite3Queue+  { conn      :: Connection+  , queueName :: String+  , mlock     :: MVar ()+  }++instance BackendQueue Sqlite3Queue where+  readQueue    = readDBQueue+  peekQueue    = peekDBQueue+  updateQueue  = updateDBQueue+  deleteQueue  = deleteDBQueue+  writeQueue   = writeDBQueue+  listQueue    = listDBQueue+  itemsQueue   = itemsDBQueue+  countQueue   = countDBQueue+  closeQueue   = const $ return ()++openSqlite3Backend :: String -> IO Backend+openSqlite3Backend filePath = do+  c <- connectSqlite3 filePath+  m <- newMVar ()+  return $ Backend {+      bOpenQueue = \qn -> do+        _ <- withLock m $ withTransaction c $ \c' -> do+          run c' ("CREATE TABLE IF NOT EXISTS '" ++ qn ++ "' (key INTEGER PRIMARY KEY AUTOINCREMENT, prio INTEGER, value TEXT, version INTEGER)") []+        return (Sqlite3Queue c qn m)+    , bClose = disconnect c+    }++newSqlite3Backend :: Connection -> IO Backend+newSqlite3Backend c = do+  m <- newMVar ()+  return $ Backend {+      bOpenQueue = \qn -> do+        return (Sqlite3Queue c qn m)+    , bClose = return ()+    }++readDBQueue :: Sqlite3Queue -> IO (Maybe (BS.ByteString, String))+readDBQueue Sqlite3Queue {..} = withLock mlock $ withTransaction conn $ \conn' -> do+  sqlvalues <- quickQuery' conn' ("SELECT key, value FROM '" ++ queueName ++ "' ORDER BY prio, key LIMIT 1") []+  case sqlvalues of+    ((key:value:_):_) -> do+      _ <- run conn' ("DELETE FROM '" ++ queueName ++ "' WHERE key = ?") [toSql key]+      return (Just (fromSql value, fromSql key))+    _ -> return (Nothing)++peekDBQueue :: Sqlite3Queue -> IO (Maybe (BS.ByteString, String, String, Int))+peekDBQueue Sqlite3Queue {..} = withLock mlock $ withTransaction conn $ \conn' -> do+  sqlvalues <- quickQuery' conn' ("SELECT key, value, version FROM '" ++ queueName ++ "' ORDER BY prio, key LIMIT 1") []+  case sqlvalues of+    ((key:value:version:_):_) -> return (Just (fromSql value, fromSql key, fromSql key, fromSql version))+    _ -> return (Nothing)++writeDBQueue :: Sqlite3Queue -> BS.ByteString -> Int -> IO (String)+writeDBQueue Sqlite3Queue {..} value prio = do+  withLock mlock $ withTransaction conn $ \conn' -> do+    _ <- run conn' ("INSERT INTO '" ++ queueName ++ "'(prio, value, version) VALUES (?,?,0)") [toSql prio, toSql value]+    sqlvalues <- quickQuery' conn' ("SELECT seq FROM sqlite_sequence where name = '" ++ queueName ++ "'") []+    case sqlvalues of+      ((key:_):_) -> do+        return (fromSql key)+      _ -> return ("")++deleteDBQueue :: Sqlite3Queue -> String -> IO (Bool)+deleteDBQueue Sqlite3Queue {..} key = withLock mlock $ withTransaction conn $ \conn' -> do+  _ <- run conn' ("DELETE FROM '" ++ queueName ++ "' WHERE key = ?") [toSql key]+  return (True)++updateDBQueue :: Sqlite3Queue -> String -> BS.ByteString -> Int -> IO (Bool)+updateDBQueue Sqlite3Queue {..} key value version = do+  withLock mlock $ withTransaction conn $ \conn' -> do+    nrows <- run conn' ("UPDATE '" ++ queueName ++ "' SET value = ?, version = ? WHERE key = ? AND version = ?") [toSql value, toSql (version+1), toSql key, toSql version]+    return $ if nrows > 0 then True else False++countDBQueue :: Sqlite3Queue -> IO (Int)+countDBQueue Sqlite3Queue {..} = withLock mlock $ withTransaction conn $ \conn' -> do+  sqlvalues <- quickQuery' conn' ("SELECT COUNT (*) FROM '" ++ queueName ++ "' ORDER BY prio, key LIMIT 1") []+  case sqlvalues of+    ((count:_):_) -> return (fromSql count)+    _ -> return (0)++itemsDBQueue :: Sqlite3Queue -> IO ([String])+itemsDBQueue Sqlite3Queue {..} = withLock mlock $ withTransaction conn $ \conn' -> do+  sqlvalues <- quickQuery' conn' ("SELECT key FROM '" ++ queueName ++ "' ORDER BY prio, key") []+  case sqlvalues of+    keys -> return (map (fromSql . head) keys)++listDBQueue :: Sqlite3Queue -> IO ([BS.ByteString])+listDBQueue Sqlite3Queue {..} = withLock mlock $ withTransaction conn $ \conn' -> do+  sqlvalues <- quickQuery' conn' ("SELECT value FROM '" ++ queueName ++ "' ORDER BY prio, key") []+  case sqlvalues of+    keys -> return (map (fromSql . head) keys)++withLock :: MVar () -> IO a -> IO a+withLock m act = handleSql (\err -> throwIO $ SessionError (show err)) $ do+  bracket (takeMVar m) (putMVar m) $ const act
+ src/Network/JobQueue/Backend/Types.hs view
@@ -0,0 +1,24 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE GADTs #-}+{-# LANGUAGE DeriveDataTypeable #-}++module Network.JobQueue.Backend.Types (Backend(..), BackendError(..)) where++import Control.Exception+import Network.JobQueue.Backend.Class+import Data.Typeable++data Backend where+  Backend :: (BackendQueue q) => {+      bOpenQueue :: String -> IO q+    , bClose :: IO ()+    } -> Backend++data BackendError = +    NotFound String+  | SessionError String+  deriving (Show, Typeable)++instance Exception BackendError
+ src/Network/JobQueue/Backend/Zookeeper.hs view
@@ -0,0 +1,58 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++module Network.JobQueue.Backend.Zookeeper (+    openZookeeperBackend+  , newZookeeperBackend+  ) where++import qualified Database.Zookeeper as Z+import Control.Concurrent+import Control.Concurrent.STM+import Network.JobQueue.Backend.Types+import Network.JobQueue.Backend.Zookeeper.ZookeeperQueue++openZookeeperBackend :: String -> IO Backend+openZookeeperBackend endpoint = do+  Z.setDebugLevel Z.ZLogError+  zvar <- newTVarIO Nothing+  stateVar <- newTVarIO Z.ConnectingState+  _ <- forkIO $ Z.withZookeeper endpoint 100000 (Just $ watcher stateVar) Nothing $ \z -> do+    atomically $ do+      state <- readTVar stateVar+      case state of+        Z.ConnectingState -> retry+        _ -> return ()+    atomically $ writeTVar zvar (Just z)+    atomically $ do+      mz <- readTVar zvar+      case mz of+        Just _ -> retry+        Nothing -> return ()+  return $ Backend {+      bOpenQueue = openQueue zvar+    , bClose = atomically $ writeTVar zvar Nothing+    }+  where+    openQueue :: TVar (Maybe Z.Zookeeper) -> String -> IO (ZookeeperQueue)+    openQueue zvar queueName = do+      z <- atomically $ readTVar zvar >>= maybe retry return+      zq <- initZQueue z (basePath queueName) Z.OpenAclUnsafe+      return zq++    watcher :: TVar Z.State -> Z.Watcher+    watcher stateVar _z event state _mZnode = do+      case event of+        Z.SessionEvent -> atomically $ writeTVar stateVar state+        _ -> return ()++newZookeeperBackend :: Z.Zookeeper -> Backend+newZookeeperBackend zh = Backend {+      bOpenQueue = \queueName -> initZQueue zh (basePath queueName) Z.OpenAclUnsafe+      , bClose = return ()+    }++basePath :: String -> String+basePath queueName = case queueName of+  '/':_ -> queueName+  _ -> '/':queueName
+ src/Network/JobQueue/Backend/Zookeeper/ZookeeperQueue.hs view
@@ -0,0 +1,228 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++module Network.JobQueue.Backend.Zookeeper.ZookeeperQueue (+    ZookeeperQueue+  , initZQueue+  , readZQueue+  , peekZQueue+  , updateZQueue+  , deleteZQueue+  , writeZQueue+  , destroyZQueue+  , listZQueue+  , itemsZQueue+  , countZQueue+  ) where++import qualified Database.Zookeeper as Z+import qualified Data.ByteString.Char8 as C+import Control.Exception hiding (handle)+import Data.List+import Control.Monad+import Data.Maybe+import Data.List.Split++import Network.JobQueue.Backend.Class+import Network.JobQueue.Backend.Types++data ZookeeperQueue = ZookeeperQueue {+    zqHandle         :: Z.Zookeeper+  , zqBasePath       :: String+  , zqNodeName       :: String+  , zqAcls           :: Z.AclList+  }++instance BackendQueue ZookeeperQueue where+  readQueue    = readZQueue+  peekQueue    = peekZQueue+  updateQueue  = updateZQueue+  deleteQueue  = deleteZQueue+  writeQueue   = writeZQueue+  listQueue    = listZQueue+  itemsQueue   = itemsZQueue+  countQueue   = countZQueue+  ++maxPrio :: Int+maxPrio = 999++minPrio :: Int+minPrio = -999++qnPrefix :: String+qnPrefix = "qn-"++---- init++initZQueue :: Z.Zookeeper -> String -> Z.AclList -> IO (ZookeeperQueue)+initZQueue z path acls = do+  e <- createZnodeRecursively z path Nothing acls []+  case e of+    Right _ -> return ()+    Left zkerr -> throwZKError "initZQueue" zkerr+  return (ZookeeperQueue z path qnPrefix acls)++-- take+readZQueue :: ZookeeperQueue -> IO (Maybe (C.ByteString, String))+readZQueue zkQueue = do+  children <- getChildren zkQueue+  case children of+    [] -> return (Nothing)+    _  -> takeHead (sortChildren children)+  where+    takeHead [] = return (Nothing)+    takeHead (nodeName:xs) = do+      let path = zqBasePath zkQueue ++ "/" ++ nodeName+      e <- Z.get (zqHandle zkQueue) path Nothing+      case e of+        Right (Just value, _stat) -> do+          e' <- Z.delete (zqHandle zkQueue) path Nothing+          case e' of+            Right () -> return (Just (value, nodeName))+            Left _zkerr -> do+              r <- Z.exists (zqHandle zkQueue) path Nothing+              case r of+                Right _stat -> takeHead (nodeName:xs)+                Left Z.NoNodeError -> takeHead xs+                Left zkerr -> throwZKError "readZQueue" zkerr+        Right (Nothing, _stat) -> takeHead xs -- ignore if the content is empty+        Left Z.NoNodeError -> return (Nothing)+        Left zkerr -> throwZKError "readZQueue" zkerr++-- peek+peekZQueue :: ZookeeperQueue -> IO (Maybe (C.ByteString, String, String, Int))+peekZQueue zkQueue = do+  children <- getChildren zkQueue+  case children of+    [] -> return Nothing+    _  -> getHead (sortChildren children)+  where+    idSuffixLen :: Int+    idSuffixLen = 10+    +    getHead :: [String] -> IO (Maybe (C.ByteString, String, String, Int))+    getHead [] = return Nothing+    getHead (x:xs) = do+      e <- Z.get (zqHandle zkQueue) (fullPath zkQueue x) Nothing+      case e of+        Right (mValue, stat) -> do+          case mValue of+            Just v -> return $ Just (v, x, drop (length x - idSuffixLen) x, fromIntegral $ Z.statVersion stat)+            Nothing -> getHead xs+        Left Z.NoNodeError -> peekZQueue zkQueue+        Left zkerr -> throwZKError "peekZQueue" zkerr+    +-- update+updateZQueue :: ZookeeperQueue -> String -> C.ByteString -> Int -> IO (Bool)+updateZQueue zkQueue znodeName value version = do+  e <- Z.set (zqHandle zkQueue) (fullPath zkQueue znodeName) (Just value) (Just (fromIntegral version))+  case e of+    Right _stat -> return (True)+    Left Z.BadVersionError -> return (False)+    Left Z.NoNodeError -> return (False)+    Left zkerr -> throwZKError "updateZQueue" zkerr++-- delete+deleteZQueue :: ZookeeperQueue -> String -> IO (Bool)+deleteZQueue zkQueue nodeName = do+  let nodeName' = zqBasePath zkQueue ++ "/" ++ nodeName+  e <- Z.delete (zqHandle zkQueue) nodeName' Nothing+  case e of+    Right () -> return (True)+    Left Z.NoNodeError -> throwIO $ NotFound nodeName+    Left zkerr -> throwZKError ("deleteZQueue(nodeName=" ++ nodeName' ++ ")") zkerr++-- offer+writeZQueue :: ZookeeperQueue -> C.ByteString -> Int -> IO (String)+writeZQueue zkQueue value prio = do+  r <- Z.create (zqHandle zkQueue)+                (zqBasePath zkQueue ++ "/" ++ (nodePrefix (zqNodeName zkQueue) prio))+                (Just value)+                (zqAcls zkQueue)+                [Z.Sequence]+  case r of+    Right znode -> return $ head $ reverse (splitOn "/" znode)+    Left zkerr -> throwZKError "writeZQueue" zkerr++-- destroy+destroyZQueue :: ZookeeperQueue -> IO ()+destroyZQueue _zkQueue = return ()++-- elems+listZQueue :: ZookeeperQueue -> IO ([C.ByteString])+listZQueue zkQueue = do+  results <- getChildren zkQueue+  values <- forM (sortChildren results) getItem+  return (catMaybes values)+  where+    getItem x = do+      e <- Z.get (zqHandle zkQueue) (zqBasePath zkQueue ++ "/" ++ x) Nothing   +      case e of+        Right (mValue, _stat) -> return (mValue)+        Left Z.NoNodeError -> return (Nothing)+        Left zkerr -> throwZKError "listZQueue" zkerr++-- items+itemsZQueue :: ZookeeperQueue -> IO ([String])+itemsZQueue zkQueue = do+  items <- getChildren zkQueue+  return (sortChildren items)++-- count+countZQueue :: ZookeeperQueue -> IO (Int)+countZQueue zkQueue = do+  items <- getChildren zkQueue+  return (length items)++----++getChildren :: ZookeeperQueue -> IO ([String])+getChildren zkQueue = do+  e <- Z.getChildren (zqHandle zkQueue) (zqBasePath zkQueue) Nothing+  case e of+    Right results -> return (results)+    Left zkerr -> throwZKError "getChildren" zkerr++sortChildren :: [String] -> [String]+sortChildren = sort . filter (isPrefixOf qnPrefix)++fullPath :: ZookeeperQueue -> String -> String+fullPath zkQueue x = (zqBasePath zkQueue ++ "/" ++ x)++nodePrefix :: String -> Int -> String+nodePrefix base prio = base ++ priorityPart' ++ "-"+  where+    priority = if prio > maxPrio then maxPrio else (if prio < minPrio then minPrio else prio)+    plus = priority >= 0+    priorityPart = show $ if plus then abs priority else maxPrio + 1 + priority+    priorityPart' = (if plus then "0" else "-")+                    ++ (take (3 - length priorityPart) $ repeat '0')+                    ++ priorityPart++throwZKError :: String -> Z.ZKError -> IO a+throwZKError func zkerr = throwIO $ SessionError (func ++ ": " ++ show zkerr)++createZnodeRecursively :: Z.Zookeeper -> String -> Maybe C.ByteString -> Z.AclList -> [Z.CreateFlag] -> IO (Either Z.ZKError String)+createZnodeRecursively z path mData acls flags = do+  createZnodeRecursively' z (reverse $ splitOn "/" path) mData acls flags++createZnodeRecursively' :: Z.Zookeeper -> [String] -> Maybe C.ByteString -> Z.AclList -> [Z.CreateFlag] -> IO (Either Z.ZKError String)+createZnodeRecursively' _ [] _ _ _ = return $ Right "/"+createZnodeRecursively' _ ("":[]) _ _ _ = return $ Right "/"+createZnodeRecursively' z revZnodes value acls cflags = do+  let path = intercalate "/" (reverse revZnodes)+  eStats <- Z.exists z path Nothing+  case eStats of+    Right _stat -> return $ Right path+    Left Z.NoNodeError -> do+      e <- createZnodeRecursively' z (tail revZnodes) Nothing acls cflags+      case e of+        Right _ -> do+          r <- Z.create z path value acls cflags+          return $ case r of+            Right newPath -> Right newPath+            Left zkerr -> Left zkerr+        Left zkerr -> return (Left zkerr)+    Left zkerr -> return (Left zkerr)+
+ src/Network/JobQueue/Class.hs view
@@ -0,0 +1,47 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{- | Class definitions+-}+module Network.JobQueue.Class where++{- | Environment class+-}+class Env a where++{- | Environment with a parameter set+-}+class (Env a) => ParamEnv a where+  envParameters :: a -> [(String, String)]+  envParameters _env = []++{- | Description class+-}+class (Show a) => Desc a where+  {- | Define the description of a unit.+  -}+  desc :: a -> String+  desc x = show x+  +  {- | Define the short description of a unit.+  -}+  shortDesc :: a -> String+  shortDesc x = takeWhile (/= ' ') $ show x++{- | Unit class+-}+class (Read a, Show a, Desc a, Eq a) => Unit a where+  {- | Define the priority of a unit.+  -}+  getPriority :: a -> Int+  getPriority _ju = 1+  +  {- | Define the recovery state of a unit.+  -}+  getRecovery :: a -> a+  getRecovery ju = ju++  {- | Define the logging necessity of a unit.+  -}+  toBeLogged :: a -> Bool+  toBeLogged _ju = False
+ src/Network/JobQueue/Job.hs view
@@ -0,0 +1,36 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++module Network.JobQueue.Job (+    Job(jobState, jobUnit, jobCTime, jobOnTime, jobId, jobGroup, jobPriority, StopTheWorld)+  , JobState(..)+  , buildActionState+  , process+  , createJob+  , createOnTimeJob+  , printJob+  , module Network.JobQueue.Types+  , module Network.JobQueue.Action+  ) where++import Control.Monad.State hiding (state)++import Network.JobQueue.Class+import Network.JobQueue.AuxClass+import Network.JobQueue.Types+import Network.JobQueue.Action+import Network.JobQueue.Job.Internal++--------------------------------++buildActionState :: (Env e, Unit a) => JobM e a () -> IO (JobActionState e a)+buildActionState jobs = execStateT (runS jobs) (JobActionState [])++{- | Declare a function which accepts a unit and execute the action of it if possible.+-}+process :: (Aux e, Env e, Unit a) => (a -> ActionM e a ()) -> JobM e a ()+process action = modify $ addAction $ eval action++eval :: (Aux e, Env e, Unit a) => (a -> ActionM e a ()) -> ActionFn e a+eval action env ju = runAction env ju (action ju)+
+ src/Network/JobQueue/Job/Internal.hs view
@@ -0,0 +1,67 @@++module Network.JobQueue.Job.Internal where++import Data.Time.Clock+import System.Log.Logger+import System.IO++import Network.JobQueue.Class++data JobState = Initialized | Runnable | Running | Aborted | Finished+  deriving (Show, Read, Eq)++{- | Job control block+Job consists of /State/, /Unit/, /CTime/, /OnTime/, /Id/, /Group/, and /Priority/.++- State - takes one of 5 states (initialized, runnable, running, aborted and finished)++- Unit - an instance of Unit class, which is specified by type parameter of Job data type++- CTime - creation time++- OnTime - the time at which this job starts++- Id - Identifier of this job++- Group - Group ID of this job++- Priority - the priority of this job++-}+data Job a =+    Job {+      jobState    :: JobState+    , jobUnit     :: a+    , jobCTime    :: UTCTime+    , jobOnTime   :: UTCTime+    , jobId       :: Int+    , jobGroup    :: Int+    , jobPriority :: Int }+  | StopTheWorld+  deriving (Show, Read, Eq)++createJob :: (Unit a) => JobState -> a -> IO (Job a)+createJob state unit = do+  ctime <- getCurrentTime+  return (Job state unit ctime ctime (defaultId) (defaultGroup) (getPriority unit))++createOnTimeJob :: (Unit a) => JobState -> UTCTime -> a -> IO (Job a)+createOnTimeJob state ontime unit = do+  ctime <- getCurrentTime+  return (Job state unit ctime ontime (defaultId) (defaultGroup) (getPriority unit))++printJob :: (Unit a) => Job a -> IO ()+printJob job = case job of+  Job {} -> do+    noticeM "job" $ show (jobUnit job)+    hPutStrLn stdout $ desc (jobUnit job)+    hFlush stdout+  StopTheWorld -> return ()++---------------------------------------------------------------- PRIVATE++defaultId :: Int+defaultId = -1++defaultGroup :: Int+defaultGroup = -1
+ src/Network/JobQueue/JobQueue.hs view
@@ -0,0 +1,185 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE ScopedTypeVariables #-}++module Network.JobQueue.JobQueue (+    JobQueue+  , Session+  , openSession+  , newSession+  , closeSession+  , openJobQueue+  , closeJobQueue+  , countJobQueue+  , resumeJobQueue+  , suspendJobQueue+  , executeJob+  , scheduleJob+  , deleteJob+  , clearJobs+  , peekJob+  ) where++import Control.Applicative+import Control.Concurrent+import Control.Monad+import Control.Exception+import qualified Data.ByteString.Char8 as BS+import Data.Maybe++import Network.JobQueue.Class+import Network.JobQueue.AuxClass+import Network.JobQueue.Types+import Network.JobQueue.Job+import Network.JobQueue.Backend+import Network.JobQueue.Backend.Class+import Network.JobQueue.Backend.Types++import Network.JobQueue.JobQueue.Internal+++{- | A session handler++     A session usually represents a database session to access job queues stored in the backend+     database.+-}+data Session = Session Bool String Backend++{- | Open a queue session with a resource locator+-}+openSession :: String          -- ^ a resource locator+               -> IO (Session) -- ^ a session handler+openSession locator = Session True locator <$> openBackend locator++{- | Create a queue session with a backend handler+-}+newSession :: String                -- ^ a resource locator (dummy)+              -> Backend -> Session -- ^ a session handler+newSession dummyLocator backend = Session False dummyLocator backend++{- | Close a queue session if needed+-}+closeSession :: Session -> IO ()+closeSession (Session isOwner _locator backend) = when isOwner $ bClose backend++{- | Open a job queue with a session.+-}+openJobQueue :: (Env e, Unit a)+                => Session     -- ^ a session handler+                -> String      -- ^ a queue name+                -> JobM e a () -- ^ a state machine definition+                -> IO (JobQueue e a)+openJobQueue (Session _isOwner _locator _backend@(Backend { bOpenQueue = oq }))+             name+             jobm = do+  JobQueue <$> oq name <*> buildActionState jobm++{- | Close a job queue.+-}+closeJobQueue :: (Env e, Unit a) => JobQueue e a -> IO ()+closeJobQueue JobQueue { jqBackendQueue = bq } = closeQueue bq++{- | Count the number of jobs queued in a job queue.+-}+countJobQueue :: (Env e, Unit a) => JobQueue e a -> IO (Int)+countJobQueue JobQueue { jqBackendQueue = bq } = countQueue bq++{- | Resume a job queue+-}+resumeJobQueue :: (Env e, Unit a) => JobQueue e a -> IO (Bool)+resumeJobQueue jobqueue = do+  r <- peekJob' jobqueue+  case r of+    Just (job, nodeName, idName, _version) -> case actionForJob job idName of+      Execute StopTheWorld -> resume jobqueue nodeName+      _ -> return True+    _ -> return True+  where+    resume JobQueue { jqBackendQueue = bq } key = deleteQueue bq key++{- | Suspend a job queue+-}+suspendJobQueue :: forall e. forall a. (Env e, Unit a) => JobQueue e a -> IO (Bool)+suspendJobQueue jobqueue = do+  r <- peekJob' jobqueue+  case r of+    Just (job, _nodeName, idName, _version) -> case actionForJob job idName of+      Execute StopTheWorld -> return False+      _ -> suspend jobqueue >> return True+    _ -> suspend jobqueue >> return True+  where+    suspend JobQueue { jqBackendQueue = bq } = writeQueue bq (pack (StopTheWorld :: Job a)) (-1)++{- | Execute an action of the head job in a job queue.+-}+executeJob :: (Aux e, Env e, Unit a) => JobQueue e a -> e -> IO ()+executeJob jobqueue env = do+  r <- peekJob' jobqueue+  case r of+    Just (job, nodeName, idName, version) -> case actionForJob job idName of+      Execute StopTheWorld -> do+        threadDelay 1000000+        return ()+      Execute job' -> do+        isUpdated <- updateJob jobqueue nodeName job' version+        when (isUpdated && jobState job == Runnable && jobState job' == Running) $ do+          executeJob' jobqueue env nodeName job' version >>= afterExecuteJob jobqueue env nodeName job' version+          auxHandleAfterExecute env job'+      Delete -> do+        void $ deleteJob jobqueue nodeName+        executeJob jobqueue env+      Skip -> return ()+    Nothing -> return ()++{- | Schedule a job.+-}+scheduleJob :: (Unit a)+               => JobQueue e a -- ^ a job queue+               -> a            -- ^ a unit+               -> IO ()+scheduleJob JobQueue { jqBackendQueue = bq } ju = do+  job <- createJob Initialized ju+  void $ writeQueue bq (pack job) (getPriority ju)++{- | Delete a job from a job queue.+-}+deleteJob :: (Unit a)+             => JobQueue e a -- ^ a job queue+             -> String       -- ^ a job identifier+             -> IO Bool+deleteJob JobQueue { jqBackendQueue = bq } nodeName = do+  deleteQueue bq nodeName `catch` \e -> case e of+    NotFound _ -> return True+    _ -> throwIO e++{- | Clear all jobs from a job queue.+-}+clearJobs :: (Unit a)+             => JobQueue e a -- ^ a job queue+             -> IO [(String, Job a)]+clearJobs JobQueue { jqBackendQueue = bq } = loop []+  where+    loop dequeued = do+      obj <- readQueue bq+      case obj of+        Nothing -> return dequeued+        Just (bs, nodeName) -> case (fmap fst . listToMaybe . reads) $ BS.unpack bs of+          Nothing -> return dequeued+          Just job -> loop ((nodeName, job):dequeued)++{- | Peek a job form a job queue.+-}+peekJob :: (Unit a)+           => JobQueue e a -- ^ a job queue+           -> IO (Maybe (Job a))+peekJob jobqueue = do+  mjob <- peekJob' jobqueue+  return $ case mjob of+    Just (job, _nodeName, _idName, _version) -> Just job+    Nothing -> Nothing++---------------------------------------------------------------- PRIVATE++  +
+ src/Network/JobQueue/JobQueue/Internal.hs view
@@ -0,0 +1,112 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE GADTs #-}++module Network.JobQueue.JobQueue.Internal where++import qualified Data.ByteString.Char8 as BS+import Control.Exception+import Data.Time.Clock+import Control.Monad+import Data.Maybe++import Network.JobQueue.Class+import Network.JobQueue.AuxClass+import Network.JobQueue.Types+import Network.JobQueue.Action+import Network.JobQueue.Job+import Network.JobQueue.Backend.Class+import Network.JobQueue.Backend.Types+++data JobQueue e a where+  JobQueue :: (BackendQueue q) => {+    jqBackendQueue :: q+  , jqActionState :: JobActionState e a+  } -> JobQueue e a++data ActionForJob a = (Unit a) => Execute (Job a) | Delete | Skip+++actionForJob :: Unit a => Job a -> String -> ActionForJob a+actionForJob job idName = do+  case job of+    StopTheWorld -> Execute job+    _ -> case jobState job of+           Initialized -> case (fmap fst . listToMaybe . reads) idName of+             Just ident -> Execute $ job { jobState = Runnable, jobId = ident }+             Nothing -> Execute $ job { jobState = Runnable, jobId = (-1) }+           Runnable -> Execute $ job { jobState = Running }+           Running -> Skip+           Aborted -> Skip+           Finished -> Delete++peekJob' :: (Unit a) => JobQueue e a -> IO (Maybe (Job a, String, String, Int))+peekJob' JobQueue { jqBackendQueue = bq } = do+  obj <- peekQueue bq+  case obj of+    Nothing -> return (Nothing)+    Just (value, nodeName, idName, version) -> do+      case (fmap fst . listToMaybe . reads) $ BS.unpack value of+        Nothing -> return (Nothing)+        Just job -> return (Just (job, nodeName, idName, version))++executeJob' :: (Aux e, Env e, Unit a) => JobQueue e a -> e -> String -> Job a -> Int -> IO (Either Break (Maybe (RuntimeState a)))+executeJob' jqueue@JobQueue { jqBackendQueue = bq, jqActionState = actionState } env nodeName currentJob version = do+  currentTime <- getCurrentTime+  if jobOnTime currentJob < currentTime+    then do+      runActionState actionState env (jobUnit currentJob)+    else do+      r <- updateJob jqueue nodeName currentJob { jobState = Finished } (version+1)+      when r $ void $ writeQueue bq (pack $ currentJob { jobState = Runnable } ) (jobPriority currentJob)+      return $ Right Nothing++afterExecuteJob :: (Aux e, Env e, Unit a) => JobQueue e a -> e -> String -> Job a -> Int -> Either Break (Maybe (RuntimeState a)) -> IO ()+afterExecuteJob jqueue env nodeName currentJob version mResult = case mResult of+  Right (Just (RS mNextJu forks _)) -> do+    case mNextJu of+      Just nextJu -> do+        _r <- updateJob jqueue nodeName currentJob { jobState = Runnable, jobUnit = nextJu } (version+1)+        return ()+      Nothing -> do+        _r <- updateJob jqueue nodeName currentJob { jobState = Finished } (version+1)+        return ()+    forM_ (reverse forks) $ \(forked, ontime) -> rescheduleJob jqueue ontime forked+  Right (Nothing) -> recover Nothing -- nothing to do anymore+  Left (Failure _msg) -> do+    n <- auxHandleFailure env (Just currentJob)+    recover n+  Left Retriable -> do+    _r <- updateJob jqueue nodeName currentJob { jobState = Runnable } (version+1)+    return ()+  Left (Unhandled _someException) -> do+    _r <- updateJob jqueue nodeName currentJob { jobState = Finished } (version+1)+    return ()+  where+    recover n = case n of+      Just nextJu -> do+        _r <- updateJob jqueue nodeName nextJu (version+1)+        return ()+      Nothing -> do+        _r <- updateJob jqueue nodeName currentJob { jobState = Finished } (version+1)+        return ()++rescheduleJob :: (Unit a) => JobQueue e a -> Maybe UTCTime -> a -> IO ()+rescheduleJob JobQueue { jqBackendQueue = bq } mOntime ju = do+  job <- case mOntime of+    Just ontime -> createOnTimeJob Initialized ontime ju+    Nothing -> createJob Initialized ju+  void $ writeQueue bq (pack $ job) (getPriority ju)++updateJob :: (Unit a) => JobQueue e a -> String -> Job a -> Int -> IO (Bool)+updateJob JobQueue { jqBackendQueue = bq } nodeName job version = do+  updateQueue bq nodeName (pack job) version `catch` handleError+  where+    handleError :: BackendError -> IO (Bool)+    handleError _ = return (False)++pack :: (Unit a) => Job a -> BS.ByteString+pack = BS.pack . show+
+ src/Network/JobQueue/Logger.hs view
@@ -0,0 +1,41 @@++{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE OverloadedStrings #-}++module Network.JobQueue.Logger+  ( logDebug+  , logInfo+  , logWarn+  , logError+  , logNotice+  , logCritical+  , Only+  ) where++import qualified Data.Text as T+import qualified Data.Text.Lazy as LT+import qualified Control.Monad.Logger as ML+import Language.Haskell.TH.Syntax (Q, Exp, qLocation)+import Data.Text.Format++logTH :: ML.LogLevel -> Q Exp+logTH level = [|\a b -> ML.monadLoggerLog $(qLocation >>= ML.liftLoc) (T.pack "") level (LT.toStrict $ format (a :: Format) b)|]++logDebug :: Q Exp+logDebug = logTH ML.LevelDebug++logInfo :: Q Exp+logInfo = logTH ML.LevelInfo++logWarn :: Q Exp+logWarn = logTH ML.LevelWarn++logError :: Q Exp+logError = logTH ML.LevelError++logNotice :: Q Exp+logNotice = logTH (ML.LevelOther "notice")++logCritical :: Q Exp+logCritical = logTH (ML.LevelOther "critical")+
+ src/Network/JobQueue/Types.hs view
@@ -0,0 +1,129 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, UndecidableInstances #-}+++module Network.JobQueue.Types+  ( JobActionState(..)+  , JobM+  , ActionM+  , ActionT+  , ActionFn+  , ActionEnv(..)+  , Unit(..)+  , RuntimeState(..)+  , Break(..)+  , LogLevel(..)+  , setNextJob+  , setNextJobIfEmpty+  , emptyNextJob+  , addForkJob+  , incrementCommits+  , getCommits+  , runS+  , runAM+  , addAction+  , setResult+  ) where++import Data.Time.Clock++import Control.Applicative+import Control.Monad.Base+import Control.Monad.Trans.Control+import Control.Monad.Except+import Control.Monad.Reader+import Control.Monad.State+import Control.Monad.Logger+import Control.Exception+import Data.Default (Default, def)++import Network.JobQueue.Class++-------------------------------- Types++data Break = Unhandled SomeException | Failure String | Retriable deriving (Show)++-------------------------------- State in Action++data RuntimeState a = RS+  { rsNextJob :: (Maybe a)+  , rsNextForks :: [(a, Maybe UTCTime)]+  , rsCommits :: Int+  } deriving (Show)++instance (Unit a) => Default (RuntimeState a) where+  def = RS Nothing [] 0++setNextJob :: (Unit a) => a -> (RuntimeState a) -> (RuntimeState a)+setNextJob x next@(RS _ _ _) = next { rsNextJob = Just x }++setNextJobIfEmpty :: (Unit a) => a -> (RuntimeState a) -> (RuntimeState a)+setNextJobIfEmpty x next@(RS mju _ _) = maybe (next { rsNextJob = Just x }) (const next) mju++emptyNextJob :: (Unit a) => (RuntimeState a) -> (RuntimeState a)+emptyNextJob next@(RS _ _ _) = next { rsNextJob = Nothing }++addForkJob :: (Unit a) => (a, Maybe UTCTime) -> (RuntimeState a) -> (RuntimeState a)+addForkJob (x, mt) next@(RS _ xs _) = next { rsNextForks = ((x, mt):xs) }++incrementCommits :: (Unit a) => (RuntimeState a) -> (RuntimeState a)+incrementCommits next@(RS _ _ cnt) = next { rsCommits = cnt + 1 }++getCommits :: (Unit a) => (RuntimeState a) -> Int+getCommits (RS _ _ cnt) = cnt++-------------------------------- JobActionState++type ActionFn e a = e -> a -> IO (Either Break (Maybe (RuntimeState a)))++data JobActionState e a = JobActionState { jobActions :: [ActionFn e a] }++addAction :: (Env e, Unit a) => ActionFn e a -> JobActionState e a -> JobActionState e a+addAction action s@(JobActionState { jobActions = actions }) = s { jobActions = action:actions }++instance Default (JobActionState e a) where+  def = JobActionState []++-------------------------------- ActionM++newtype (Env e, Unit a) => JobM e a b = JobM { runS :: StateT (JobActionState e a) IO b }+  deriving (Monad, MonadIO, Functor, MonadState (JobActionState e a))++data ActionEnv e a = ActionEnv+  { getJobEnv :: e+  , getJobUnit :: a+  }++newtype ActionT e a m b = ActionT+  { runAM :: ExceptT Break (ReaderT (ActionEnv e a) (StateT (Maybe (RuntimeState a)) (LoggingT m))) b+  } deriving ( Applicative, Functor, Monad, MonadIO, MonadLogger, MonadError Break+             , MonadReader (ActionEnv e a), MonadState (Maybe (RuntimeState a)), MonadBase base)++type ActionM e a b = ActionT e a IO b++instance MonadTrans (ActionT e a) where+  lift = ActionT . lift . lift . lift . lift++instance MonadTransControl (ActionT e a) where+  newtype StT (ActionT e a) b = StAction { unStAction :: (Either Break b, Maybe (RuntimeState a)) }+  restoreT = ActionT . ExceptT . ReaderT . const . StateT . const+           . LoggingT . const . liftM unStAction+  liftWith f = ActionT . ExceptT . ReaderT $ \r -> StateT $ \s -> LoggingT $ \l ->+    liftM (\x -> (Right x, s))+          (f $ \t -> liftM StAction (runLoggingT (runStateT (runReaderT (runExceptT (runAM t)) r) s) l))++instance MonadBaseControl base m => MonadBaseControl base (ActionT e a m) where+  newtype StM (ActionT e a m) b = StMActionT { unStMActionT :: ComposeSt (ActionT e a) m b }+  liftBaseWith = defaultLiftBaseWith StMActionT+  restoreM = defaultRestoreM unStMActionT++setResult :: (Unit a) => Maybe (RuntimeState a) -> Maybe (RuntimeState a) -> Maybe (RuntimeState a)+setResult result _ = result+++--------------------------------
+ test/Main.hs view
@@ -0,0 +1,22 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++module Main where++import Test.Hspec+import System.Environment (lookupEnv)+import Data.Maybe+import System.IO++import Action+import BackendQueue+import JobQueue++main :: IO ()+main = do+  hSetBuffering stderr LineBuffering+  backend <- fmap (fromMaybe "sqlite3://test.sqlite3") $ lookupEnv "JOBQUEUE_TEST_BACKEND"+  hspec $ do+    testAction backend+    testJobQueueBackend backend+    testJobQueue backend