natskell 0.0.0.1 → 0.1.0.0

raw patch · 10 files changed

+208/−20 lines, 10 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• client/API.hs +25/−4
• client/Client.hs +10/−7
• internal/Plumbing/Parser/Attoparsec.hs +53/−2
• internal/Policy/Publish.hs +1/−1
• internal/Policy/Publish/Config.hs +2/−2
• internal/Policy/Subscription/Types.hs +4/−1
• natskell.cabal +1/−2
• test/Integration/ClientSpec.hs +64/−1
• test/Unit/ParserSpec.hs +40/−0
• test/Unit/SubSpec.hs +8/−0

@@ -6,8 +6,10 @@   , PublishOption   , SubscribeOption   , withSubscriptionExpiry+  , withQueueGroup   , withPayload   , withReplyCallback+  , withReplyTo   , withHeaders   ) where @@ -29,6 +31,8 @@                   -- ^ Subscribe with request semantics and auto-unsubscribe after a reply.                 , unsubscribe :: SID -> IO ()                   -- ^ Unsubscribe from a subscription by SID.+                , newInbox :: IO Subject+                  -- ^ Create a unique inbox subject for replies.                 , ping :: IO () -> IO ()                   -- ^ Send a ping and run the callback when a pong arrives.                 , flush :: IO ()@@ -71,6 +75,11 @@ withSubscriptionExpiry :: NominalDiffTime -> SubscribeOption withSubscriptionExpiry expirySeconds cfg = cfg { expiry = Just expirySeconds } +-- | withQueueGroup sets the queue group for a subscription.+-- Default: no queue group.+withQueueGroup :: Subject -> SubscribeOption+withQueueGroup queueGroup cfg = cfg { subscribeQueueGroup = Just queueGroup }+ -- | withPayload is used to set the payload for a publish operation. -- Default: no payload. --@@ -82,7 +91,8 @@ -- publish client \"updates\" [withPayload \"hello\"] -- @ withPayload :: Payload -> PublishOption-withPayload payload (_, callback, headers) = (Just payload, callback, headers)+withPayload payload (_, callback, headers, replyTo') =+  (Just payload, callback, headers, replyTo')  -- | withReplyCallback is used to set a callback for a reply to a publish operation. -- Default: no reply subscription; publishes are fire-and-forget.@@ -95,15 +105,25 @@ -- publish client \"service.echo\" [withReplyCallback print] -- @ withReplyCallback :: (Maybe MsgView -> IO ()) -> PublishOption-withReplyCallback callback (payload, _, headers) =+withReplyCallback callback (payload, _, headers, replyTo') =   (payload, Just (callback . fmap (\msg -> MsgView     { subject = Msg.subject msg     , sid = Msg.sid msg     , replyTo = Msg.replyTo msg     , payload = Msg.payload msg     , headers = Msg.headers msg-    })), headers)+    })), headers, replyTo') +-- | withReplyTo sets an explicit reply subject for a publish operation.+-- Default: no reply subject unless a reply callback is configured.+--+-- This option only controls the outgoing reply subject. It does not create a+-- subscription; callers that expect multiple replies should subscribe to the+-- reply subject themselves.+withReplyTo :: Subject -> PublishOption+withReplyTo replySubject (payload, callback, headers, _) =+  (payload, callback, headers, Just replySubject)+ -- | withHeaders is used to set headers for a publish operation. -- Default: no headers. --@@ -115,4 +135,5 @@ -- publish client \"updates\" [withHeaders [(\"source\", \"test\")]] -- @ withHeaders :: Headers -> PublishOption-withHeaders headers (payload, callback, _) = (payload, callback, Just headers)+withHeaders headers (payload, callback, _, replyTo') =+  (payload, callback, Just headers, replyTo')

@@ -101,7 +101,7 @@     , unregister     ) import           Subscription.Types-    ( SubscribeConfig (SubscribeConfig)+    ( SubscribeConfig (..)     , SubscriptionMeta (SubscriptionMeta)     ) import qualified Types.Connect            as Connect@@ -201,6 +201,7 @@         let cfg = applyCallOptions subscribeOptions defaultSubscribeConfig         subscribeClient clientState store True subject cfg (toInternalCallback callback)     , unsubscribe = unsubscribeClient clientState store+    , newInbox = nextInbox clientState     , ping = pingClient clientState     , flush = flushClient clientState     , reset = resetClient connectionApi clientState store@@ -289,7 +290,7 @@     }  defaultSubscribeConfig :: SubscribeConfig-defaultSubscribeConfig = SubscribeConfig Nothing+defaultSubscribeConfig = SubscribeConfig Nothing Nothing  selectAuth :: ClientAuth -> Auth selectAuth authSelection =@@ -354,14 +355,14 @@     }  publishClient :: ClientState -> SubscriptionStore -> Msg.Subject -> PublishConfig -> IO ()-publishClient client store subject (payload, callback, headers) = do+publishClient client store subject (payload, callback, headers, configuredReplyTo) = do   runClient client $     logMessage Debug ("publishing to subject: " ++ show subject)   replyTo <- case callback of     Nothing ->-      pure Nothing+      pure configuredReplyTo     Just replyCallback -> do-      inbox <- nextInbox client+      inbox <- maybe (nextInbox client) pure configuredReplyTo       sid <- nextSid client       let meta =             SubscriptionMeta inbox Nothing True@@ -394,14 +395,16 @@   runClient client $     logMessage Debug ("subscribing to subject: " ++ show subject)   sid <- nextSid client+  let queueGroup =+        subscribeQueueGroup cfg   let meta =-        SubscriptionMeta subject Nothing isReply+        SubscriptionMeta subject queueGroup isReply   register store sid meta cfg callback   enqueue client $     QueueItem       Sub.Sub         { Sub.subject = subject-        , Sub.queueGroup = Nothing+        , Sub.queueGroup = queueGroup         , Sub.sid = sid         }   when isReply $

@@ -183,9 +183,47 @@ headerBlockParser :: A.Parser [(BS.ByteString, BS.ByteString)] headerBlockParser = do   _ <- A.string "NATS/"-  _ <- lineParser-  headerPairsParser []+  prelude <- lineParser+  statusHeaders <- inlineStatusHeaders prelude+  headerPairs <- headerPairsParser []+  pure (statusHeaders ++ headerPairs) +inlineStatusHeaders :: BS.ByteString -> A.Parser [(BS.ByteString, BS.ByteString)]+inlineStatusHeaders prelude =+  case BS.dropWhile isHorizontalSpaceByte rest of+    statusBytes+      | BS.null statusBytes ->+          pure []+      | otherwise ->+          parseInlineStatus statusBytes+  where+    (_, rest) = BS.break isHorizontalSpaceByte prelude++parseInlineStatus :: BS.ByteString -> A.Parser [(BS.ByteString, BS.ByteString)]+parseInlineStatus bytes+  | BS.length bytes < inlineStatusLength =+      fail ("invalid HMSG status: " ++ B8.unpack bytes)+  | not (isInlineStatusCode statusCode) =+      fail ("invalid HMSG status: " ++ B8.unpack statusCode)+  | BS.null remainder =+      pure [("Status", statusCode)]+  | isHorizontalSpaceByte (BS.head remainder) =+      pure (("Status", statusCode) : descriptionHeader)+  | otherwise =+      fail ("invalid HMSG status: " ++ B8.unpack bytes)+  where+    (statusCode, remainder) = BS.splitAt inlineStatusLength bytes+    description = stripHorizontalSpace remainder+    descriptionHeader+      | BS.null description =+          []+      | otherwise =+          [("Description", description)]++isInlineStatusCode :: BS.ByteString -> Bool+isInlineStatusCode bytes =+  BS.length bytes == inlineStatusLength && BS.all isDigitByte bytes+ headerPairsParser :: [(BS.ByteString, BS.ByteString)] -> A.Parser [(BS.ByteString, BS.ByteString)] headerPairsParser reversedPairs = do   nextByte <- A.peekWord8'@@ -276,6 +314,13 @@ isHorizontalSpaceByte word8Value =   word8Value == space || word8Value == horizontalTab +isDigitByte :: Word8 -> Bool+isDigitByte word8Value =+  word8Value >= zero && word8Value <= nine++inlineStatusLength :: Int+inlineStatusLength = 3+ carriageReturn :: Word8 carriageReturn = 13 @@ -290,3 +335,9 @@  horizontalTab :: Word8 horizontalTab = 9++zero :: Word8+zero = 48++nine :: Word8+nine = 57

@@ -5,4 +5,4 @@ import           Publish.Config (PublishConfig)  defaultPublishConfig :: PublishConfig-defaultPublishConfig = (Nothing, Nothing, Nothing)+defaultPublishConfig = (Nothing, Nothing, Nothing, Nothing)

@@ -5,6 +5,6 @@   ) where  import qualified Types.Msg as M-import           Types.Msg (Headers, Payload)+import           Types.Msg (Headers, Payload, Subject) -type PublishConfig = (Maybe Payload, Maybe (Maybe M.Msg -> IO ()), Maybe Headers)+type PublishConfig = (Maybe Payload, Maybe (Maybe M.Msg -> IO ()), Maybe Headers, Maybe Subject)

@@ -6,7 +6,10 @@ import           Data.Time.Clock (NominalDiffTime) import           Types.Msg       (Subject) -newtype SubscribeConfig = SubscribeConfig { expiry :: Maybe NominalDiffTime }+data SubscribeConfig = SubscribeConfig+                         { expiry              :: Maybe NominalDiffTime+                         , subscribeQueueGroup :: Maybe Subject+                         }  data SubscriptionMeta = SubscriptionMeta                           { subject    :: Subject

@@ -1,6 +1,6 @@ cabal-version:          3.0 name:                   natskell-version:                0.0.0.1+version:                0.1.0.0 synopsis:               A NATS client library written in Haskell tested-with:   GHC ==8.8,@@ -50,7 +50,6 @@     natskell-internal,     stm  >= 2.5 && < 2.6,     bytestring  >= 0.10 && < 0.13,-    heap >= 0.6 && < 1.1,     time >= 1.9 && < 1.16    exposed-modules:

@@ -5,6 +5,9 @@ import           API     ( Client (..)     , MsgView (..)+    , withPayload+    , withQueueGroup+    , withReplyTo     , withSubscriptionExpiry     ) import           Client@@ -24,7 +27,7 @@ import           Data.Word8 import           Network.Socket            (Socket, accept, listen) import qualified Network.Socket-import           Network.Socket.ByteString (sendAll)+import           Network.Socket.ByteString (recv, sendAll) import           Network.Socket.Free import           System.Timeout            (timeout) import           Test.Hspec@@ -74,6 +77,33 @@     , "\r\n"     ] +recvUntil :: Socket -> (BS.ByteString -> Bool) -> IO BS.ByteString+recvUntil sock done =+  go BS.empty+  where+    go acc+      | done acc = pure acc+      | otherwise = do+          chunk <- recv sock 4096+          if BS.null chunk+            then pure acc+            else go (acc <> chunk)++expectClientCommand :: Socket -> BS.ByteString -> IO ()+expectClientCommand sock command = do+  result <- timeout 1000000 $+    recvUntil sock (BS.isInfixOf command)+  case result of+    Nothing ->+      expectationFailure ("client did not send command: " ++ show command)+    Just bytes ->+      unless (BS.isInfixOf command bytes) $+        expectationFailure ("unexpected client bytes: " ++ show bytes)++expectQueuedSub :: Socket -> BS.ByteString -> BS.ByteString -> BS.ByteString -> IO ()+expectQueuedSub sock subject queueGroup sid =+  expectClientCommand sock (BS.concat ["SUB ", subject, " ", queueGroup, " ", sid, "\r\n"])+ startClientWith extraOptions = do   (p, sock) <- openFreePort   listen sock 1@@ -145,6 +175,12 @@         case result of           Nothing -> expectationFailure "flush did not return after PONG"           Just () -> pure ()+      it "subscribes with a queue group" $ \(serv, client, _, _) -> do+        sid <- subscribe client "JOBS" [withQueueGroup "WORKERS"] (const (pure ()))+        expectQueuedSub serv "JOBS" "WORKERS" sid+      it "publishes with an explicit reply subject" $ \(serv, client, _, _) -> do+        publish client "REQUESTS" [withPayload "hello", withReplyTo "_INBOX.custom"]+        expectClientCommand serv "PUB REQUESTS _INBOX.custom 5\r\nhello\r\n"       it "PONG resolves one ping" $ \(serv, client, _, _) -> do         first <- newEmptyMVar         second <- newEmptyMVar@@ -299,6 +335,33 @@           ping client $ done wg           sendAll secondConn "PONG\r\n"           wait wg+          close client+          Network.Socket.close secondConn+      Network.Socket.close sock+    it "resubscribes with a queue group after reconnect" $ do+      (p, sock) <- openFreePort+      listen sock 2+      clientVar <- newEmptyMVar+      void . forkIO $ do+        let configOptions =+              [ withMinimumLogLevel Debug+              , withConnectionAttempts 2+              , withConnectName "test-client"+              ]+        c <- newClient [("127.0.0.1", p)] configOptions+        putMVar clientVar c+      (firstConn, _) <- accept sock+      sendAll firstConn defaultINFO+      clientResult <- timeout 1000000 (takeMVar clientVar)+      case clientResult of+        Nothing -> expectationFailure "client did not connect"+        Just client -> do+          sid <- subscribe client "JOBS" [withQueueGroup "WORKERS"] (const (pure ()))+          expectQueuedSub firstConn "JOBS" "WORKERS" sid+          Network.Socket.close firstConn+          (secondConn, _) <- accept sock+          sendAll secondConn defaultINFO+          expectQueuedSub secondConn "JOBS" "WORKERS" sid           close client           Network.Socket.close secondConn       Network.Socket.close sock

@@ -58,3 +58,43 @@               expectationFailure ("unexpected parse result: " ++ show other)         other ->           expectationFailure ("unexpected parse result: " ++ show other)++    it "exposes inline HMSG status and description as headers" $ do+      let input =+            "HMSG FOO 13 28 28\r\nNATS/1.0 404 No Messages\r\n\r\n\r\n"+      case parse parserApi input of+        Emit parsed rest -> do+          rest `shouldBe` ""+          case parsed of+            ParsedMsg msg' -> do+              Msg.headers msg' `shouldBe` Just [("Status", "404"), ("Description", "No Messages")]+              Msg.payload msg' `shouldBe` Just ""+            other ->+              expectationFailure ("unexpected parse result: " ++ show other)+        other ->+          expectationFailure ("unexpected parse result: " ++ show other)++    it "exposes inline HMSG status without a description" $ do+      let input =+            "HMSG FOO 13 16 16\r\nNATS/1.0 404\r\n\r\n\r\n"+      case parse parserApi input of+        Emit parsed rest -> do+          rest `shouldBe` ""+          case parsed of+            ParsedMsg msg' -> do+              Msg.headers msg' `shouldBe` Just [("Status", "404")]+              Msg.payload msg' `shouldBe` Just ""+            other ->+              expectationFailure ("unexpected parse result: " ++ show other)+        other ->+          expectationFailure ("unexpected parse result: " ++ show other)++    it "rejects malformed inline HMSG statuses" $ do+      let input =+            "HMSG FOO 13 17 17\r\nNATS/1.0 404x\r\n\r\n\r\n"+      case parse parserApi input of+        DropPrefix n reason -> do+          n `shouldSatisfy` (> 0)+          reason `shouldContain` "invalid HMSG status"+        other ->+          expectationFailure ("unexpected parse result: " ++ show other)

@@ -5,6 +5,7 @@ import           Control.Monad import           Data.ByteString import qualified Data.ByteString.Lazy      as LBS+import           Subscription.Types import           Test.Hspec import           Text.Printf import           Transformers.Transformers@@ -13,8 +14,15 @@  spec :: Spec spec = do+  subscribeOptionCases   transformerCases   validateCases++subscribeOptionCases = parallel $ do+  describe "subscribe config" $ do+    it "stores the queue group" $ do+      subscribeQueueGroup (SubscribeConfig Nothing (Just "WORKERS"))+        `shouldBe` Just "WORKERS"  explicitTransformerCases :: [(Sub, ByteString)] explicitTransformerCases = [