jobqueue 0.1.4 → 0.1.5

raw patch · 9 files changed

+363/−25 lines, 9 filesdep +aesonnew-uploader

Dependencies added: aeson

Files

• jobqueue.cabal +6/−1
• src/Network/JobQueue.hs +2/−0
• src/Network/JobQueue/Action.hs +0/−17
• src/Network/JobQueue/Class.hs +0/−6
• src/Network/JobQueue/Param.hs +71/−0
• src/Network/JobQueue/Types.hs +16/−1
• test/Action.hs +91/−0
• test/BackendQueue.hs +63/−0
• test/JobQueue.hs +114/−0

@@ -1,5 +1,5 @@ Name:           jobqueue-Version:        0.1.4+Version:        0.1.5 Synopsis:       A job queue library License:        MIT License-File:   LICENSE@@ -46,6 +46,7 @@                  , transformers-base                  , lifted-base                  , regex-posix+                 , aeson   Hs-source-dirs:  src   Exposed-modules: Network.JobQueue                  , Network.JobQueue.Class@@ -63,6 +64,7 @@                  , Network.JobQueue.Backend.Sqlite3                  , Network.JobQueue.Logger                  , Network.JobQueue.Util+                 , Network.JobQueue.Param   Other-modules:   Network.JobQueue.Backend.Zookeeper.ZookeeperQueue   Extensions:      DeriveDataTypeable @@ -80,5 +82,8 @@                  , stm   Type:            exitcode-stdio-1.0   Hs-source-dirs:  test+  Other-modules:   Action+                 , BackendQueue+                 , JobQueue   Main-is:         Main.hs 

@@ -103,6 +103,7 @@   , forkOnTime   , abort   , getEnv+  , Param   , param   , commitIO   , liftIO@@ -122,6 +123,7 @@ import Network.JobQueue.JobQueue import Network.JobQueue.Job import Network.JobQueue.Logger+import Network.JobQueue.Param (Param,param)  {- | Build a function that takes a function (('JobQueue' a -> 'IO' ()) -> IO ()) as its first parameter. 

@@ -9,7 +9,6 @@   , runActionState   , runAction   , getEnv-  , param   , next   , orNext   , fin@@ -86,22 +85,6 @@ -} 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        ---------------- 

@@ -9,12 +9,6 @@ -} 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

@@ -0,0 +1,71 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, UndecidableInstances #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE OverloadedStrings #-}+++module Network.JobQueue.Param+  ( ParamEnv+  , envParameters+  , Param+  , decodeParam+  , encodeParam+  , param+  ) where++import Data.Maybe+import qualified Data.Aeson as A+import qualified Data.ByteString.Lazy.Char8 as B+import Network.JobQueue.Class+import Network.JobQueue.Types+import Network.JobQueue.Action (getEnv,abort)+import Network.JobQueue.Logger++{- | Environment with a parameter set+-}+class (Env a) => ParamEnv a where+  envParameters :: a -> [(String, String)]+  envParameters _env = []++class Param a where+  decodeParam :: String -> Maybe a+  encodeParam :: a -> String++instance Param String where+  decodeParam str = (fmap fst . listToMaybe . reads) str+  encodeParam val = show val++instance Param Int where+  decodeParam str = (fmap fst . listToMaybe . reads) str+  encodeParam val = show val++instance Param Integer where+  decodeParam str = (fmap fst . listToMaybe . reads) str+  encodeParam val = show val++instance Param Double where+  decodeParam str = (fmap fst . listToMaybe . reads) str+  encodeParam val = show val++instance Param A.Value where+  decodeParam str = A.decode (B.pack str)+  encodeParam val = B.unpack $ A.encode val++{- | Get a parameter value with a key from the environment in action.+     This is a special function for ParamEnv.+-}+param :: (ParamEnv e, Unit a, Param b) => (String, String) -> ActionM e a b+param (key, defaultValue) = do+  env <- getEnv+  case decodeParam 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' (decodeParam value))+      Nothing -> return (defaultValue')

@@ -5,7 +5,7 @@ {-# LANGUAGE TypeSynonymInstances #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE TypeFamilies, MultiParamTypeClasses, UndecidableInstances #-}-+{-# LANGUAGE CPP #-}  module Network.JobQueue.Types   ( JobActionState(..)@@ -110,17 +110,32 @@   lift = ActionT . lift . lift . lift . lift  instance MonadTransControl (ActionT e a) where+#if MIN_VERSION_monad_control(1,0,0)+  type StT (ActionT e a) b = (Either Break b, Maybe (RuntimeState a))+  restoreT = ActionT . ExceptT . ReaderT . const . StateT . const+           . LoggingT . const+  liftWith f = ActionT . ExceptT . ReaderT $ \r -> StateT $ \s -> LoggingT $ \l ->+    liftM (\x -> (Right x, s))+          (f $ \t -> (runLoggingT (runStateT (runReaderT (runExceptT (runAM t)) r) s) l))+#else   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))+#endif  instance MonadBaseControl base m => MonadBaseControl base (ActionT e a m) where+#if MIN_VERSION_monad_control(1,0,0)+  type StM (ActionT e a m) b = ComposeSt (ActionT e a) m b+  liftBaseWith = defaultLiftBaseWith+  restoreM = defaultRestoreM+#else   newtype StM (ActionT e a m) b = StMActionT { unStMActionT :: ComposeSt (ActionT e a) m b }   liftBaseWith = defaultLiftBaseWith StMActionT   restoreM = defaultRestoreM unStMActionT+#endif  setResult :: (Unit a) => Maybe (RuntimeState a) -> Maybe (RuntimeState a) -> Maybe (RuntimeState a) setResult result _ = result

@@ -0,0 +1,91 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE OverloadedStrings #-}++module Action (testAction) where++import Control.Exception+import Test.Hspec+import Control.Monad+import System.Directory+import System.IO.Error (isDoesNotExistError)+import Control.Concurrent+import Control.Concurrent.Async+import System.IO+import Control.Concurrent.STM++import Network.JobQueue++data JobEnv = JobEnv {+    jeHello      :: String+  } deriving (Eq, Show)++instance Env JobEnv where++instance Aux JobEnv where+  auxHandleFailure _ mjob = do+    case mjob of+      Just job -> do+        nextJob <- createJob Runnable (getRecovery (jobUnit job))+        return (Just nextJob)+      Nothing -> return (Nothing)++data JobUnit = Initial | Recovery deriving (Show, Read, Eq, Ord)++instance Unit JobUnit where+  getPriority _ju = 1+  getRecovery _ju = Recovery+  toBeLogged _ = False++instance Desc JobUnit where++newMarker val = do+  var <- liftIO $ newTVarIO val+  return (liftIO $ readTVarIO var, liftIO . atomically . (writeTVar var))++testAction :: String -> Spec+testAction backend = do+  describe "action" $ do+    it "unhandled error" $ do+      (get, set) <- newMarker 0+      go $ buildJobQueue backend "/unhandled_error_1" $ do+            process $ \Initial -> do+              set 1+              $(logWarn) "Throw an IOError." ()+              liftIO $ throwIO $ userError "an IOError"+              $(logError) "Never reach here." ()+              set 2+              fin+            process $ \Recovery -> do+              set 3+              fin+      get `shouldReturn` 1++    it "abort and recover" $ do+      (get, set) <- newMarker 0+      go $ buildJobQueue backend "/abort_and_recover_1" $ do+            process $ \Initial -> do+              set 1+              $(logWarn) "Abort" ()+              abort+              $(logError) "Never reach here." ()+              set 2+              fin+            process $ \Recovery -> do+              set 3+              fin+      get `shouldReturn` 3++  where+    go withJobQueue = withJobQueue $ \jq -> do+      scheduleJob jq Initial+      countJobQueue jq `shouldReturn` 1+      let loop = \env jq' -> do+            executeJob jq' env+            count <- countJobQueue jq'+            when (count > 0) $ loop env jq'+      loop (JobEnv "hello") jq+      countJobQueue jq `shouldReturn` 0+

@@ -0,0 +1,63 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE TemplateHaskell #-}++module BackendQueue (testJobQueueBackend) where++import Control.Exception+import Test.Hspec+import System.Directory+import System.IO.Error (isDoesNotExistError)+import System.Environment (lookupEnv)++import qualified Data.ByteString.Char8 as BS+import Network.JobQueue.Backend+import Network.JobQueue.Backend.Types+import Network.JobQueue.Backend.Class++testJobQueueBackend :: String -> Spec+testJobQueueBackend backend = do+  describe "backend queue" $ do+    it "peeks" $ do+      withBackend backend $ \(Backend { bOpenQueue = openQueue }) -> do+        q <- openQueue "/case/peek_1"+        k <- writeQueue q (BS.pack "hoge") 0+        Just (bs, name, idName, version) <- peekQueue q+        _ <- deleteQueue q name+        countQueue q `shouldReturn` 0++    it "writes and reads" $ do+      withBackend backend $ \(Backend { bOpenQueue = openQueue }) -> do+        q <- openQueue "/case/read_and_write_1"+        k <- writeQueue q (BS.pack "hoge") 0+        readQueue q `shouldReturn` Just (BS.pack "hoge", k)++    it "counts" $ do+      withBackend backend $ \(Backend { bOpenQueue = openQueue }) -> do+        q <- openQueue "/case/count_1"+        _ <- writeQueue q (BS.pack "hoge1") 0+        countQueue q `shouldReturn` 1+        _ <- writeQueue q (BS.pack "hoge2") 0+        countQueue q `shouldReturn` 2+        _ <- readQueue q+        _ <- readQueue q+        return ()++    it "has items" $ do+      withBackend backend $ \(Backend { bOpenQueue = openQueue }) -> do+        q <- openQueue "/case/items_1"+        k1 <- writeQueue q (BS.pack "hoge1") 0+        itemsQueue q `shouldReturn` [k1]+        k2 <- writeQueue q (BS.pack "hoge2") 0+        itemsQueue q `shouldReturn` [k1, k2]+        _ <- readQueue q+        _ <- readQueue q+        return ()++---------------------------------------------------------------- Utils++withBackend :: String -> (Backend -> IO ()) -> IO ()+withBackend backend act = do+  bracket (openBackend backend) bClose act+

@@ -0,0 +1,114 @@+-- Copyright (c) Gree, Inc. 2013+-- License: MIT-style++{-# LANGUAGE TemplateHaskell #-}++module JobQueue (testJobQueue) where++import Control.Exception+import Test.Hspec+import Control.Monad+import System.Directory+import System.IO.Error (isDoesNotExistError)+import Control.Concurrent+import Control.Concurrent.Async+import System.IO++import Network.JobQueue++data JobEnv = JobEnv {+    jeHello      :: String+  } deriving (Eq, Show)++instance Env JobEnv where+instance Aux JobEnv where++data JobUnit = HelloStep | WorldStep deriving (Show, Read, Eq, Ord)++instance Unit JobUnit where+  getPriority _ju = 1+  getRecovery _ju = HelloStep++instance Desc JobUnit where++data Looping = Looping Int deriving (Show, Read, Eq, Ord)++instance Unit Looping where+  getPriority _ju = 1+  getRecovery _ju = (Looping 0)+  toBeLogged _ = False++instance Desc Looping where++testJobQueue :: String -> Spec+testJobQueue backend = do+  describe "job queue" $ do+    it "says hello" $ do+      let withJobQueue = buildJobQueue backend "/says_hello_1" $ do+            process $ \WorldStep -> commitIO (putStrLn "world") >> fin+            process $ \HelloStep -> do+              env <- getEnv+              commitIO (putStr $ (jeHello env) ++ ", ")+              next WorldStep+      withJobQueue $ \jq -> do+        scheduleJob jq HelloStep+        countJobQueue jq `shouldReturn` 1+      withJobQueue $ \jq -> do+        let loop = \env jq' -> do+              executeJob jq' env+              count <- countJobQueue jq'+              when (count > 0) $ loop env jq'+        loop (JobEnv "hello") jq+        countJobQueue jq `shouldReturn` 0+    +    it "suspends" $ do+      let withJobQueue = buildJobQueue backend "/suspends_1" $ do+            process $ \WorldStep -> commitIO (putStrLn "world") >> fin+            process $ \HelloStep -> do+              env <- getEnv+              commitIO (putStr $ (jeHello env) ++ ", ")+              next WorldStep+      withJobQueue $ \jq -> do+        scheduleJob jq HelloStep+        suspendJobQueue jq `shouldReturn` True+        suspendJobQueue jq `shouldReturn` False+        step (JobEnv "hello") jq 5+        countJobQueue jq `shouldReturn` 2+      withJobQueue $ \jq -> do+        resumeJobQueue jq `shouldReturn` True+        step (JobEnv "hello") jq 5+        countJobQueue jq `shouldReturn` 0++    it "can be used concurrently" $ do+      let p = process $ \(Looping count) -> do+                if count > 0+                  then liftIO (hPutStr stderr $ " " ++ show count) >> fork (Looping (count - 1))+                  else liftIO (hPutStrLn stderr ".") >> fin+          env0 = (JobEnv "hello")+      buildJobQueue backend "/concurrently_1" p $ \jq -> do+        scheduleJob jq (Looping 100)+        countJobQueue jq `shouldReturn` 1+      bracket (openSession backend) (closeSession) $ \session -> do+        let loop = \env jq' -> do+                        executeJob jq' env+                        count <- countJobQueue jq'+                        when (count > 0) $ loop env jq'+        _ <- flip mapConcurrently [1..50] $ \_ -> do+          jq <- openJobQueue session "/concurrently_1" p+          loop env0 jq+          closeJobQueue jq+        return ()+      buildJobQueue backend "/concurrently_1" p $ \jq -> do+        executeJob jq env0+        countJobQueue jq `shouldReturn` 0+      return ()++---------------------------------------------------------------- Utils++step :: (Aux e, Env e, Unit a) => e -> JobQueue e a -> Int -> IO ()+step env jq c+  | c > 0 = do+    executeJob jq env+    step env jq (pred c)+  | otherwise = return ()+