hats (empty) → 0.1.0.0
raw patch · 26 files changed
+3179/−0 lines, 26 filesdep +HUnitdep +QuickCheckdep +aesonsetup-changed
Dependencies added: HUnit, QuickCheck, aeson, async, attoparsec, base, bytestring, conduit, conduit-extra, connection, criterion, deepseq, hats, network, network-uri, process, random, stm, test-framework, test-framework-hunit, test-framework-quickcheck2, text, transformers, unordered-containers
Files
- CHANGES.md +8/−0
- LICENSE +22/−0
- README.md +80/−0
- Setup.hs +2/−0
- bench/Bench.hs +177/−0
- example/Examples.hs +164/−0
- hats.cabal +109/−0
- src/Network/Nats.hs +268/−0
- src/Network/Nats/Api.hs +209/−0
- src/Network/Nats/Conduit.hs +78/−0
- src/Network/Nats/Connection.hs +218/−0
- src/Network/Nats/ConnectionManager.hs +179/−0
- src/Network/Nats/Dispatcher.hs +117/−0
- src/Network/Nats/JsonApi.hs +29/−0
- src/Network/Nats/Message/Message.hs +141/−0
- src/Network/Nats/Message/Parser.hs +354/−0
- src/Network/Nats/Message/Writer.hs +172/−0
- src/Network/Nats/Subscriber.hs +86/−0
- src/Network/Nats/Types.hs +117/−0
- test/CallbackTests.hs +71/−0
- test/Gnatsd.hs +55/−0
- test/JsonTests.hs +74/−0
- test/MessageProps.hs +141/−0
- test/NatsTests.hs +133/−0
- test/ReconnectionTests.hs +112/−0
- test/Spec.hs +63/−0
+ CHANGES.md view
@@ -0,0 +1,8 @@+# 0.1.0.0++* Initial release of hats - NATS client for Haskell.++ The hats library provides client access for Haskell applications+ using the NATS messaging system [https://nats.io](https://nats.io).++ A few examples can be found in [https://github.com/kosmoskatten/hats/blob/master/example/Examples.hs](https://github.com/kosmoskatten/hats/blob/master/example/Examples.hs).
+ LICENSE view
@@ -0,0 +1,22 @@+The MIT License (MIT)++Copyright (c) 2016 Patrik Sandahl++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.+
+ README.md view
@@ -0,0 +1,80 @@+HATS - Haskell NATS client+===++Haskell client for the NATS messaging system (see https://nats.io for+a general introduction to NATS).++Examples:++This section gives a simple messaging example using this library. The+example requires the presence of a NATS server, running on localhost+using the default port 4222. If other host or port, adapt the+example.++```haskell+{-# LANGUAGE OverloadedStrings #-}+module Main+ ( main+ ) where++import Network.Nats+import Text.Printf++main :: IO ()+main =+ withNats defaultSettings ["nats://localhost"] $ \nats -> do++ -- Subscribe to the topic "foo".+ (s, q) <- subscribe nats "foo" Nothing++ -- Publish to topic "foo", do not request a reply.+ publish nats "foo" Nothing "Some payload"++ -- Wait for a message, print the message's payload+ msg <- nextMsg q+ printf "Received %s\n" (show $ payload msg)++ -- Unsubscribe from topic "foo".+ unsubscribe nats s Nothing+```++Beside from the subscription mode where messages, synchronously, are+fetched from a queue there is also an asynchronous mode where each+request is handled immediately in their own thread.++```haskell+{-# LANGUAGE OverloadedStrings #-}+module Main+ ( main+ ) where++import Control.Monad+import Data.Maybe+import Network.Nats+import Text.Printf++main :: IO ()+main =+ withNats defaultSettings ["nats://localhost"] $ \nats -> do+ + -- A simple - asynchronous - help service that will answer+ -- requesters that give a reply topic with "I can help".+ s1 <- subscribeAsync nats "help" Nothing $ \msg -> do+ printf "Help service received: %s\n" (show $ payload msg)+ when (isJust $ replyTo msg) $+ publish nats (fromJust $ replyTo msg) Nothing "I can help"++ -- Subscribe to help replies.+ (s2, q) <- subscribe nats "help.reply" Nothing++ -- Request help.+ publish nats "help" (Just "help.reply") "Please ..."++ -- Wait for reply.+ msg <- nextMsg q+ printf "Received: %s\n" (show $ payload msg)++ -- Unsubscribe.+ unsubscribe nats s1 Nothing+ unsubscribe nats s2 Nothing+```
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ bench/Bench.hs view
@@ -0,0 +1,177 @@+{-# LANGUAGE OverloadedStrings #-}+module Main+ ( main+ ) where++import Control.Monad (replicateM_, void)+import Control.Concurrent.Async (async, wait)+import Control.Concurrent.STM ( STM, TVar, atomically, modifyTVar+ , newTVarIO, readTVar, retry+ )+import Criterion.Main ( Benchmark, defaultMain, bgroup, bench+ , env, nf, whnfIO+ )+import Data.Attoparsec.ByteString.Char8 (IResult (..), Result, parse)+import Data.ByteString.Lazy.Builder (lazyByteString, toLazyByteString)++import Network.Nats+import Network.Nats.Message.Message (Message (..))+import Network.Nats.Message.Parser (parseMessage)+import Network.Nats.Message.Writer (writeMessage)++import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy.Char8 as LBS++main :: IO ()+main = defaultMain suite++suite :: [Benchmark]+suite =+ [ bgroup "pub-writer"+ [ env smallPubMessages $ \xs ->+ bench "million * 48 bytes" $ nf writePubs xs++ , env mediumPubMessages $ \xs ->+ bench "million * 480 bytes" $ nf writePubs xs++ , env largePubMessages $ \xs ->+ bench "million * 4800 bytes" $ nf writePubs xs+ ]+ , bgroup "msg-parser"+ [ env smallMsgMessages $ \xs ->+ bench "million * 48 bytes" $ nf parseMsgs xs++ , env mediumMsgMessages $ \xs ->+ bench "million * 480 bytes" $ nf parseMsgs xs++ , env largeMsgMessages $ \xs ->+ bench "million * 4800 bytes" $ nf parseMsgs xs+ ]+ , bgroup "pubsub-nats"+ [ bench "million pub" $ whnfIO (pubPerf million)+ , bench "100000 pubsub/queue" $ whnfIO (pubSubPerf a100000)+ , bench "100000 pubsub/async" $ whnfIO (pubSubAsyncPerf a100000)+ ]+ ]++-- | Write a list of Pub messages to a list of lazy ByteStrings.+writePubs :: [Message] -> [LBS.ByteString]+writePubs = map writeMessage++-- | Parse a list of ByteStrings to a list of Msg messages.+parseMsgs :: [BS.ByteString] -> [Message]+parseMsgs = map (fromResult . parse parseMessage)+ where+ fromResult :: Result Message -> Message+ fromResult (Done _ msg) = msg+ fromResult (Partial cont) = fromResult (cont "")+ fromResult _ = error "Shall not happen"++-- | Send the given number of Pub messages containing the payload "hello".+-- This benchmark requires a running NATS server.+pubPerf :: Int -> IO ()+pubPerf rep =+ withNats defaultSettings [defaultURI] $ \nats ->+ replicateM_ rep $ publish nats "bench" Nothing "hello"++pubSubPerf :: Int -> IO ()+pubSubPerf rep =+ withNats defaultSettings [defaultURI] $ \nats -> do+ (_, queue) <- subscribe nats "bench" Nothing+ rec <- async $ receiver queue rep++ replicateM_ rep $ publish nats "bench" Nothing "hello"+ wait rec++receiver :: MsgQueue -> Int -> IO ()+receiver queue limit = go 0+ where+ go :: Int -> IO ()+ go cnt+ | cnt == limit = return ()+ | otherwise = do+ void $ nextMsg queue+ go (cnt + 1)++-- | Send the given number of Pub messages containing the payload "hello"+-- Subscribe to and receive - using asyncronous subscription - the +-- same number of messages.+-- This benchmark requires a running NATS server.+pubSubAsyncPerf :: Int -> IO ()+pubSubAsyncPerf rep =+ withNats defaultSettings [defaultURI] $ \nats -> do+ cnt <- newTVarIO 0+ void $ subscribeAsync nats "bench" Nothing $ asyncReceiver cnt+ replicateM_ rep $ publish nats "bench" Nothing "hello"++ atomically $ waitForValue cnt rep++asyncReceiver :: TVar Int -> Msg -> IO ()+asyncReceiver cnt _ = atomically $ modifyTVar cnt (+ 1)++waitForValue :: TVar Int -> Int -> STM ()+waitForValue tvar value = do+ value' <- readTVar tvar+ if value' /= value+ then retry+ else return ()++million :: Int+million = 1000000++a100000 :: Int+a100000 = 100000++small :: Int+small = 1++medium :: Int+medium = 10++large :: Int+large = 100++smallPubMessages :: IO [Message]+smallPubMessages = million `pubMessages` small++mediumPubMessages :: IO [Message]+mediumPubMessages = million `pubMessages` medium++largePubMessages :: IO [Message]+largePubMessages = million `pubMessages` large++smallMsgMessages :: IO [BS.ByteString]+smallMsgMessages = million `msgMessages` small++mediumMsgMessages :: IO [BS.ByteString]+mediumMsgMessages = million `msgMessages` medium++largeMsgMessages :: IO [BS.ByteString]+largeMsgMessages = million `msgMessages` large++pubMessages :: Int -> Int -> IO [Message]+pubMessages rep size = return $ replicate rep (pubMessage size)++msgMessages :: Int -> Int -> IO [BS.ByteString]+msgMessages rep size = do+ let xs = replicate rep (msgMessage size)+ return $ map (LBS.toStrict . writeMessage) xs++pubMessage :: Int -> Message+pubMessage = PUB "TOPIC.INBOX" (Just "REPLY.INBOX") . replicatePayload++msgMessage :: Int -> Message+msgMessage = + MSG "TOPIC.INBOX" 123456 (Just "REPLY.INBOX") . replicatePayload++replicatePayload :: Int -> LBS.ByteString+replicatePayload n =+ let p = map lazyByteString $ replicate n payloadChunk+ in toLazyByteString $ mconcat p++-- | A basic "random" payload chunk with 48 characters.+payloadChunk :: LBS.ByteString+payloadChunk = "pq01ow92ie83ue74ur74yt65jf82nc8emr8dj48v.dksme2z"++defaultURI :: String+defaultURI = "nats://localhost:4222"
+ example/Examples.hs view
@@ -0,0 +1,164 @@+{-# LANGUAGE OverloadedStrings #-}+-- | A set of example programs to demonstrate NATS features and the+-- API of the "Network.Nats" library.+module Main+ ( main+ ) where++import Control.Exception+import Control.Monad+import Data.Maybe+import Network.Nats+import System.Environment+import Text.Printf++main :: IO ()+main =+ natsHandler `handle` do+ demo <- getArgs+ case demo of+ ["sync-sub"] -> syncSub + ["async-sub"] -> asyncSub+ ["async-req"] -> asyncReq+ ["topic"] -> topic'+ ["queue-group"] -> queueGroup+ _ -> mapM_ putStrLn usage+ where+ -- Take care of the exceptions that can be thrown out from within+ -- 'withNats'.+ natsHandler :: NatsException -> IO ()+ natsHandler e =+ case e of+ ConnectionGiveUpException -> putStrLn "No NATS connection!"+ AuthorizationException -> putStrLn "Can't authorize!"+ URIError err -> putStrLn err+ _ -> throwIO e++-- | Simple messaging.+syncSub :: IO ()+syncSub =+ withNats defaultSettings ["nats://localhost"] $ \nats -> do++ -- Subscribe to the topic "foo".+ (s, q) <- subscribe nats "foo" Nothing++ -- Publish to topic "foo", do not request a reply.+ publish nats "foo" Nothing "Some payload"++ -- Wait for a message, print the message's payload+ msg <- nextMsg q+ printf "Received %s\n" (show $ payload msg)++ -- Unsubscribe from topic "foo".+ unsubscribe nats s Nothing++-- | Request help from a simple help service. The help service is+-- asynchronous.+asyncSub :: IO ()+asyncSub =+ withNats defaultSettings ["nats://localhost"] $ \nats -> do+ + -- A simple - asynchronous - help service that will answer+ -- requesters that give a reply topic with "I can help".+ s1 <- subscribeAsync nats "help" Nothing $ \msg -> do+ printf "Help service received: %s\n" (show $ payload msg)+ when (isJust $ replyTo msg) $+ publish nats (fromJust $ replyTo msg) Nothing "I can help"++ -- Subscribe to help replies.+ (s2, q) <- subscribe nats "help.reply" Nothing++ -- Request help.+ publish nats "help" (Just "help.reply") "Please ..."++ -- Wait for reply.+ msg <- nextMsg q+ printf "Received: %s\n" (show $ payload msg)++ -- Unsubscribe from topics.+ unsubscribe nats s1 Nothing+ unsubscribe nats s2 Nothing++-- | As 'asyncSub', but using the 'request' function to simplify.+asyncReq :: IO ()+asyncReq =+ withNats defaultSettings ["nats://localhost"] $ \nats -> do+ + -- A simple - asynchronous - help service that will answer+ -- requesters that give a reply topic with "I can help".+ s <- subscribeAsync nats "help" Nothing $ \msg -> do+ printf "Help service received: %s\n" (show $ payload msg)+ when (isJust $ replyTo msg) $+ publish nats (fromJust $ replyTo msg) Nothing "I can help"++ -- Request help.+ msg <- request nats "help" "Please ..."+ printf "Received: %s\n" (show $ payload msg)++ -- Unsubscribe.+ unsubscribe nats s Nothing++-- | Demonstration of topic strings and how they are interpreted by+-- NATS.+topic' :: IO ()+topic' =+ withNats defaultSettings ["nats://localhost"] $ \nats -> do++ -- "*" matches any token, at any level of the subject.+ (_, queue1) <- subscribe nats "foo.*.baz" Nothing+ (_, queue2) <- subscribe nats "foo.bar.*" Nothing++ -- ">" matches any length of the tail of the subject, and can+ -- only be the last token.+ (_, queue3) <- subscribe nats "foo.>" Nothing++ -- This publishing matches all the above.+ publish nats "foo.bar.baz" Nothing "Hello world"++ -- Show that the message showed up on all queues.+ forM_ [queue1, queue2, queue3] $ \queue -> do+ msg <- nextMsg queue+ printf "Received: %s\n" (show $ payload msg)++ -- The NATS server will purge the subscriptions once we+ -- have disconnected.++-- | Some fun with queue groups. Subscribers that share the same +-- queue group will be load shared by NATS, i.e. only one subscriber+-- will answer each request.+queueGroup :: IO ()+queueGroup =+ withNats defaultSettings ["nats://localhost"] $ \nats -> do++ -- Install a couple of message echo workers. All sharing the+ -- same queue group.+ void $ subscribeAsync nats "echo" (Just "workers") $ worker nats "one"+ void $ subscribeAsync nats "echo" (Just "workers") $ worker nats "two"+ void $ subscribeAsync nats "echo" (Just "workers") $ worker nats "three"+ void $ subscribeAsync nats "echo" (Just "workers") $ worker nats "four"++ -- Request some echos. There will only be one of the echo+ -- workers answering each request.+ msg1 <- request nats "echo" "E1 E1 E1"+ printf "Received: %s\n" (show $ payload msg1)+ msg2 <- request nats "echo" "E2 E2 E2"+ printf "Received: %s\n" (show $ payload msg2)+ where+ worker :: Nats -> String -> Msg -> IO ()+ worker nats name msg = do+ printf "Request handled by %s\n" name+ when (isJust $ replyTo msg) $+ publish nats (fromJust $ replyTo msg) Nothing (payload msg)++usage :: [String]+usage =+ [ "Usage: hats-examples <example>"+ , ""+ , "Examples:"+ , ""+ , "sync-sub : Demo of synchronous handling of messages."+ , "async-sub : Demo of asynchronous handling of messages."+ , "async-req : Demo of the request API."+ , "topic : Demo of topic structure."+ , "queue-group : Demo of queue group handling."+ ]
+ hats.cabal view
@@ -0,0 +1,109 @@+name: hats+version: 0.1.0.0+synopsis: Haskell client for the NATS messaging system+description:+ A Haskell client for the NATS messaging system. To get started,+ see the documentation for the "Network.Nats" module. Or see the+ example programs in the example directory.+ .+ A general introduction to NATS can be found at <https://nats.io>.+homepage: https://github.com/kosmoskatten/hats+bug-reports: https://github.com/kosmoskatten/hats/issues+license: MIT+license-file: LICENSE+author: Patrik Sandahl+maintainer: patrik.sandahl@gmail.com+copyright: 2016 Patrik Sandahl+stability: experimental+tested-with: GHC == 7.10.3, GHC == 8.0.1+category: Network+build-type: Simple+extra-source-files: README.md+ CHANGES.md+cabal-version: >=1.10++library+ hs-source-dirs: src+ exposed-modules: Network.Nats+ Network.Nats.Types+ Network.Nats.Message.Message+ Network.Nats.Message.Parser+ Network.Nats.Message.Writer++ other-modules: Network.Nats.Api+ Network.Nats.Connection+ Network.Nats.ConnectionManager+ Network.Nats.Dispatcher+ Network.Nats.JsonApi+ Network.Nats.Subscriber+ Network.Nats.Conduit+ build-depends: base >= 4.7 && < 5+ , aeson+ , async+ , attoparsec+ , bytestring+ , conduit+ , conduit-extra+ , connection+ , deepseq+ , network+ , network-uri+ , random+ , stm+ , text+ , transformers+ , unordered-containers+ ghc-options: -Wall+ default-language: Haskell2010++executable hats-examples+ hs-source-dirs: example+ main-is: Examples.hs+ build-depends: base+ , hats+ ghc-options: -Wall+ default-language: Haskell2010++test-suite hats-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ other-modules: MessageProps+ CallbackTests+ JsonTests+ NatsTests+ ReconnectionTests+ Gnatsd+ build-depends: base+ , aeson+ , attoparsec+ , bytestring+ , hats+ , HUnit+ , process+ , test-framework+ , test-framework-hunit+ , test-framework-quickcheck2+ , text+ , QuickCheck+ ghc-options: -Wall -threaded+ default-language: Haskell2010++benchmark hats-bench+ type: exitcode-stdio-1.0+ hs-source-dirs: bench+ main-is: Bench.hs+ build-depends: base+ , async+ , attoparsec+ , bytestring+ , criterion+ , hats+ , random+ , stm+ ghc-options: -Wall -threaded+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/kosmoskatten/hats
+ src/Network/Nats.hs view
@@ -0,0 +1,268 @@+-- |+-- Module: Network.Nats+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- A Haskell client for the NATS messaging system.+-- See <https://nats.io> for general information and documentation.+module Network.Nats+ (+ -- * Limitations in implementation+ -- $limitations++ -- * Simple messaging example.+ -- $simple_messaging++ -- * Ascyncronous message handling.+ -- $async_handling++ -- * Convenience API for the request pattern.+ -- $request_pattern++ -- * Topic structure.+ -- $topic_structure++ Nats+ , Msg+ , Sid+ , Payload+ , Topic+ , QueueGroup+ , MsgQueue+ , ManagerSettings (..)+ , NatsException ( URIError, ConnectionGiveUpException+ , AuthorizationException+ )+ , SockAddr+ , withNats+ , publish+ , publishJson+ , subscribe+ , subscribeAsync+ , request+ , requestJson+ , unsubscribe+ , nextMsg+ , topic+ , replyTo+ , sid+ , payload+ , jsonPayload+ , jsonPayload'+ , defaultSettings+ ) where++import Control.Exception (bracket, throwIO)+import Data.Maybe (mapMaybe)+import Network.URI (URI (..), parseAbsoluteURI)++import Network.Nats.Api ( Nats, publish, subscribe, subscribeAsync+ , request, unsubscribe, nextMsg+ , initNats, termNats+ )+import Network.Nats.ConnectionManager ( ManagerSettings (..)+ , SockAddr+ , defaultSettings+ )+import Network.Nats.JsonApi (publishJson, requestJson)+import Network.Nats.Types ( Sid, Payload , Topic, QueueGroup+ , NatsException (..), MsgQueue+ , Msg, topic, replyTo+ , sid, payload, jsonPayload, jsonPayload'+ )++-- | Run an IO action while connection towards NATS is maintained. If+-- a NATS connection is lost, the connection manager will try to+-- reconnect the same or one of the other NATS servers+-- (as specified by the provided URI strings).+-- Strategies for reconnection is specified+-- in the 'ManagerSettings'. All subscriptions will be automatically+-- replayed once a new connection is made.+withNats :: ManagerSettings+ -- ^ Settings for the connection manager. Default+ -- 'ManagerSettings' can be obtained by+ -- 'defaultSettings'.+ -> [String]+ -- ^ A list of URI strings to specify the NATS servers+ -- available. If any URI string is malformed an 'URIError'+ -- exception is thrown. Parsing of URIs is performed using+ -- the 'parseAbsoluteURI' function.+ -> (Nats -> IO a)+ -- ^ The user provided action. Once the action is terminated+ -- the connection will close.+ -> IO a+withNats settings uriStrings action =+ either (throwIO . URIError)+ (\uris -> bracket (initNats settings uris) termNats action)+ (convertURIs uriStrings)++-- | Convert a list of strings to a list of 'URI'. If there are+-- errors during the conversion an error description is returned.+convertURIs :: [String] -> Either String [URI]+convertURIs [] = Left "Must be at least one URI"+convertURIs ss =+ let uris = toURIs ss+ eqLen = length ss == length uris+ toURIs = filter expectedScheme . mapMaybe parseAbsoluteURI+ in if eqLen then Right uris+ else Left "Malformed URI(s)"++-- | Only expect the schemes of nats or tls.+expectedScheme :: URI -> Bool+expectedScheme uri = uriScheme uri == "nats:" || uriScheme uri == "tls:"++-- $limitations+--+-- 1) The current version of this library does not yet support TLS.+--+-- 2) The current version of this library does not yet support +-- authorization tokens (but support user names and passwords+-- in the URI strings).+--+-- $simple_messaging+--+-- This section gives a simple messaging example using this library. The+-- example requires the presence of a NATS server, running on localhost+-- using the default port 4222. If other host or port, adapt the+-- example.+--+-- > {-# LANGUAGE OverloadedStrings #-}+-- > module Main+-- > ( main+-- > ) where+-- >+-- > import Network.Nats+-- > import Text.Printf+-- >+-- > main :: IO ()+-- > main =+-- > withNats defaultSettings ["nats://localhost"] $ \nats -> do+-- >+-- > -- Subscribe to the topic "foo".+-- > (s, q) <- subscribe nats "foo" Nothing+-- >+-- > -- Publish to topic "foo", do not request a reply.+-- > publish nats "foo" Nothing "Some payload"+-- >+-- > -- Wait for a message, print the message's payload+-- > msg <- nextMsg q+-- > printf "Received %s\n" (show $ payload msg)+-- >+-- > -- Unsubscribe from topic "foo".+-- > unsubscribe nats s Nothing+--+-- $async_handling+--+-- Beside from the subscription mode where messages, synchronously, are+-- fetched from a queue there is also an asynchronous mode where each+-- request is handled immediately in their own thread.+--+-- > {-# LANGUAGE OverloadedStrings #-}+-- > module Main+-- > ( main+-- > ) where+-- >+-- > import Control.Monad+-- > import Data.Maybe+-- > import Network.Nats+-- > import Text.Printf+-- >+-- > main :: IO ()+-- > main =+-- > withNats defaultSettings ["nats://localhost"] $ \nats -> do+-- > +-- > -- A simple - asynchronous - help service that will answer+-- > -- requesters that give a reply topic with "I can help".+-- > s1 <- subscribeAsync nats "help" Nothing $ \msg -> do+-- > printf "Help service received: %s\n" (show $ payload msg)+-- > when (isJust $ replyTo msg) $+-- > publish nats (fromJust $ replyTo msg) Nothing "I can help"+-- >+-- > -- Subscribe to help replies.+-- > (s2, q) <- subscribe nats "help.reply" Nothing+-- >+-- > -- Request help.+-- > publish nats "help" (Just "help.reply") "Please ..."+-- >+-- > -- Wait for reply.+-- > msg <- nextMsg q+-- > printf "Received: %s\n" (show $ payload msg)+-- >+-- > -- Unsubscribe.+-- > unsubscribe nats s1 Nothing+-- > unsubscribe nats s2 Nothing+--+-- $request_pattern+--+-- In the example above there's a common request pattern. Sending a+-- message to a topic, requesting a reply, subscribing to the reply topic,+-- receiving the reply message and then unsubscribe from the reply topic.+--+-- This pattern can be handled more simply using the 'request' function.+--+-- > {-# LANGUAGE OverloadedStrings #-}+-- > module Main+-- > ( main+-- > ) where+-- >+-- > import Control.Monad+-- > import Data.Maybe+-- > import Network.Nats+-- > import Text.Printf+-- >+-- > main :: IO ()+-- > main =+-- > withNats defaultSettings ["nats://localhost"] $ \nats -> do+-- > +-- > -- A simple - asynchronous - help service that will answer+-- > -- requesters that give a reply topic with "I can help".+-- > s <- subscribeAsync nats "help" Nothing $ \msg -> do+-- > printf "Help service received: %s\n" (show $ payload msg)+-- > when (isJust $ replyTo msg) $+-- > publish nats (fromJust $ replyTo msg) Nothing "I can help"+-- >+-- > -- Request help.+-- > msg <- request nats "help" "Please ..."+-- > printf "Received: %s\n" (show $ payload msg)+-- >+-- > -- Unsubscribing the help service only.+-- > unsubscribe nats s Nothing+--+-- $topic_structure+--+-- Topic structure is tree like similar to file systems, or the Haskell+-- module structure, and components in the tree is separated by dots.+-- A subscriber of a topic can use wildcards to specify patterns.+--+-- > {-# LANGUAGE OverloadedStrings #-}+-- > module Main+-- > ( main+-- > ) where+-- >+-- > import Control.Monad+-- > import Data.Maybe+-- > import Network.Nats+-- > import Text.Printf+-- >+-- > main :: IO ()+-- > main =+-- > withNats defaultSettings ["nats://localhost"] $ \nats -> do+-- > +-- > -- "*" matches any token, at any level of the subject.+-- > (_, queue1) <- subscribe nats "foo.*.baz" Nothing+-- > (_, queue2) <- subscribe nats "foo.bar.*" Nothing+-- >+-- > -- ">" matches any length of the tail of the subject, and can+-- > -- only be the last token.+-- > (_, queue3) <- subscribe nats "foo.>" Nothing+-- >+-- > -- This publishing matches all the above.+-- > publish nats "foo.bar.baz" Nothing "Hello world"+-- >+-- > -- Show that the message showed up on all queues.+-- > forM_ [queue1, queue2, queue3] $ \queue -> do+-- > msg <- nextMsg queue+-- > printf "Received: %s\n" (show $ payload msg)
+ src/Network/Nats/Api.hs view
@@ -0,0 +1,209 @@+{-# LANGUAGE OverloadedStrings #-}+-- |+-- Module: Network.Nats.Api+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- NATS base API as provided by this library. A thin JSON layer is+-- added in "Network.Nats.JsonApi".+module Network.Nats.Api+ ( Nats+ , initNats+ , termNats+ , publish+ , subscribe+ , subscribeAsync+ , request+ , unsubscribe+ , nextMsg+ ) where++import Control.Concurrent.STM (atomically , newTQueueIO, readTQueue)+import Control.Exception (bracket)++import Network.Nats.Conduit (Downstream, Upstream, upstreamMessage)+import Network.Nats.ConnectionManager ( ConnectionManager+ , ManagerSettings+ , startConnectionManager+ , stopConnectionManager+ )+import Network.Nats.Dispatcher (Dispatcher, startDispatcher, stopDispatcher)+import Network.Nats.Types ( MsgQueue (..), Msg, Sid+ , Topic, Payload, QueueGroup+ )+import Network.Nats.Subscriber ( SubscriberMap, newSubscriberMap+ , addSubscriber, addAsyncSubscriber+ , removeSubscriber+ )+import Network.Nats.Message.Message (Message (..))++import Network.URI (URI)+import System.Random (randomRIO)++import qualified Data.ByteString.Char8 as BS++-- | The type of the handle used by the API. To the user this+-- type is opaque. The Nats handle is only valid within the scope of+-- 'Network.Nats.withNats' function.+data Nats = Nats+ { subscriberMap :: SubscriberMap+ -- ^ A map to hold 'Topic' subscribers.++ , connectionManager :: !ConnectionManager+ -- ^ The 'ConnectionManager'.++ , downstream :: !Downstream+ -- ^ The stream of messages from the NATS server to the client.++ , upstream :: !Upstream+ -- ^ The stream of messages from the client to the NATS server.++ , dispatcher :: !Dispatcher+ -- ^ The 'Dispatcher'.+ }++-- | Setting up of all the necessary resources needed by 'Nats'.+-- Suited for use with the style of resource management given by+-- 'Control.Exception.bracket'.+initNats :: ManagerSettings -> [URI] -> IO Nats+initNats config uris = do+ subscriberMap' <- newSubscriberMap+ downstream' <- newTQueueIO+ upstream' <- newTQueueIO+ manager <- startConnectionManager config upstream' + downstream' subscriberMap'+ uris+ dispatcher' <- startDispatcher downstream' + upstream' subscriberMap'++ return Nats { subscriberMap = subscriberMap'+ , connectionManager = manager+ , downstream = downstream'+ , upstream = upstream'+ , dispatcher = dispatcher'+ }++-- | Clean up 'Nats' resource. Used to clean up after 'initNats'.+termNats :: Nats -> IO ()+termNats nats = do+ stopDispatcher $ dispatcher nats+ stopConnectionManager $ connectionManager nats++-- | Publish some 'Payload' message to a 'Topic'. The NATS server will+-- distribute the message to subscribers of the 'Topic'.+--+-- > publish nats "GREETINGS" Nothing "Hello, there!"+--+-- Will publish the string Hello, there! to subscribers of GREETINGS. No+-- reply-to 'Topic' is provided. To request a reply, provide a 'Topic'+-- where the subscriber can publish a reply.+--+-- > publish nats "GREETINGS" (Just "THANKS") "Hello, there!"+publish :: Nats -> Topic -> Maybe Topic -> Payload -> IO ()+publish nats topic replyTo payload =+ upstreamMessage (upstream nats) $ PUB topic replyTo payload+{-# INLINE publish #-}++-- | Subscribe to a 'Topic'. Optionally a subscription can be part of+-- a 'QueueGroup'. The function will immediately return with a tuple of+-- a 'Sid' for the subscription, and a 'SubQueue' from where messages can+-- be fetched using 'nextMsg'.+--+-- > (sid, queue) <- subscribe nats "do.stuff" Nothing+--+-- Or+--+-- > (sid, queue) <- subscribe nats "do.stuff" (Just "stuffworkers")+subscribe :: Nats -> Topic -> Maybe QueueGroup -> IO (Sid, MsgQueue)+subscribe nats topic queueGroup = do+ sid <- newSid+ let msg = SUB topic queueGroup sid+ subQueue <- addSubscriber (subscriberMap nats) sid msg+ upstreamMessage (upstream nats) msg+ return (sid, subQueue)++-- | Subscribe to a 'Topic'. Optionally a subscription can be part of+-- a 'QueueGroup'.+--+-- Subscriptions using this function will be asynchronous, and each+-- message will be handled in its own thread. A message handler is+-- an IO action taking a 'Msg' as its argument. The function return+-- the 'Sid' for the subscription.+--+-- > sid <- subscribeAsync nats "do.stuff" Nothing $ \msg -> do+-- > -- Do stuff with the msg+--+-- Or+--+-- > sid <- subscribeAsync nats "do.stuff" Nothing messageHandler+-- >+-- > messageHandler :: Msg -> IO ()+-- > messageHandler msg = do+-- > -- Do stuff with the msg+subscribeAsync :: Nats -> Topic -> Maybe QueueGroup+ -> (Msg -> IO ()) -> IO Sid+subscribeAsync nats topic queueGroup action = do+ sid <- newSid+ let msg = SUB topic queueGroup sid+ addAsyncSubscriber (subscriberMap nats) sid msg action + upstreamMessage (upstream nats) msg+ return sid++-- | Request is publishing a 'Payload' to a 'Topic' and waiting for a+-- 'Msg'. Request is a blocking operation, but can be interrupted+-- by 'System.Timeout.timeout'.+--+-- > msg <- request nats "do.stuff" "A little payload"+--+-- Or+--+-- > maybeMsg <- timeout tmo $ request nats "do.stuff" "A little payload"+request :: Nats -> Topic -> Payload -> IO Msg+request nats topic payload = do+ replyTo <- randomReplyTo+ bracket (subscribe nats replyTo Nothing)+ (\(sid, _) -> unsubscribe nats sid Nothing)+ (\(_, queue) -> do+ publish nats topic (Just replyTo) payload+ nextMsg queue+ )++-- | Unsubscribe from a subscription using its 'Sid'. Optionally a limit+-- for automatic unsubscription can be given. Unsubscription will happen+-- once the number of messages - the limit - has been reached.+--+-- > unsubscribe nats sid Nothing+--+-- Or+--+-- > unsubscribe nats sid (Just 100)+unsubscribe :: Nats -> Sid -> Maybe Int -> IO ()+unsubscribe nats sid limit = do+ let msg = UNSUB sid limit+ removeSubscriber (subscriberMap nats) sid+ upstreamMessage (upstream nats) msg++-- | Fetch a new 'Msg' from the 'SubQueue'. Fetching a message is a+-- blocking operation, but can be interrupted by 'System.Timeout.timeout'.+--+-- > msg <- nextMsg queue+--+-- Or+--+-- > maybeMsg <- timeout tmo $ nextMsg queue+nextMsg :: MsgQueue -> IO Msg+nextMsg (MsgQueue queue) = atomically $ readTQueue queue+{-# INLINE nextMsg #-}++newSid :: IO Sid+newSid = randomRIO (0, maxBound)+{-# INLINE newSid #-}++randomReplyTo :: IO Topic+randomReplyTo = do+ value <- BS.pack . show <$> randomRIO (0, maxBound :: Int)+ return $ "INBOX." `BS.append` value+{-# INLINE randomReplyTo #-}
+ src/Network/Nats/Conduit.hs view
@@ -0,0 +1,78 @@+-- |+-- Module: Network.Nats.Conduit+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- 'Conduit' style helper functions.+module Network.Nats.Conduit+ ( Downstream+ , Upstream + , connectionSource+ , connectionSink+ , streamSource+ , streamSink+ , messageChunker+ , upstreamMessage+ ) where++import Control.Concurrent.STM (TQueue, atomically, readTQueue, writeTQueue)+import Control.DeepSeq (deepseq)+import Control.Monad (forever)+import Control.Monad.IO.Class (liftIO)+import Data.ByteString (ByteString)+import Data.Conduit (Conduit, Source, Sink, awaitForever, yield)++import Network.Nats.Message.Message (Message)+import Network.Nats.Message.Writer (writeMessage)++import qualified Data.ByteString.Lazy as LBS+import qualified Network.Connection as NC++-- | Downstream data from the client to the NATS server.+type Downstream = TQueue ByteString++-- | Upstream data from the NATS server to the client.+type Upstream = TQueue ByteString++-- | Source from a 'NC.Connection' to a 'ByteString'.+connectionSource :: NC.Connection -> Source IO ByteString+connectionSource c =+ forever $ yield =<< liftIO (NC.connectionGetChunk c)++-- | Sink a 'ByteString' to a 'NC.Connection'.+connectionSink :: NC.Connection -> Sink ByteString IO ()+connectionSink c = awaitForever $ + \chunk -> liftIO $ NC.connectionPut c chunk++-- | Source from an 'Upstream' to a 'ByteString'.+streamSource :: Upstream -> Source IO ByteString+streamSource stream =+ forever $ yield =<< liftIO (atomically $ readTQueue stream)++-- | Sink a 'ByteString' to a 'Downstream'.+streamSink :: Downstream -> Sink ByteString IO ()+streamSink stream = awaitForever $+ \chunk -> liftIO $ atomically $ writeTQueue stream chunk++-- | Take one 'Message', encode it and create chunks of it.+messageChunker :: Conduit Message IO ByteString+messageChunker = awaitForever $+ \msg -> mapM_ yield (LBS.toChunks $ writeMessage msg)++-- | Not really a conduit, but a pusher of 'Message' chunks to the+-- 'Upstream'.+upstreamMessage :: Upstream -> Message -> IO ()+upstreamMessage upstream msg = do+ -- Most likely there are different threads pushing messages upstream+ -- using this function. Force as much work as possible outside of+ -- the 'atomically' operations, to prevent contention on the TQueue.+ let chunks = LBS.toChunks $ writeMessage msg+ chunks `deepseq` upstreamMessage' chunks+ where+ upstreamMessage' :: [ByteString] -> IO ()+ upstreamMessage' chunks = + atomically $ mapM_ (writeTQueue upstream) chunks+{-# INLINE upstreamMessage #-}
+ src/Network/Nats/Connection.hs view
@@ -0,0 +1,218 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+-- |+-- Module: Network.Nats.ConnectionManager+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- | Abstraction for a connection towards a NATS server. It owns the+-- networking stuff and performs NATS handshaking necessary.+module Network.Nats.Connection+ ( Connection (sockAddr)+ , Upstream+ , Downstream+ , makeConnection+ , clientShutdown+ , waitForShutdown+ ) where++import Control.Concurrent.Async (Async, async, waitAnyCatchCancel)+import Control.Exception (SomeException, fromException, throwIO, handle)+import Control.Monad (void, when)+import Data.Conduit (($$), (=$=))+import Data.Conduit.Attoparsec (sinkParser)+import Data.Conduit.List (sourceList)+import Data.List+import Data.Maybe (fromJust, isNothing)+import Network.Socket ( AddrInfo (..), HostName, PortNumber+ , SockAddr, defaultHints, getAddrInfo+ )+import Network.URI (URI, uriAuthority, uriRegName, uriPort, uriUserInfo)+import System.Timeout (timeout)++import Network.Nats.Types (NatsException (..))+import Network.Nats.Conduit ( Upstream, Downstream, connectionSource+ , connectionSink, streamSource+ , streamSink, messageChunker+ )+import Network.Nats.Subscriber (SubscriberMap, subscribeMessages)+import Network.Nats.Message.Message (Message (..))+import Network.Nats.Message.Parser (parseMessage)+import Network.Nats.Message.Writer (writeMessage)++import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy as LBS+import qualified Network.Connection as NC++-- | Type alias for a microsecond timeout.+type Tmo = Int++-- | Record representing an active connection towards the NATS server.+data Connection = Connection+ { connection :: !NC.Connection+ , sockAddr :: !SockAddr+ , fromNet :: !(Async ())+ , toNet :: !(Async ())+ }++-- | Make a new 'Connection' as specified by the URI. Provide one+-- 'Upstream' queue with data from the application to the server, and+-- one 'Downstream' queue with data from the server to the application.+makeConnection :: Tmo -> URI -> Upstream -> Downstream -> SubscriberMap + -> IO (Maybe Connection)+makeConnection tmo uri fromApp toApp subscriberMap =+ connectionError `handle` (Just <$> makeConnection' tmo uri fromApp + toApp subscriberMap)+ where+ connectionError :: SomeException -> IO (Maybe Connection)+ connectionError e+ | isConnectionRefused e = return Nothing+ | isResolvError e = return Nothing+ | otherwise =+ case fromException e of+ (Just HandshakeException) -> return Nothing+ _ -> throwIO e+ +makeConnection' :: Tmo -> URI -> Upstream -> Downstream -> SubscriberMap+ -> IO Connection+makeConnection' tmo uri fromApp toApp subscriberMap = do+ let host = hostFromUri uri+ port = portFromUri uri++ -- Make the connection.+ ctx <- NC.initConnectionContext+ conn <- NC.connectTo ctx+ NC.ConnectionParams+ { NC.connectionHostname = host+ , NC.connectionPort = port+ , NC.connectionUseSocks = Nothing+ , NC.connectionUseSecure = Nothing+ }++ -- Perform the handshaking of 'Info' and 'Connect' messages between+ -- the client and the server. If the time to receive the 'Info'+ -- message exceeds the timeout, there's a HandshakeException.+ msg <- timeout tmo $ getSingleMessage conn+ when (isNothing msg) $ do+ NC.connectionClose conn+ throwIO HandshakeException++ -- Continue with the handshake.+ handshake uri conn $ fromJust msg++ -- Fetch already made subscriptions for replay.+ msgs <- subscribeMessages subscriberMap++ -- Now start the pipeline threads and let the fun begin.+ Connection conn <$> toSockAddr host port+ <*> async (recvPipe conn toApp)+ <*> async (do+ replaySubscriptions conn msgs+ sendPipe fromApp conn)++-- | Pipeline to run the 'Downstream' conduit.+recvPipe :: NC.Connection -> Downstream -> IO ()+recvPipe conn toApp = connectionSource conn $$ streamSink toApp++-- | Pipeline to run the 'Upstream' conduit.+sendPipe :: Upstream -> NC.Connection -> IO ()+sendPipe fromApp conn = streamSource fromApp $$ connectionSink conn++-- | Replay all stored subscriptions to the 'NC.Connection'.+replaySubscriptions :: NC.Connection -> [Message] -> IO ()+replaySubscriptions conn msgs =+ sourceList msgs =$= messageChunker $$ connectionSink conn++-- | Shut down a 'Connection' by cancel the threads.+clientShutdown :: Connection -> IO ()+clientShutdown conn = do+ -- The close of the connection will make the threads terminate.+ NC.connectionClose $ connection conn+ void $ waitAnyCatchCancel [ fromNet conn, toNet conn ]++-- | Blocking wait for the connection to shutdown (perhaps it+-- never does).+waitForShutdown :: Connection -> IO ()+waitForShutdown conn = do+ void $ waitAnyCatchCancel [ fromNet conn, toNet conn ]+ NC.connectionClose $ connection conn++-- | Perform the handshake.+-- TODO: More handshaking, tls, tokens etc.+handshake :: URI -> NC.Connection -> Message -> IO ()+handshake uri conn INFO {..} = do+ let (user, pass) = credentialsFromUri uri+ connect = CONNECT { clientVerbose = Just False+ , clientPedantic = Just False+ , clientSslRequired = Just False+ , clientAuthToken = Nothing+ , clientUser = user+ , clientPass = pass+ , clientName = Just "hats"+ , clientLang = Just "Haskell"+ , clientVersion = Just "0.1.0.0"+ }+ mapM_ (NC.connectionPut conn) $ LBS.toChunks (writeMessage connect)++handshake _ _ _ = throwIO HandshakeException++-- | Select the host part from the 'URI'.+hostFromUri :: URI -> HostName+hostFromUri = uriRegName . fromJust . uriAuthority++-- | Select the port part from the 'URI'.+portFromUri :: URI -> PortNumber+portFromUri = fromIntegral . extractPort . uriPort . fromJust . uriAuthority++-- | Extract credentials (if any) from the 'URI'.+credentialsFromUri :: URI -> (Maybe BS.ByteString, Maybe BS.ByteString)+credentialsFromUri = + toBS . extractCredentials . uriUserInfo. fromJust . uriAuthority+ where+ toBS (user, pass) = (BS.pack <$> user, BS.pack <$> pass)++-- | Resolve a 'HostName' and a 'PortNumber' to a 'SockAddr'.+toSockAddr :: HostName -> PortNumber -> IO SockAddr+toSockAddr host port =+ addrAddress . head <$> getAddrInfo (Just defaultHints) + (Just host)+ (Just $ show port)++-- | When selected the port is a string of format ":4222". Skip the colon,+-- or if the port is missing, give the default port of 4222.+extractPort :: String -> Int+extractPort [] = 4222+extractPort ":" = 4222+extractPort (':':str) = read str+extractPort _ = error "This is no valid port, ehh?"++-- | Extract the credentials from a String.+extractCredentials :: String -> (Maybe String, Maybe String)+extractCredentials "" = (Nothing, Nothing)+extractCredentials str =+ let str' = takeWhile (/= '@') str+ colon = elemIndex ':' str'+ in+ if isNothing colon+ then (Just str', Nothing)+ else+ let (user, _:pass) = splitAt (fromJust colon) str'+ in (Just user, Just pass)++-- | Awkward, but this is how to check for connection refuse.+isConnectionRefused :: SomeException -> Bool+isConnectionRefused e =+ show e == "connect: does not exist (Connection refused)"++-- | Equally awkward, but this is how to check for resolv errors.+isResolvError :: SomeException -> Bool+isResolvError e =+ show e == "getAddrInfo: does not exist (Name or service not known)"++-- | Get one single message from the 'NC.Connection'. It should be the+-- initial 'Info' message from the NATS server.+getSingleMessage :: NC.Connection -> IO Message+getSingleMessage c = connectionSource c $$ sinkParser parseMessage
+ src/Network/Nats/ConnectionManager.hs view
@@ -0,0 +1,179 @@+{-# LANGUAGE RecordWildCards #-}+-- |+-- Module: Network.Nats.ConnectionManager+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- Configuration and functionality for setting up and maintaining+-- connections towards a NATS messaging server.+module Network.Nats.ConnectionManager+ ( ConnectionManager+ , ManagerSettings (..)+ , SockAddr+ , startConnectionManager+ , stopConnectionManager+ , defaultSettings+ , randomSelect+ , roundRobinSelect+ ) where++import Control.Concurrent (ThreadId, forkIO, killThread, myThreadId)+import Control.Concurrent.STM ( TVar, atomically, newTVarIO+ , readTVarIO, writeTVar+ )+import Control.Exception (handle, throwIO, throwTo)+import Network.URI (URI)+import Network.Socket (SockAddr)+import System.Random (randomRIO)++import Network.Nats.Connection ( Connection, Downstream+ , Upstream, makeConnection+ , clientShutdown, waitForShutdown, sockAddr+ )+import Network.Nats.Subscriber (SubscriberMap)+import Network.Nats.Types (NatsException (..))++-- | Resourced aquired by the connection manager. The record is+-- opaque to the user.+data ConnectionManager = ConnectionManager+ { connection :: TVar (Maybe Connection)+ , managerThread :: ThreadId+ }++-- | Internal runtime context for the connection manager.+data ManagerContext = ManagerContext+ { upstream :: !Upstream+ , downstream :: !Downstream+ , subscriberMap :: !SubscriberMap+ , uris :: ![URI]+ , currUriIdx :: !Int+ , currConn :: TVar (Maybe Connection)+ , callerThread :: !ThreadId+ }++-- | A set of parameters to guide the behavior of the connection manager.+-- A default set of parameters can be obtained by calling+-- 'defaultSettings'.+data ManagerSettings = ManagerSettings+ { reconnectionAttempts :: !Int+ -- ^ The number of times the connection manager shall try to+ -- connect a server before giving up.+ + , maxWaitTimeMS :: !Int+ -- ^ Maximum waiting between a connection is made and a CONNECT+ -- message is received from the NATS server. If exceeded the+ -- connection is terminated and a new server selection is made.+ -- The unit for the time is in milliseconds.+ + , serverSelect :: ([URI], Int) -> IO (URI, Int)+ -- ^ A function to select one of the servers from the+ -- server pool. The arguments to the selector is the list of server+ -- uris and the current index. The reply is the chosen server and+ -- its index.++ , connectedTo :: SockAddr -> IO ()+ -- ^ Callback to inform that a connection is made to the NATS+ -- server, with the 'SockAddr' for the server. This callback is+ -- made in the connection manager's thread, and the callback's+ -- execution time must be minimized. 'Control.Concurrent.forkIO'+ -- if longer execution times are needed.++ , disconnectedFrom :: SockAddr -> IO ()+ -- ^ Callback to inform that a disconnection to the NATS server+ -- has happen. Give the 'SockAddr' for the server. This callback is+ -- made in the connection manager's thread, and the callback's+ -- execution time must be minimized. 'Control.Concurrent.forkIO'+ -- if longer execution times are needed.++ }++-- | Start the connection manager.+startConnectionManager :: ManagerSettings+ -> Upstream+ -> Downstream+ -> SubscriberMap+ -> [URI]+ -> IO ConnectionManager+startConnectionManager settings upstream' downstream' + subscriberMap' uris' = do+ -- The transactional 'Connection' is shared between the 'ManagerContext'+ -- and the 'ConnectionManager'.+ conn <- newTVarIO Nothing+ caller <- myThreadId+ let context = ManagerContext { upstream = upstream'+ , downstream = downstream'+ , subscriberMap = subscriberMap'+ , uris = uris'+ , currUriIdx = -1+ , currConn = conn+ , callerThread = caller+ }+ ConnectionManager <$> pure conn+ <*> forkIO (managerLoop settings context)++-- | Stop the connection manager, clean up stuff.+stopConnectionManager :: ConnectionManager -> IO ()+stopConnectionManager mgr = do+ -- The order when shutting down things is important. First the+ -- managerThread must be stopped (so it not tries to create new+ -- connections). Then the 'Connection' can be stopped.+ killThread $ managerThread mgr++ -- The 'Connection' is folded in 'Maybe'.+ mapM_ clientShutdown =<< readTVarIO (connection mgr)++-- | Create a default 'ManagerSettings'.+defaultSettings :: ManagerSettings+defaultSettings =+ ManagerSettings+ { reconnectionAttempts = 5+ , maxWaitTimeMS = 2000+ , serverSelect = roundRobinSelect+ , connectedTo = const (return ())+ , disconnectedFrom = const (return ())+ }++-- | Make a random selection of a server.+randomSelect :: ([URI], Int) -> IO (URI, Int)+randomSelect (xs, _) = do+ idx <- randomRIO (0, length xs - 1)+ return (xs !! idx, idx)++-- | Use round robin to select a server.+roundRobinSelect :: ([URI], Int) -> IO (URI, Int)+roundRobinSelect (xs, currIdx)+ | currIdx == length xs - 1 = return (head xs, 0)+ | otherwise = return (xs !! (currIdx + 1), currIdx + 1)++-- | Connect to a server and maintain the connection.+managerLoop :: ManagerSettings -> ManagerContext -> IO ()+managerLoop mgr@ManagerSettings {..} ctx@ManagerContext {..} = + exceptionForward callerThread `handle` do+ (newIdx, conn) <- tryConnect mgr ctx reconnectionAttempts+ atomically $ writeTVar currConn (Just conn)+ connectedTo $ sockAddr conn+ waitForShutdown conn+ atomically $ writeTVar currConn Nothing+ disconnectedFrom $ sockAddr conn+ managerLoop mgr $ ctx { currUriIdx = newIdx }++-- | Select a server and connect.+tryConnect :: ManagerSettings -> ManagerContext -> Int+ -> IO (Int, Connection)+tryConnect _ _ 0 = throwIO ConnectionGiveUpException+tryConnect mgr@ManagerSettings {..} ctx@ManagerContext {..} attempts = do+ (uri, uriIdx) <- serverSelect (uris, currUriIdx)+ maybe (tryConnect mgr ctx (attempts - 1))+ (\conn -> return (uriIdx, conn)) + =<< makeConnection (toUS maxWaitTimeMS) uri + upstream downstream subscriberMap++-- | Throw 'NatsException's to the provided thread.+exceptionForward :: ThreadId -> NatsException -> IO ()+exceptionForward = throwTo++toUS :: Int -> Int+toUS n = n * 1000
+ src/Network/Nats/Dispatcher.hs view
@@ -0,0 +1,117 @@+-- |+-- Module: Network.Nats.Dispatcher+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- The dispatcher is receiving 'Downstream' messages from the NATS+-- server and dispatches them to their receivers.+module Network.Nats.Dispatcher+ ( Dispatcher+ , startDispatcher+ , stopDispatcher+ ) where++import Control.Concurrent (ThreadId, forkIO, killThread, myThreadId)+import Control.Concurrent.STM (atomically, writeTQueue)+import Control.Exception (SomeException, handle, throwTo)+import Control.Monad (void)+import Control.Monad.IO.Class (liftIO)+import Data.Conduit (Sink, ($$), (=$=), awaitForever)+import Data.Conduit.Attoparsec ( ParseError, PositionRange+ , conduitParserEither+ )++import Network.Nats.Conduit ( Downstream, Upstream+ , streamSource, upstreamMessage+ )+import Network.Nats.Subscriber ( Subscriber (..), SubscriberMap+ , lookupSubscriber+ )+import Network.Nats.Types (Msg (..), NatsException (..))+import Network.Nats.Message.Message (Message (..), ProtocolError (..))+import Network.Nats.Message.Parser (parseMessage)++-- | A data type to hold the 'ThreadId' for the dispatcher thread.+newtype Dispatcher = Dispatcher ThreadId++-- | Start the dispatcher thread. Give the dispatcher thread the callers+-- 'ThreadId', and in this case it is assumed that the caller is the+-- same thread as the caller of withNats. The caller thread is used in+-- case an 'AuthorizationException' need to be thrown.+startDispatcher :: Downstream -> Upstream -> SubscriberMap+ -> IO Dispatcher+startDispatcher downstream upstream subscriberMap = do+ caller <- myThreadId+ Dispatcher <$>+ forkIO (dispatcher caller downstream upstream subscriberMap)++-- | Kill the dispatcher thread.+stopDispatcher :: Dispatcher -> IO ()+stopDispatcher (Dispatcher t) = killThread t++-- | The dispatcher pipeline from the 'Downstream', through the message+-- parser and to the core dispatcher.+dispatcher :: ThreadId -> Downstream -> Upstream -> SubscriberMap -> IO ()+dispatcher caller downstream upstream subscriberMap =+ streamSource downstream =$=+ conduitParserEither parseMessage $$+ messageSink caller upstream subscriberMap++-- | The message 'Sink'. Forever receive messages, if there are+-- parser error print those, otherwise just dispatch the message.+messageSink :: ThreadId -> Upstream -> SubscriberMap+ -> Sink (Either ParseError (PositionRange, Message)) IO ()+messageSink caller upstream subscriberMap =+ awaitForever $+ \eMsg -> case eMsg of+ Right (_, msg) -> liftIO $ dispatchMessage caller upstream+ subscriberMap msg+ Left err -> liftIO $ print err+{-# INLINE messageSink #-}++-- | Dispatch on 'M.Message. Handles 'MSG' and 'PING'.+dispatchMessage :: ThreadId -> Upstream -> SubscriberMap -> Message -> IO ()++-- Receive one 'MSG'. Lookup its 'Subscriber' and feed it with the+-- message. If no 'Subscriber' is found, the message is silently+-- discarded.+dispatchMessage _ _ subscriberMap (MSG topic sid replyTo payload) = do+ let msg = Msg topic replyTo sid payload+ maybe (return ()) (feedSubscriber msg) =<<+ lookupSubscriber subscriberMap sid++-- Handle 'PING' messages. Just reply with 'PONG'.+dispatchMessage _ upstream _ PING = upstreamMessage upstream PONG++-- Handle 'ERR' messages. If there are authorization violation, the+-- dispatcher will throw. Other errors are just logged for now.+dispatchMessage caller _ _ (ERR err)+ | err == AuthorizationViolation = throwTo caller AuthorizationException+ | otherwise = print err++-- Other are messages this dispatcher doesn't care about. The 'INFO'+-- message is handled by 'Connection' at connection handshake.+dispatchMessage _ _ _ _ = return ()+{-# INLINE dispatchMessage #-}++-- | Feed a 'Subscriber' with a 'Msg'.+feedSubscriber :: Msg -> Subscriber -> IO ()++-- Queue subscriber. Write the queue.+feedSubscriber msg (Subscriber queue _) =+ atomically $ writeTQueue queue msg++-- Async subscriber. Fork a handler.+feedSubscriber msg (AsyncSubscriber action _) =+ void $ forkIO $ asyncTrampoline (action msg)+{-# INLINE feedSubscriber #-}++-- | Trampoline to handle exceptions from an async handler.+asyncTrampoline :: IO () -> IO ()+asyncTrampoline action = allExceptions `handle` action+ where+ allExceptions :: SomeException -> IO ()+ allExceptions e = putStrLn $ "Async handler crash: " ++ show e
+ src/Network/Nats/JsonApi.hs view
@@ -0,0 +1,29 @@+-- |+-- Module: Network.Nats.JsonApi+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- NATS base library extended with with JSON coding of payload. JSON+-- handling is provided by "Data.Aeson".+module Network.Nats.JsonApi+ ( publishJson+ , requestJson+ ) where++import Data.Aeson (ToJSON, encode)++import Network.Nats.Api (Nats, publish, request)+import Network.Nats.Types (Msg, Topic)++-- | As 'publish', but with JSON payload.+publishJson :: ToJSON a => Nats -> Topic -> Maybe Topic -> a -> IO ()+publishJson nats topic replyTo = publish nats topic replyTo . encode+{-# INLINE publishJson #-}++-- | As 'request', but with JSON payload and 'JsonMsg' reply.+requestJson :: ToJSON a => Nats -> Topic -> a -> IO Msg+requestJson nats topic = request nats topic . encode+{-# INLINE requestJson #-}
+ src/Network/Nats/Message/Message.hs view
@@ -0,0 +1,141 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+-- |+-- Module: Network.Nats.Message.Message+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- Message definitions for the NATS protocol.+module Network.Nats.Message.Message+ ( Message (..)+ , ProtocolError (..)+ ) where++import Control.DeepSeq (NFData)+import GHC.Generics (Generic)+import qualified Data.ByteString as BS++import Network.Nats.Types (Topic, Payload, Sid, QueueGroup)++-- | Protocol error enumeration.+data ProtocolError+ = UnknownProtocolOperation+ | AuthorizationViolation+ | AuthorizationTimeout+ | ParserError+ | StaleConnection+ | SlowConsumer+ | MaximumPayloadExceeded+ | InvalidSubject+ deriving (Bounded, Enum, Eq, Generic, NFData, Show)++-- | The kind of messages that can be exchanged between the NATS server+-- and a NATS client.+-- Some of the documentation strings are taken from:+-- <https://nats.io/documentation/internals/nats-protocol/>+data Message =+ -- | As soon as the server accepts a connection from the client, it+ -- will send information about itself and the configuration and+ -- security requirements that are necessary for the client to+ -- successfully authenticate with the server and+ -- exchange messages.+ INFO { serverId :: !(Maybe BS.ByteString)+ -- ^ The unique identifier of the NATS server.+ , serverVersion :: !(Maybe BS.ByteString)+ -- ^ The version of the NATS server.+ , goVersion :: !(Maybe BS.ByteString)+ -- ^ The version of golang the server was built with.+ , serverHost :: !(Maybe BS.ByteString)+ -- ^ The IP address of the NATS server host.+ , serverPort :: !(Maybe Int)+ -- ^ The port number the NATS server is configured to listen on.+ , serverAuthRequired :: !(Maybe Bool)+ -- ^ If set, the client should try to authenticate.+ , serverSslRequired :: !(Maybe Bool)+ -- ^ If set, the client must authenticate using SSL.+ , serverTlsRequired :: !(Maybe Bool)+ , serverTlsVerify :: !(Maybe Bool)+ , maxPayload :: !(Maybe Int)+ -- ^ Maximum payload size that server will accept from client.+ }++ -- | The CONNECT message is the reply to the Info message. Once the+ -- client has established a TCP/IP socket connection with the NATS+ -- server, and an Info message has been received from the server,+ -- the client may sent a Connect message to the NATS server to+ -- provide more information about the current connection as+ -- well as security information.+ | CONNECT { clientVerbose :: !(Maybe Bool)+ -- ^ Turns on +OK protocol acknowledgements.+ , clientPedantic :: !(Maybe Bool)+ -- ^ Turns on additional strict format checking.+ , clientSslRequired :: !(Maybe Bool)+ -- ^ Indicates whether the client requires an SSL connection.+ , clientAuthToken :: !(Maybe BS.ByteString)+ -- ^ Client authorization token.+ , clientUser :: !(Maybe BS.ByteString)+ -- ^ Connection username (if auth_required is set).+ , clientPass :: !(Maybe BS.ByteString)+ -- ^ Connection password (if auth_required is set).+ , clientName :: !(Maybe BS.ByteString)+ -- ^ Optional client name.+ , clientLang :: !(Maybe BS.ByteString)+ -- The implementation of the client.+ , clientVersion :: !(Maybe BS.ByteString)+ -- ^ The version of the client.+ }++ -- | The MSG message carries payload from the server to the client.+ -- subject: Subject name this message was received on.+ -- sid: The unique alphanumeric subscription ID of the subject.+ -- reply-to: The inbox subject on which the publisher is listening+ -- for responses.+ -- #bytes: Implicit by the length of the payload.+ -- payload: The message payload data.+ | MSG !Topic {-# UNPACK #-} !Sid !(Maybe Topic) !Payload++ -- | The PUB message publishes the message payload to the given+ -- subject name. If a reply subject is supplied, it will be+ -- delivered to eligible subscribers along with the supplied+ -- payload. Payload can be zero size.+ -- subject: The destination subject to publish to.+ -- reply-to: The reply inbox subject that subscribers can use to+ -- send a response back to the publisher/requestor.+ -- #bytes: Implicit by the length of payload.+ -- payload: The message payload data.+ | PUB !Topic !(Maybe Topic) !Payload++ -- | The SUB message initiates a subscription to a subject,+ -- optionally joining a distributed queue group.+ -- subject: The subject name to subscribe to.+ -- (Optional) queue group: If specified, the subscriper will join + -- this queue group.+ -- sid: A unique alphanumeric SubscriptionId.+ | SUB !Topic !(Maybe QueueGroup) {-# UNPACK #-} !Sid++ -- | The UNSUB message unsubscribes the connection from the specified+ -- subject, or auto-unsubscribes after the specified number of + -- messages has been received.+ | UNSUB {-# UNPACK #-} !Sid !(Maybe Int)++ -- | The PING and PONG messages are the keep-alive mechanism between+ -- the client and the server. The server will continously send+ -- Ping messages to the client. If the client does not reply+ -- within time the connection is terminated by the server.+ | PING+ | PONG++ -- | When the verbose (clientVerbose) option is set to true, the+ -- server acknowledges each well-formed prototol message from the+ -- client with a +OK message.+ | OK++ -- | The -ERR message is used by the server indicate a protocol,+ -- authorization, or other runtime connection error to the client.+ -- Most of those errors result in the server closing the+ -- connection. InvalidSubject is the exception.+ | ERR !ProtocolError+ deriving (Eq, Generic, NFData, Show)
+ src/Network/Nats/Message/Parser.hs view
@@ -0,0 +1,354 @@+{-# LANGUAGE OverloadedStrings #-}+-- |+-- Module: Network.Nats.Message.Parser+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- NATS protocol 'Message' parser. To be used with the+-- "Data.Attoparsec.ByteString" library.+module Network.Nats.Message.Parser+ ( parseMessage+ ) where++import Control.Applicative ((<|>))+import Control.Monad (void)+import Data.Attoparsec.ByteString.Char8+import Data.ByteString.Char8 (ByteString)++import qualified Data.Attoparsec.ByteString.Char8 as AP+import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy.Char8 as LBS++import Network.Nats.Message.Message ( Message (..)+ , ProtocolError (..)+ )+import Network.Nats.Types (Sid)++data HandshakeMessageValue =+ Bool !Bool+ | String !ByteString+ | Int !Int+ deriving Show++type HandshakeMessageField = (ByteString, HandshakeMessageValue)++-- | Parse a NATS message.+parseMessage :: Parser Message+parseMessage = skipSpace *> parseMessage'+ where+ parseMessage' = msgMessage+ <|> infoMessage + <|> connectMessage+ <|> pubMessage+ <|> subMessage+ <|> unsubMessage+ <|> pingMessage+ <|> pongMessage+ <|> okMessage + <|> errMessage++-- | The parsing of the Info message is not performance critical.+infoMessage :: Parser Message+infoMessage = do+ spacedMsgName "INFO"+ void $ char '{'+ fields <- parseInfoMessageFields+ void $ char '}'+ mkInfoMessage fields++parseInfoMessageFields :: Parser [HandshakeMessageField]+parseInfoMessageFields = infoMessageField `sepBy` char ','+ where+ infoMessageField = parseServerId+ <|> parseVersion+ <|> parseGoVersion+ <|> parseServerHost+ <|> parseServerPort+ <|> parseServerAuthRequired+ <|> parseSslRequired+ <|> parseTlsRequired+ <|> parseTlsVerify+ <|> parseMaxPayload++-- | Nor is the parsing the Connect message performace critical.+connectMessage :: Parser Message+connectMessage = do+ spacedMsgName "CONNECT"+ void $ char '{'+ fields <- parseConnectMessageFields+ void $ char '}'+ newLine+ mkConnectMessage fields++parseConnectMessageFields :: Parser [HandshakeMessageField]+parseConnectMessageFields = connectMessageField `sepBy` char ','+ where+ connectMessageField = parseClientVerbose+ <|> parseClientPedantic+ <|> parseSslRequired+ <|> parseClientAuthToken+ <|> parseClientUser+ <|> parseClientPass+ <|> parseClientName+ <|> parseClientLang+ <|> parseVersion++-- | Parse a MSG message ...+msgMessage :: Parser Message+msgMessage = msgMessageWithReply <|> msgMessageWithoutReply++-- | ... with a reply-to field.+msgMessageWithReply :: Parser Message+msgMessageWithReply = do+ spacedMsgName "MSG"+ MSG <$> takeTill isSpace <* singleSpace+ <*> parseSid <* singleSpace+ <*> (Just <$> takeTill isSpace <* singleSpace)+ <*> readPayload <* newLine++-- | ... and without a reply-to.+msgMessageWithoutReply :: Parser Message+msgMessageWithoutReply = do+ spacedMsgName "MSG"+ MSG <$> takeTill isSpace <* singleSpace+ <*> parseSid <* singleSpace+ <*> pure Nothing+ <*> readPayload <* newLine++-- | Parse a PUB message ...+pubMessage :: Parser Message+pubMessage = pubMessageWithReply <|> pubMessageWithoutReply++-- | ... with a reply-to field.+pubMessageWithReply :: Parser Message+pubMessageWithReply = do+ spacedMsgName "PUB"+ PUB <$> takeTill isSpace <* singleSpace+ <*> (Just <$> takeTill isSpace <* singleSpace)+ <*> readPayload <* newLine++-- | ... and without a reply-to.+pubMessageWithoutReply :: Parser Message+pubMessageWithoutReply = do+ spacedMsgName "PUB"+ PUB <$> takeTill isSpace <* singleSpace+ <*> pure Nothing+ <*> readPayload <* newLine++-- | Helper parser to read the length/payload pair from a PUB/MSG+-- message.+readPayload :: Parser LBS.ByteString+readPayload = do+ len <- decimal <* newLine+ LBS.fromStrict <$> AP.take len+{-# INLINE readPayload #-}++-- | Parse a SUB message ...+subMessage :: Parser Message+subMessage = subMessageWithQueue <|> subMessageWithoutQueue++-- | ... with a queue group.+subMessageWithQueue :: Parser Message+subMessageWithQueue = do+ spacedMsgName "SUB"+ SUB <$> takeTill isSpace <* singleSpace+ <*> (Just <$> takeTill isSpace <* singleSpace)+ <*> parseSid <* newLine+ +-- | ... and without a queue group.+subMessageWithoutQueue :: Parser Message+subMessageWithoutQueue = do+ spacedMsgName "SUB"+ SUB <$> takeTill isSpace <* singleSpace+ <*> pure Nothing+ <*> parseSid <* newLine++-- | Parse an UNSUB message ...+unsubMessage :: Parser Message+unsubMessage = unsubMessageWithLimit <|> unsubMessageWithoutLimit++-- | ... with an unsubscribe limit.+unsubMessageWithLimit :: Parser Message+unsubMessageWithLimit = do+ spacedMsgName "UNSUB"+ UNSUB <$> parseSid <* singleSpace+ <*> (Just <$> decimal) <* newLine++-- | ... and without an unsubscribe limit.+unsubMessageWithoutLimit :: Parser Message+unsubMessageWithoutLimit = do+ spacedMsgName "UNSUB"+ UNSUB <$> parseSid <* newLine+ <*> pure Nothing++pingMessage :: Parser Message+pingMessage = (msgName "PING" >> newLine) *> pure PING++pongMessage :: Parser Message+pongMessage = (msgName "PONG" >> newLine) *> pure PONG++okMessage :: Parser Message+okMessage = (msgName "+OK" >> newLine) *> pure OK++errMessage :: Parser Message+errMessage = do+ spacedMsgName "-ERR"+ ERR <$> protocolError <* newLine++parseServerId :: Parser HandshakeMessageField+parseServerId = pair "\"server_id\"" quotedString "server_id" String++parseVersion :: Parser HandshakeMessageField+parseVersion = pair "\"version\"" quotedString "version" String++parseGoVersion :: Parser HandshakeMessageField+parseGoVersion = pair "\"go\"" quotedString "go" String++parseServerHost :: Parser HandshakeMessageField+parseServerHost = pair "\"host\"" quotedString "host" String++parseServerPort :: Parser HandshakeMessageField+parseServerPort = pair "\"port\"" decimal "port" Int++parseServerAuthRequired :: Parser HandshakeMessageField+parseServerAuthRequired = + pair "\"auth_required\"" boolean "auth_required" Bool++parseSslRequired :: Parser HandshakeMessageField+parseSslRequired = pair "\"ssl_required\"" boolean "ssl_required" Bool++parseTlsRequired :: Parser HandshakeMessageField+parseTlsRequired = pair "\"tls_required\"" boolean "tls_required" Bool++parseTlsVerify :: Parser HandshakeMessageField+parseTlsVerify = pair "\"tls_verify\"" boolean "tls_verify" Bool++parseMaxPayload :: Parser HandshakeMessageField+parseMaxPayload = pair "\"max_payload\"" decimal "max_payload" Int++parseClientVerbose :: Parser HandshakeMessageField+parseClientVerbose = pair "\"verbose\"" boolean "verbose" Bool++parseClientPedantic :: Parser HandshakeMessageField+parseClientPedantic = pair "\"pedantic\"" boolean "pedantic" Bool++parseClientAuthToken :: Parser HandshakeMessageField+parseClientAuthToken =+ pair "\"auth_token\"" quotedString "auth_token" String++parseClientUser :: Parser HandshakeMessageField+parseClientUser = pair "\"user\"" quotedString "user" String++parseClientPass :: Parser HandshakeMessageField+parseClientPass = pair "\"pass\"" quotedString "pass" String++parseClientName :: Parser HandshakeMessageField+parseClientName = pair "\"name\"" quotedString "name" String++parseClientLang :: Parser HandshakeMessageField+parseClientLang = pair "\"lang\"" quotedString "lang" String++pair :: ByteString -> Parser a -> ByteString + -> (a -> HandshakeMessageValue) + -> Parser HandshakeMessageField+pair fieldName parser keyName ctor = do+ void $ string fieldName+ void $ char ':'+ value <- parser+ return (keyName, ctor value)++quotedString :: Parser ByteString+quotedString = BS.pack <$> (char '\"' *> manyTill anyChar (char '\"'))++boolean :: Parser Bool+boolean = string "false" *> return False <|> string "true" *> return True++protocolError :: Parser ProtocolError+protocolError =+ stringCI "\'Unknown Protocol Operation\'" + *> return UnknownProtocolOperation <|>++ stringCI "\'Authorization Violation\'"+ *> return AuthorizationViolation <|>++ stringCI "\'Authorization Timeout\'"+ *> return AuthorizationTimeout <|>++ stringCI "\'Parser Error\'"+ *> return ParserError <|>++ stringCI "\'Stale Connection\'"+ *> return StaleConnection <|>++ stringCI "\'Slow Consumer\'"+ *> return SlowConsumer <|>++ stringCI "\'Maximum Payload Exceeded\'"+ *> return MaximumPayloadExceeded <|>++ stringCI "\'Invalid Subject\'"+ *> return InvalidSubject++parseSid :: Parser Sid+parseSid = decimal++mkInfoMessage :: [HandshakeMessageField] -> Parser Message+mkInfoMessage fields =+ INFO <$> asByteString (lookup "server_id" fields)+ <*> asByteString (lookup "version" fields)+ <*> asByteString (lookup "go" fields)+ <*> asByteString (lookup "host" fields)+ <*> asInt (lookup "port" fields)+ <*> asBool (lookup "auth_required" fields)+ <*> asBool (lookup "ssl_required" fields)+ <*> asBool (lookup "tls_required" fields)+ <*> asBool (lookup "tls_verify" fields)+ <*> asInt (lookup "max_payload" fields)++mkConnectMessage :: [HandshakeMessageField] -> Parser Message+mkConnectMessage fields =+ CONNECT <$> asBool (lookup "verbose" fields)+ <*> asBool (lookup "pedantic" fields)+ <*> asBool (lookup "ssl_required" fields)+ <*> asByteString (lookup "auth_token" fields)+ <*> asByteString (lookup "user" fields)+ <*> asByteString (lookup "pass" fields)+ <*> asByteString (lookup "name" fields)+ <*> asByteString (lookup "lang" fields)+ <*> asByteString (lookup "version" fields)++asByteString :: Maybe HandshakeMessageValue -> Parser (Maybe ByteString)+asByteString Nothing = return Nothing+asByteString (Just (String value)) = return (Just value)+asByteString _ = fail "Expected a ByteString"++asBool :: Maybe HandshakeMessageValue -> Parser (Maybe Bool)+asBool Nothing = return Nothing+asBool (Just (Bool value)) = return (Just value)+asBool _ = fail "Expected a boolean"++asInt :: Maybe HandshakeMessageValue -> Parser (Maybe Int)+asInt Nothing = return Nothing+asInt (Just (Int value)) = return (Just value)+asInt _ = fail "Expected an Int"++spacedMsgName :: ByteString -> Parser ()+spacedMsgName name = do+ msgName name+ singleSpace+{-# INLINE spacedMsgName #-}++singleSpace :: Parser ()+singleSpace = void space+{-# INLINE singleSpace #-}++newLine :: Parser ()+newLine = void $ string "\r\n"+{-# INLINE newLine #-}++msgName :: ByteString -> Parser ()+msgName = void . stringCI+{-# INLINE msgName #-}
+ src/Network/Nats/Message/Writer.hs view
@@ -0,0 +1,172 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+-- |+-- Module: Network.Nats.Message.Writer+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- Serialize NATS 'Message's to 'LBS.ByteString's.+module Network.Nats.Message.Writer+ ( writeMessage+ ) where++import Data.ByteString (ByteString)+import Data.ByteString.Builder+import Data.Monoid ((<>))+import Data.List (foldl', intersperse)++import qualified Data.ByteString.Lazy as LBS++import Network.Nats.Message.Message (Message (..), ProtocolError (..))+import Network.Nats.Types (Sid)++-- | Existentially quantified Field type, to allow for a polymorph+-- list of Fields. All fields with the contraint of beeing Writeable.+data Field = forall w. Writeable w => Field !w++-- | Helper class used for the writing of "handshake message" fields.+class Writeable w where+ write :: w -> Builder++-- | Instance for 'Bool'.+instance Writeable Bool where+ write False = byteString "false"+ write True = byteString "true"++-- | Instance for 'Int'.+instance Writeable Int where+ write = intDec++-- | Instance for 'ByteString'.+instance Writeable ByteString where+ write value = charUtf8 '\"' <> byteString value <> charUtf8 '\"'++-- | Translate a 'Message' value to a 'LBS.ByteString'.+writeMessage :: Message -> LBS.ByteString+writeMessage = toLazyByteString . writeMessage'++-- | Translate a Message value to a Builder.+writeMessage' :: Message -> Builder++-- The first of the handshake messages; INFO.+writeMessage' INFO {..} =+ let fields = foldl' writeField [] + [ ("\"server_id\"", Field <$> serverId)+ , ("\"version\"", Field <$> serverVersion)+ , ("\"go\"", Field <$> goVersion)+ , ("\"host\"", Field <$> serverHost)+ , ("\"port\"", Field <$> serverPort)+ , ("\"auth_required\"", Field <$> serverAuthRequired)+ , ("\"ssl_required\"", Field <$> serverSslRequired)+ , ("\"tls_required\"", Field <$> serverTlsRequired)+ , ("\"tls_verify\"", Field <$> serverTlsVerify)+ , ("\"max_payload\"", Field <$> maxPayload)+ ]+ fields' = intersperse (charUtf8 ',') $ reverse fields+ in mconcat $ byteString "INFO {":(fields' ++ [charUtf8 '}'])++-- The second of the handshake messages; CONNECT.+writeMessage' CONNECT {..} =+ let fields = foldl' writeField []+ [ ("\"verbose\"", Field <$> clientVerbose)+ , ("\"pedantic\"", Field <$> clientPedantic)+ , ("\"ssl_required\"", Field <$> clientSslRequired)+ , ("\"auth_token\"", Field <$> clientAuthToken)+ , ("\"user\"", Field <$> clientUser)+ , ("\"pass\"", Field <$> clientPass)+ , ("\"name\"", Field <$> clientName)+ , ("\"lang\"", Field <$> clientLang)+ , ("\"version\"", Field <$> clientVersion)+ ]+ fields' = intersperse (charUtf8 ',') $ reverse fields+ in mconcat $ byteString "CONNECT {":(fields' ++ [byteString "}\r\n"])++-- MSG message without a reply subject.+writeMessage' (MSG subject sid Nothing payload) =+ byteString "MSG " <> byteString subject <> charUtf8 ' '+ <> writeSid sid <> charUtf8 ' '+ <> int64Dec (LBS.length payload) <> byteString "\r\n"+ <> lazyByteString payload <> byteString "\r\n"++-- MSG message with a reply subject.+writeMessage' (MSG subject sid (Just reply) payload) =+ byteString "MSG " <> byteString subject <> charUtf8 ' '+ <> writeSid sid <> charUtf8 ' '+ <> byteString reply <> charUtf8 ' '+ <> int64Dec (LBS.length payload) <> byteString "\r\n"+ <> lazyByteString payload <> byteString "\r\n"++-- PUB message without a reply subject.+writeMessage' (PUB subject Nothing payload) =+ byteString "PUB " <> byteString subject <> charUtf8 ' '+ <> int64Dec (LBS.length payload) <> byteString "\r\n"+ <> lazyByteString payload <> byteString "\r\n"++-- PUB message with a reply subject.+writeMessage' (PUB subject (Just reply) payload) =+ byteString "PUB " <> byteString subject <> charUtf8 ' '+ <> byteString reply <> charUtf8 ' '+ <> int64Dec (LBS.length payload) <> byteString "\r\n"+ <> lazyByteString payload <> byteString "\r\n"++-- SUB message without a queue group.+writeMessage' (SUB subject Nothing sid) =+ byteString "SUB " <> byteString subject <> charUtf8 ' ' + <> writeSid sid <> byteString "\r\n"++-- SUB message with a queue group.+writeMessage' (SUB subject (Just queue) sid) =+ byteString "SUB " <> byteString subject <> charUtf8 ' '+ <> byteString queue <> charUtf8 ' '+ <> writeSid sid <> byteString "\r\n"++-- UNSUB message without auto-unsubscribe limit.+writeMessage' (UNSUB sid Nothing) =+ byteString "UNSUB " <> writeSid sid <> byteString "\r\n"++-- UNSUB message with auto-unsubscribe limit.+writeMessage' (UNSUB sid (Just maxMsgs)) =+ byteString "UNSUB " <> writeSid sid <> charUtf8 ' '+ <> intDec maxMsgs <> byteString "\r\n"++-- PING message.+writeMessage' PING = byteString "PING" <> byteString "\r\n"++-- PONG message.+writeMessage' PONG = byteString "PONG" <> byteString "\r\n"++-- Server acknowledge of a well-formed message.+writeMessage' OK = byteString "+OK\r\n"++-- | Server indication of a protocol, authorization, or other+-- runtime connection error.+writeMessage' (ERR pe) = byteString "-ERR " <> writePE pe <> "\r\n"++-- | The translate a Field to a Builder and prepend it to the list of+-- Builders.+writeField :: [Builder] -> (ByteString, Maybe Field) -> [Builder]+writeField xs (name, Just (Field value)) = + let x = byteString name <> charUtf8 ':' <> write value+ in x:xs++-- There's a Nothing Field. Just return the unmodified Builder list.+writeField xs (_, Nothing) = xs++-- | Translate a ProtocolError to a Builder.+writePE :: ProtocolError -> Builder+writePE UnknownProtocolOperation = byteString "\'Unknown Protocol Operation\'"+writePE AuthorizationViolation = byteString "\'Authorization Violation\'"+writePE AuthorizationTimeout = byteString "\'Authorization Timeout\'"+writePE ParserError = byteString "\'Parser Error\'"+writePE StaleConnection = byteString "\'Stale Connection\'"+writePE SlowConsumer = byteString "\'Slow Consumer\'"+writePE MaximumPayloadExceeded = byteString "\'Maximum Payload Exceeded\'"+writePE InvalidSubject = byteString "\'Invalid Subject\'"++-- | Translate a Sid to a Builder.+writeSid :: Sid -> Builder+writeSid = int64Dec
+ src/Network/Nats/Subscriber.hs view
@@ -0,0 +1,86 @@+-- |+-- Module: Network.Nats.Subscriber+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- Data structures and API for handling of subscribers.+module Network.Nats.Subscriber+ ( SubscriberMap+ , Subscriber (..)+ , newSubscriberMap+ , addSubscriber+ , addAsyncSubscriber+ , removeSubscriber+ , lookupSubscriber+ , subscribeMessages+ ) where++import Network.Nats.Types (MsgQueue (..), Msg, Sid)+import Network.Nats.Message.Message (Message (..))++import Control.Concurrent.STM ( TQueue, TVar, atomically, newTVarIO+ , newTQueueIO, modifyTVar, readTVar+ , readTVarIO+ )+import Data.HashMap.Strict (HashMap)++import qualified Data.HashMap.Strict as HM++-- | Map from 'Sid' to 'Subscriber'. Wrapped in a 'TVar'.+type SubscriberMap = TVar (HashMap Sid Subscriber)++-- | Data structure to describe a subscriber. Each subscriber caches+-- the SUB 'Message' used to define it. Needed when replaying+-- subscriptions at server reconnects.+data Subscriber+ = Subscriber !(TQueue Msg) !Message+ -- ^ A ordinary subscriber, which is just a 'TQueue' of 'Msg's.+ | AsyncSubscriber !(Msg -> IO ()) !Message+ -- ^ An asynchronous subscriber, with an IO action taking a+ -- 'Msg'.++-- | Create a new empty 'SubscriberMap'.+newSubscriberMap :: IO SubscriberMap+newSubscriberMap = newTVarIO HM.empty++-- | Add a new subscriber to the 'SubscriberMap'.+addSubscriber :: SubscriberMap -> Sid -> Message -> IO MsgQueue+addSubscriber subscriberMap sid msg = do+ queue <- newTQueueIO+ let sub = Subscriber queue msg+ atomically $ modifyTVar subscriberMap $ HM.insert sid sub+ return $ MsgQueue queue+{-# INLINE addSubscriber #-}++-- | Add a new, asynchronous, subscriber to the 'SubscriberMap'.+addAsyncSubscriber :: SubscriberMap -> Sid -> Message+ -> (Msg -> IO ()) -> IO ()+addAsyncSubscriber subscriberMap sid msg action = do+ let sub = AsyncSubscriber action msg+ atomically $ modifyTVar subscriberMap $ HM.insert sid sub+{-# INLINE addAsyncSubscriber #-}++-- | Remove a subscriber.+removeSubscriber :: SubscriberMap -> Sid -> IO ()+removeSubscriber subscriberMap sid =+ atomically $ modifyTVar subscriberMap $ HM.delete sid+{-# INLINE removeSubscriber #-}++-- | Try to lookup a subscriber.+lookupSubscriber :: SubscriberMap -> Sid -> IO (Maybe Subscriber)+lookupSubscriber subscriberMap sid =+ HM.lookup sid <$> atomically (readTVar subscriberMap)+{-# INLINE lookupSubscriber #-}++-- | Enumerate all subscriber SUB 'Message's from the 'SubscriberMap'.+subscribeMessages :: SubscriberMap -> IO [Message]+subscribeMessages subscriberMap =+ map extractMessage . HM.elems <$> readTVarIO subscriberMap++extractMessage :: Subscriber -> Message+extractMessage (Subscriber _ msg) = msg+extractMessage (AsyncSubscriber _ msg) = msg+{-# INLINE extractMessage #-}
+ src/Network/Nats/Types.hs view
@@ -0,0 +1,117 @@+{-# LANGUAGE DeriveAnyClass #-}+-- |+-- Module: Network.Nats.Types+-- Copyright: (c) 2016 Patrik Sandahl+-- License: MIT+-- Maintainer: Patrik Sandahl <patrik.sandahl@gmail.com>+-- Stability: experimental+-- Portability: portable+--+-- Base types for the library's API. JSON support is implemented+-- with "Data.Aeson".+module Network.Nats.Types+ ( Topic+ , Payload+ , Sid+ , QueueGroup+ , NatsException (..)+ , MsgQueue (..)+ , Msg (..)+ , topic+ , replyTo+ , sid+ , payload+ , jsonPayload+ , jsonPayload'+ ) where++import Control.Concurrent.STM (TQueue)+import Control.Exception (Exception)+import Data.Aeson (FromJSON, decode, decode')+import Data.Typeable (Typeable)+import GHC.Int (Int64)++import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as LBS++-- | The type of a topic where to publish, or to subscribe on. Type+-- alias for 'BS.ByteString'.+type Topic = BS.ByteString++-- | The type of a message payload. Type alias for 'LBS.ByteString'.+type Payload = LBS.ByteString++-- | The numeric id for a subscription. An id shall be unique within+-- a NATS client. The value of the id will be generated automatically+-- by the API. Type alias for 'Int64'.+type Sid = Int64++-- | A 'Topic' subscriber can be part of a queue group, an entity+-- for load balancing in NATS. Type alias for 'BS.ByteString'.+type QueueGroup = BS.ByteString++-- | Exceptions generated from within this library.+data NatsException+ = HandshakeException+ -- ^ An exception caused by errors during the NATS connection+ -- handshake. Will not popup on user level, only handled internally.++ | ConnectionGiveUpException+ -- ^ An exception thrown when all the configured connection+ -- attempts are consumed and the connection manager has been+ -- given up.++ | AuthorizationException+ -- ^ The NATS server currently connected to has said that there are+ -- authorization violations. Don't try to survive, just tell the user+ -- that there are such errors.++ | URIError !String+ -- ^ An exception caused by invalid URI strings given to the+ -- 'Network.Nats.withNats' function.+ deriving (Typeable, Show)++instance Exception NatsException++-- | A message queue, a queue of 'Msg's handled by a 'TQueue'.+newtype MsgQueue = MsgQueue (TQueue Msg)++-- | A NATS message as received by the user. The message itself is+-- opaque to the user, but the fields can be read by the API functions+-- 'topic', 'replyTo', 'sid', 'payload', 'jsonPayload' and+-- 'jsonPayload''+data Msg = Msg !Topic !(Maybe Topic) {-# UNPACK #-} !Sid !Payload+ deriving (Eq, Show)++-- | Read the complete topic on which a message was received.+topic :: Msg -> Topic+topic (Msg t _ _ _) = t+{-# INLINE topic #-}++-- | Read the reply-to topic from a received message.+replyTo :: Msg -> Maybe Topic+replyTo (Msg _ r _ _) = r+{-# INLINE replyTo #-}++-- | Read the subscription id for the subscription on which this message+-- was received.+sid :: Msg -> Sid+sid (Msg _ _ s _) = s+{-# INLINE sid #-}++-- | Read the raw payload from a received message.+payload :: Msg -> Payload+payload (Msg _ _ _ p) = p+{-# INLINE payload #-}++-- | Decode a message's payload as JSON. Is using 'decode' for+-- the decoding.+jsonPayload :: FromJSON a => Msg -> Maybe a+jsonPayload = decode . payload+{-# INLINE jsonPayload #-}++-- | Decode a message's payload as JSON. Is using 'decode'' for+-- the decoding.+jsonPayload' :: FromJSON a => Msg -> Maybe a+jsonPayload' = decode' . payload+{-# INLINE jsonPayload' #-}
+ test/CallbackTests.hs view
@@ -0,0 +1,71 @@+module CallbackTests+ ( callingCallbacks+ ) where++import Control.Concurrent.MVar+import Control.Monad (when)+import Data.Maybe (isNothing, isJust)+import System.Timeout (timeout)+import Test.HUnit++import Gnatsd+import Network.Nats++-- | Check that callbacks are called when needed at connection and+-- disconnection.+callingCallbacks :: Assertion+callingCallbacks = do+ connect <- newEmptyMVar+ disconnect <- newEmptyMVar+ let settings = defaultSettings+ { connectedTo = connected connect+ , disconnectedFrom = disconnected disconnect+ }+ g1 <- startGnatsd+ withNats settings [defaultURI] $ \_nats -> do+ + -- Wait for connection.+ c1 <- timeout oneSec $ takeMVar connect+ when (isNothing c1) $ do+ stopGnatsd g1+ assertFailure "Shall have connectedTo callback"++ -- Check that no disconnect has happen.+ d1 <- tryTakeMVar disconnect+ when (isJust d1) $ do+ stopGnatsd g1+ assertFailure "Shall have no disconnectedFrom callback"++ -- Shoot down gnatsd ...+ stopGnatsd g1++ -- Now a disconnect shall have happen.+ d2 <- timeout oneSec $ takeMVar disconnect+ when (isNothing d2) $+ assertFailure "Shall have disconnectedFrom callback"++ -- Start gnatsd again ...+ g2 <- startGnatsd++ -- Check that we have a new connect callback.+ c2 <- timeout oneSec $ takeMVar connect+ when (isNothing c2) $ do+ stopGnatsd g2+ assertFailure "Shall have connectedTo callback"++ -- Done. Close.+ stopGnatsd g2++oneSec :: Int+oneSec = 1000000++connected :: MVar () -> SockAddr -> IO ()+connected sync sockAddr = do+ putStrLn $ "Connected to: " ++ show sockAddr+ putMVar sync ()++disconnected :: MVar () -> SockAddr -> IO ()+disconnected sync sockAddr = do+ putStrLn $ "Disconnected from: " ++ show sockAddr+ putMVar sync ()+
+ test/Gnatsd.hs view
@@ -0,0 +1,55 @@+-- | Implement a proxy towards the NATS server daemon - gnatsd. Provide+-- functions to start and stop NATS. Will require the program gnatsd+-- to be found in the PATH environment variable.+module Gnatsd+ ( ProcessHandle+ , startGnatsd+ , stopGnatsd+ , withGnatsd+ , withGnatsdUP+ , defaultURI+ , userPasswordURI+ ) where++import Control.Concurrent (threadDelay)+import Control.Exception (bracket)+import Control.Monad (void)+import System.Process ( ProcessHandle+ , spawnProcess+ , terminateProcess+ , waitForProcess+ )++-- | Start gnatsd using its default settings (i.e. open port 4222+-- for traffic). Give the server a little while to start.+startGnatsd :: IO ProcessHandle+startGnatsd = spawnProcess "gnatsd" ["-DV"] <* waitAWhile++startGnatsdUP :: IO ProcessHandle+startGnatsdUP = + spawnProcess "gnatsd" ["-DV", "-user", "user", "-pass", "pass"]+ <* waitAWhile++-- | Stop the the gnatsd server. Wait for it to stop.+stopGnatsd :: ProcessHandle -> IO ()+stopGnatsd proc = do+ terminateProcess proc+ void $ waitForProcess proc++-- | Convenience function to wrap things in a bracket.+withGnatsd :: IO () -> IO ()+withGnatsd action = bracket startGnatsd stopGnatsd (const action)++-- | Convenience function to wrap things in a bracket. Gnatsd is started+-- with options to require user and password credentials.+withGnatsdUP :: IO () -> IO ()+withGnatsdUP action = bracket startGnatsdUP stopGnatsd (const action)++waitAWhile :: IO ()+waitAWhile = threadDelay 500000++defaultURI :: String+defaultURI = "nats://localhost:4222"++userPasswordURI :: String+userPasswordURI = "nats://user:pass@localhost:4222"
+ test/JsonTests.hs view
@@ -0,0 +1,74 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+module JsonTests+ ( recSingleJsonMessage+ , requestJsonMessage+ ) where++import Control.Monad (void)+import Data.Aeson+import Data.Maybe (fromJust)+import Data.Text (Text)+import GHC.Generics (Generic)+import Test.HUnit++import Gnatsd+import Network.Nats+ +data TestRec = TestRec+ { textVal :: !Text+ , intVal :: !Int+ } deriving (Eq, Generic, Show)++instance FromJSON TestRec+instance ToJSON TestRec++-- | Subscribe on a topic and receive one Json message through a queue.+-- Expect the received 'JsonMsg' to echo the published payload.+recSingleJsonMessage, recSingleJsonMessage' :: Assertion+recSingleJsonMessage = withGnatsd recSingleJsonMessage'++recSingleJsonMessage' =+ withNats defaultSettings [defaultURI] $ \nats -> do+ let topic' = "test"+ payload' = TestRec { textVal = "Some Text"+ , intVal = 42+ }++ (sid', queue) <- subscribe nats topic' Nothing+ publishJson nats topic' Nothing payload'++ -- Wait for the response ...+ msg <- nextMsg queue+ topic' @=? topic msg+ Nothing @=? replyTo msg+ sid' @=? sid msg+ payload' @=? (fromJust $ jsonPayload msg)++-- | Request a topic. Excersize both the requestJson api and the+-- subscribeAsyncJson api, as the handler will modify the given Json+-- record.+requestJsonMessage, requestJsonMessage' :: Assertion+requestJsonMessage = withGnatsd requestJsonMessage'++requestJsonMessage' =+ withNats defaultSettings [defaultURI] $ \nats -> do+ let topic' = "test"+ payload1 = TestRec { textVal = "Some Text"+ , intVal = 42+ }+ payload2 = TestRec { textVal = "Some Text"+ , intVal = 43+ }+ + -- Async handler that receive a TestRec and increments its+ -- intVal field before sending it back.+ void $ subscribeAsync nats topic' Nothing $+ \msg-> do+ let p = fromJust $ jsonPayload msg+ reply = p { intVal = intVal p + 1 }+ publishJson nats (fromJust $ replyTo msg) Nothing reply++ msg <- requestJson nats topic' payload1++ payload2 @=? (fromJust $ jsonPayload msg)
+ test/MessageProps.hs view
@@ -0,0 +1,141 @@+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module MessageProps+ ( encodeDecodeMessage+ ) where++import Data.Attoparsec.ByteString.Char8+import Data.ByteString.Char8 (ByteString)+import Test.QuickCheck++import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy.Char8 as LBS++import Network.Nats.Types (Sid)+import Network.Nats.Message.Message (Message (..), ProtocolError (..))+import Network.Nats.Message.Parser (parseMessage)+import Network.Nats.Message.Writer (writeMessage)++-- | Arbitrary instance for ProtocolError.+instance Arbitrary ProtocolError where+ arbitrary = elements [ minBound .. ]++-- | Arbitrary instance for Message.+instance Arbitrary Message where+ arbitrary = oneof [ arbitraryInfo+ , arbitraryConnect+ , arbitraryMsg+ , arbitraryPub+ , arbitrarySub+ , arbitraryUnsub+ , arbitraryPingPong+ , arbitraryOk+ , arbitraryErr+ ]+ +-- | Arbitrary generation of Info messages.+arbitraryInfo :: Gen Message+arbitraryInfo =+ INFO <$> perhaps valueString+ <*> perhaps valueString+ <*> perhaps valueString+ <*> perhaps valueString+ <*> perhaps posInt+ <*> arbitrary+ <*> arbitrary+ <*> arbitrary+ <*> arbitrary+ <*> perhaps posInt++-- | Arbitrary generation of Connect messages.+arbitraryConnect :: Gen Message+arbitraryConnect =+ CONNECT <$> arbitrary+ <*> arbitrary+ <*> arbitrary+ <*> perhaps valueString+ <*> perhaps valueString+ <*> perhaps valueString+ <*> perhaps valueString+ <*> perhaps valueString+ <*> perhaps valueString++-- | Arbitrary generation of Msg messages.+arbitraryMsg :: Gen Message+arbitraryMsg = MSG <$> alnumString+ <*> sid+ <*> perhaps alnumString+ <*> payloadString++-- | Arbitrary generation of Pub messages.+arbitraryPub :: Gen Message+arbitraryPub = PUB <$> alnumString+ <*> perhaps alnumString+ <*> payloadString++-- | Arbitrary generation of Sub messages.+arbitrarySub :: Gen Message+arbitrarySub = SUB <$> alnumString + <*> perhaps alnumString + <*> sid++-- | Arbitrary generation of Sub messages.+arbitraryUnsub :: Gen Message+arbitraryUnsub = UNSUB <$> sid+ <*> perhaps posInt++-- | Arbitrary generation of Ping/Pong messages.+arbitraryPingPong :: Gen Message+arbitraryPingPong = oneof [ pure PING, pure PONG ]++-- | Arbitrary generation of Ok messages.+arbitraryOk :: Gen Message+arbitraryOk = pure OK++-- | Arbitrary generation of Err messages.+arbitraryErr :: Gen Message+arbitraryErr = ERR <$> arbitrary++-- | Test by write a Message to ByteString, and parse it back again.+encodeDecodeMessage :: Message -> Bool+encodeDecodeMessage msg =+ let asByteString = LBS.toStrict $ writeMessage msg+ in verify (parse parseMessage asByteString)+ where+ verify :: IResult ByteString Message -> Bool+ verify result =+ case result of+ (Done _ msg2) -> msg == msg2+ (Partial g) -> verify (g "")+ _ -> False++-- | Generate Maybe's for the tailor made generators.+perhaps :: Gen a -> Gen (Maybe a)+perhaps gen = oneof [ return Nothing, Just <$> gen ]++-- | Generate a ByteString which not contains any quote characters.+valueString :: Gen ByteString+valueString =+ BS.pack <$> listOf (elements selection)+ where+ selection :: [Char]+ selection = ['a'..'z'] ++ ['A'..'Z'] ++ ['0'..'9'] ++ "+-_!?*(){} "++payloadString :: Gen LBS.ByteString+payloadString = LBS.pack <$> listOf arbitrary++-- | Generate a non-empty ByteString which is plain alphanumeric.+alnumString :: Gen ByteString+alnumString =+ BS.pack <$> listOf1 (elements selection)+ where+ selection :: [Char]+ selection = ['a'..'z'] ++ ['A'..'Z'] ++ ['0'..'9'] ++-- | Generate a positive integer. Can be max size of Int.+posInt :: Gen Int+posInt = choose (0, maxBound)++-- | Generate a Sid+sid :: Gen Sid+sid = choose (0, maxBound)
+ test/NatsTests.hs view
@@ -0,0 +1,133 @@+{-# LANGUAGE OverloadedStrings #-}+module NatsTests+ ( recSingleMessage+ , recSingleMessageAsync+ , recMessagesWithTmo+ , requestMessage+ , unsubscribeToTopic+ ) where++import Control.Concurrent.MVar+import Control.Monad+import Data.Maybe (fromJust)+import System.Timeout (timeout)+import Test.HUnit++import Gnatsd+import Network.Nats++-- Subscribe on a topic and receive one message through a queue. Expect+-- the received 'Msg' to echo the published payload.+recSingleMessage, recSingleMessage' :: Assertion+recSingleMessage = withGnatsd recSingleMessage'++recSingleMessage' =+ withNats defaultSettings [defaultURI] $ \nats -> do+ let topic' = "test"+ replyTo' = Nothing+ payload' = "test message"++ (sid', queue) <- subscribe nats topic' Nothing+ publish nats topic' replyTo' payload'++ -- Wait for the message ...+ msg <- nextMsg queue+ sid' @=? sid msg+ topic' @=? topic msg+ replyTo' @=? replyTo msg+ payload' @=? payload msg++-- Subscribe on a topic and receive one message asynchronously. Expect+-- the message receiver to receive the expected 'Msg' data.+recSingleMessageAsync, recSingleMessageAsync' :: Assertion+recSingleMessageAsync = withGnatsd recSingleMessageAsync'++recSingleMessageAsync' =+ void $ withNats defaultSettings [defaultURI] $ \nats -> do+ let topic' = "test"+ replyTo' = Nothing+ payload' = "test message"+ recData <- newEmptyMVar+ sid' <- subscribeAsync nats topic' Nothing $ receiver recData+ publish nats topic' replyTo' payload'++ -- Wait for the MVar ...+ msg <- takeMVar recData+ sid' @=? sid msg+ topic' @=? topic msg+ replyTo' @=? replyTo msg+ payload' @=? payload msg+ where+ receiver :: MVar Msg -> Msg -> IO ()+ receiver = putMVar++-- | Subscribe to a topic, and send two messages to the topic. When+-- reading trying to read a third message from the queue, it shall+-- block. To handle the blocking 'timeout' is used.+recMessagesWithTmo, recMessagesWithTmo' :: Assertion+recMessagesWithTmo = withGnatsd recMessagesWithTmo'++recMessagesWithTmo' =+ void $ withNats defaultSettings [defaultURI] $ \nats -> do+ let topic' = "test"+ payload1 = "test message"+ payload2 = "test message 2"++ (sid', queue) <- subscribe nats topic' Nothing+ publish nats topic' Nothing payload1+ publish nats topic' Nothing payload2++ -- Wait for the messages ...+ Just msg1 <- timeout oneSec $ nextMsg queue+ sid' @=? sid msg1+ topic' @=? topic msg1+ payload1 @=? payload msg1++ Just msg2 <- timeout oneSec $ nextMsg queue+ sid' @=? sid msg2+ topic' @=? topic msg2+ payload2 @=? payload msg2++ -- This time there shall be a timeout.+ reply <- timeout oneSec $ nextMsg queue+ Nothing @=? reply++-- | Request a topic. Request is a convenience function, it subscribe,+-- publish a message and waits for a reply. The replyTo topic is+-- random generated by the function.+requestMessage, requestMessage' :: Assertion+requestMessage = withGnatsd requestMessage'++requestMessage' =+ withNats defaultSettings [defaultURI] $ \nats -> do+ let topic' = "test"+ payload' = "echo me"++ -- Register a handler for serving the request.+ void $ subscribeAsync nats topic' Nothing $+ \msg -> publish nats (fromJust $ replyTo msg) + Nothing (payload msg)++ -- Make the request and compare the reply payload.+ msg <- request nats topic' payload'+ payload' @=? payload msg++-- | Subscribe to a topic, then unsubscribe to it before publishing+-- a message. No message shall show up in the queue.+unsubscribeToTopic, unsubscribeToTopic' :: Assertion+unsubscribeToTopic = withGnatsd unsubscribeToTopic'++unsubscribeToTopic' =+ void $ withNats defaultSettings [defaultURI] $ \nats -> do+ let topic' = "test"++ (sid', queue) <- subscribe nats topic' Nothing+ unsubscribe nats sid' Nothing+ publish nats topic' Nothing "shall never arrive"++ -- As the test case is unsubscribed, nothing shall show up.+ reply <- timeout oneSec $ nextMsg queue+ Nothing @=? reply++oneSec :: Int+oneSec = 1000000
+ test/ReconnectionTests.hs view
@@ -0,0 +1,112 @@+{-# LANGUAGE OverloadedStrings #-}+module ReconnectionTests+ ( subscribeAndReconnect+ , connectionGiveUp+ , authorizationFail+ , authorizationSuccess+ ) where++import Control.Concurrent+import Control.Exception+import Control.Monad (when)+import Data.Maybe (isNothing)+import System.Timeout (timeout)+import Test.HUnit++import Gnatsd+import Network.Nats++-- | Subscribe to three topics. Take down the current NATS connection,+-- force a reconnection, publish to the topics and test that the+-- messages are received.+subscribeAndReconnect :: Assertion+subscribeAndReconnect = do+ connect <- newEmptyMVar+ let settings = defaultSettings+ { connectedTo = connected connect+ }+ g1 <- startGnatsd+ withNats settings [defaultURI] $ \nats -> do++ -- Ok, just wait for the first connect.+ c1 <- timeout oneSec $ takeMVar connect+ when (isNothing c1) $ do+ stopGnatsd g1+ assertFailure "Shall have connected :-("++ -- Subscribe to three topics.+ (_, queue1) <- subscribe nats "topic1" Nothing+ (_, queue2) <- subscribe nats "topic2" Nothing+ (_, queue3) <- subscribe nats "topic3" Nothing++ -- Let things have the time to go on network.+ threadDelay 100000++ -- Take down gnatsd.+ stopGnatsd g1++ -- Start a new instance.+ g2 <- startGnatsd++ -- Publish to the topics.+ publish nats "topic1" Nothing "payload1"+ publish nats "topic2" Nothing "payload2"+ publish nats "topic3" Nothing "payload3"++ m1 <- timeout oneSec $ nextMsg queue1+ when (isNothing m1) $ do+ stopGnatsd g2+ assertFailure "Shall have received a message (m1)"++ m2 <- timeout oneSec $ nextMsg queue2+ when (isNothing m2) $ do+ stopGnatsd g2+ assertFailure "Shall have received a message (m2)"++ m3 <- timeout oneSec $ nextMsg queue3+ when (isNothing m3) $ do+ stopGnatsd g2+ assertFailure "Shall have received a message (m3)"++ -- Survived this far? Success, but terminate gnatsd before quitting.+ stopGnatsd g2++connected :: MVar () -> SockAddr -> IO ()+connected sync _ = putMVar sync ()++-- | Test that when the reconnection logic gives up, a+-- ConnectionGiveUpException is thrown out of the withNats function.+connectionGiveUp :: Assertion+connectionGiveUp =+ (\ConnectionGiveUpException -> return ()) `handle` do+ withNats defaultSettings [defaultURI] $ \_ ->+ threadDelay oneSec++ assertFailure "Shall never come here!"++-- | Try to connect a server to which the test don't have the+-- credentials. Shall get a AuthorizationException.+authorizationFail :: Assertion+authorizationFail = withGnatsdUP authorizationFail'++authorizationFail' :: Assertion+authorizationFail' =+ (\AuthorizationException -> return ()) `handle` do+ withNats defaultSettings [defaultURI] $ \_ ->+ threadDelay oneSec++ assertFailure "Shall never come here!"++-- | Connect to a server with credentials. Shall work, no+-- exceptions thrown.+authorizationSuccess :: Assertion+authorizationSuccess = withGnatsdUP authorizationSuccess'++authorizationSuccess' :: Assertion+authorizationSuccess' =+ withNats defaultSettings [userPasswordURI] $ \_ ->+ threadDelay oneSec++oneSec :: Int+oneSec = 1000000+
+ test/Spec.hs view
@@ -0,0 +1,63 @@+module Main+ ( main+ ) where++import Test.Framework (Test, defaultMain, testGroup)+import Test.Framework.Providers.HUnit (testCase)+import Test.Framework.Providers.QuickCheck2 (testProperty)++import CallbackTests (callingCallbacks)+import JsonTests (recSingleJsonMessage, requestJsonMessage)+import NatsTests ( recSingleMessage+ , recSingleMessageAsync+ , recMessagesWithTmo+ , requestMessage+ , unsubscribeToTopic+ )+import ReconnectionTests ( subscribeAndReconnect, connectionGiveUp+ , authorizationFail, authorizationSuccess+ )+import MessageProps (encodeDecodeMessage)++main :: IO ()+main = defaultMain testSuite++testSuite :: [Test]+testSuite =+ [ testGroup "Message property tests"+ [ testProperty "Encoding and decoding of Message"+ encodeDecodeMessage+ ]+ , testGroup "Plain (non-JSON) NATS API tests"+ [ testCase "Reception of a single message"+ recSingleMessage+ , testCase "Async reception of a single message"+ recSingleMessageAsync+ , testCase "Reception of two messages, with timeout"+ recMessagesWithTmo+ , testCase "Reception of a message using request"+ requestMessage+ , testCase "Unsubscribe to a topic before publishing"+ unsubscribeToTopic+ ]+ , testGroup "JSON NATS API tests"+ [ testCase "Reception of a single Json message"+ recSingleJsonMessage+ , testCase "Reception of (modified) Json message using requestJson"+ requestJsonMessage+ ]+ , testGroup "Callback tests"+ [ testCase "Test calling of connectedTo and disconnectedFrom"+ callingCallbacks+ ]+ , testGroup "Connection tests"+ [ testCase "Test that subscription works after reconnect"+ subscribeAndReconnect+ , testCase "Test that exception is thrown when giving up"+ connectionGiveUp+ , testCase "Test that exceptions is thrown when fail authorization"+ authorizationFail+ , testCase "Test that authorization succeed."+ authorizationSuccess+ ]+ ]