faktory 1.0.1.1 → 1.0.1.2

raw patch · 5 files changed

+58/−18 lines, 5 filesPVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +11/−2
• faktory.cabal +2/−2
• library/Faktory/Client.hs +12/−14
• library/Faktory/Prelude.hs +3/−0
• tests/FaktorySpec.hs +30/−0

@@ -1,8 +1,17 @@-## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.0...master)+## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.1...master)  None -## [v1.0.1.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.0.0...1.0.1.0)+## [v1.0.1.2](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.1...v1.0.1.2)++- Fix internal handling of invalid Server Replies++## [v1.0.1.1](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.1.0...v1.0.1.1)++- Include non-OK reply in `commandOK` error+- Build with GHC-8.8++## [v1.0.1.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.0.0.0...v1.0.1.0)  - Upgrade to `megaparsec-7` 

@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: fe59297ff90aa80416c1575f9e03019036b581f05a2ebccc851f13748e618b5e+-- hash: 87c1de8469f9bffe9c0e5c342f2944a18b4725f0c07a15b3ff4bbcf88f056313  name:           faktory-version:        1.0.1.1+version:        1.0.1.2 synopsis:       Faktory Worker for Haskell description:    Haskell client and worker process for the Faktory background job server.                 .

@@ -20,6 +20,7 @@ import Control.Concurrent.MVar import Crypto.Hash (Digest, SHA256(..), hashWith) import Data.Aeson+import Data.Bitraversable (bimapM) import Data.ByteArray (ByteArrayAccess) import Data.ByteString.Lazy (ByteString, fromStrict) import qualified Data.ByteString.Lazy.Char8 as BSL8@@ -83,7 +84,9 @@     client <- Client <$> newMVar conn <*> pure settings      greeting <--      fromJustThrows "Unexpected end of HI message" =<< recvUnsafe settings conn+      fromJustThrows "Unexpected end of HI message"+      =<< fromRightThrows+      =<< recvUnsafe settings conn     stripped <-       fromJustThrows ("Missing HI prefix: " <> show greeting)         $ BSL8.stripPrefix "HI" greeting@@ -134,14 +137,14 @@ command_ :: Client -> ByteString -> [ByteString] -> IO () command_ Client {..} cmd args = withMVar clientConnection $ \conn -> do   sendUnsafe clientSettings conn cmd args-  void $ recvUnsafe clientSettings conn+  void $ fromRightThrows =<< recvUnsafe clientSettings conn  -- | Send a command, assert the response is @OK@ commandOK :: HasCallStack => Client -> ByteString -> [ByteString] -> IO () commandOK Client {..} cmd args = withMVar clientConnection $ \conn -> do   sendUnsafe clientSettings conn cmd args   response <- recvUnsafe clientSettings conn-  unless (response == Just "OK")+  unless (response == Right (Just "OK"))     $ throwString     $ "Server not OK. Reply was: "     <> show response@@ -155,8 +158,8 @@   -> IO (Either String (Maybe a)) commandJSON Client {..} cmd args = withMVar clientConnection $ \conn -> do   sendUnsafe clientSettings conn cmd args-  mByteString <- recvUnsafe clientSettings conn-  pure $ traverse eitherDecode mByteString+  emByteString <- recvUnsafe clientSettings conn+  either (pure . Left) (pure . traverse eitherDecode) emByteString  -- | Send a command to the Server socket --@@ -172,16 +175,11 @@ -- -- Do not use outside of @'withMVar'@, this is not threadsafe. ---recvUnsafe :: Settings -> Connection -> IO (Maybe ByteString)+recvUnsafe :: Settings -> Connection -> IO (Either String (Maybe ByteString)) recvUnsafe Settings {..} conn = do-  eByteString <- readReply $ connectionGet conn 4096-  settingsLogDebug $ "< " <> show eByteString--  case eByteString of-    Left err -> do-      settingsLogError err-      pure Nothing-    Right mByteString -> pure $ fromStrict <$> mByteString+  emByteString <- readReply $ connectionGet conn 4096+  settingsLogDebug $ "< " <> show emByteString+  bimapM pure (pure . fmap fromStrict) emByteString  -- | Iteratively apply a function @n@ times --

@@ -20,3 +20,6 @@ forkIOWithThrowToParent action = do   parent <- myThreadId   forkIO $ action `X.catchAny` \err -> throwTo parent err++fromRightThrows :: MonadThrow m => Either String a -> m a+fromRightThrows = either throwString pure

@@ -4,6 +4,7 @@  import Faktory.Prelude +import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.MVar import Faktory.Client import Faktory.Job@@ -44,3 +45,32 @@      jobs <- readMVar processedJobs     jobs `shouldMatchList` ["a", "b", "HALT"]++  it "correctly handles fetch timeouts" $ do+    settings' <- envSettings+    let settings = settings' { settingsWorkerIdleDelay = 0 }++    -- start a background thread that waits for longer than the fetch timeout,+    -- then stops the worker.+    --+    -- https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#fetching-jobs+    --+    -- This ensures that the worker loop experiences recieving a Nothing from+    -- the Server and handles it correctly. Setting our own idle delay to 0+    -- ensures that we'll pick up the following HALT message immediately.+    --+    void+      $ forkIO+      $ bracket (newClient settings Nothing) closeClient+      $ \client -> do+          void $ flush client+          threadDelay $ 2 * 1000000 + 250000+          void $ perform @Text mempty client "HALT"++    processedJobs <- newMVar ([] :: [Text])+    runWorker settings $ \job -> do+      modifyMVar_ processedJobs $ pure . (job :)+      when (job == "HALT") $ throw WorkerHalt++    jobs <- readMVar processedJobs+    jobs `shouldMatchList` ["HALT"]