nats-client (empty) → 0.1.0.0
raw patch · 13 files changed
+884/−0 lines, 13 filesdep +aesondep +attoparsecdep +basesetup-changed
Dependencies added: aeson, attoparsec, base, bytestring, containers, criterion, data-default, exceptions, hedgehog, hslogger, monad-control, mtl, nats-client, network, random, resource-pool, test-framework, text, time, transformers
Files
- LICENSE +30/−0
- README.md +5/−0
- Setup.hs +2/−0
- bench/Bench.hs +7/−0
- example/ExampleClient.hs +21/−0
- nats-client.cabal +95/−0
- src/Network/Nats/Client.hs +204/−0
- src/Network/Nats/Protocol.hs +143/−0
- src/Network/Nats/Protocol/Message.hs +102/−0
- src/Network/Nats/Protocol/Types.hs +106/−0
- test/Network/Nats/Protocol/Tests.hs +41/−0
- test/Network/Nats/Test/Generators.hs +112/−0
- test/Spec.hs +16/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Author name here (c) 2017++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Author name here nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,5 @@+# nats-client+A Haskell client library for [NATS](https://nats.io).++## Usage+See [example](./example) for a sample client.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ bench/Bench.hs view
@@ -0,0 +1,7 @@+module Main where++import Criterion.Main++main = defaultMain benches++benches = []
+ example/ExampleClient.hs view
@@ -0,0 +1,21 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Control.Monad (forever)+import Network.Nats.Client+import qualified Data.ByteString as B++loop :: NatsClient -> Subject -> IO ()+loop client subj = B.getLine >>= publish client subj++main :: IO ()+main = withNats connectionSettings $ \client -> do+ case createSubject "foo.*" of+ Left err -> putStrLn $ "Invalid subject " ++ err+ Right subj -> do+ subId <- subscribe client subj (\m -> putStrLn $ "RECV: " ++ (show m)) Nothing+ unsubscribe client subId (Just 90)+ forever $ loop client subj+ where+ connectionSettings = defaultConnectionSettings { host = "demo.nats.io" }
+ nats-client.cabal view
@@ -0,0 +1,95 @@+name: nats-client+version: 0.1.0.0+synopsis: Another Haskell client for NATS (https://nats.io)+description: A simple client library for the NATS messaging protocol+homepage: https://bitbucket.org/jpgneves/nats-client#readme+license: BSD3+license-file: LICENSE+author: João Neves+maintainer: sevenjp@gmail.com+copyright: 2017 João Neves+category: Network+build-type: Simple+extra-source-files: README.md+cabal-version: >=1.10++flag fail-on-warning+ description: Enables -Werror+ default: False+ manual: True++library+ hs-source-dirs: src+ exposed-modules: Network.Nats.Client+ , Network.Nats.Protocol+ , Network.Nats.Protocol.Types+ other-modules: Network.Nats.Protocol.Message+ if flag(fail-on-warning) {+ ghc-options: -Wall -Werror -Wredundant-constraints+ } else {+ ghc-options: -Wall -Wredundant-constraints+ }+ build-depends: base >= 4.7 && < 5+ , aeson+ , attoparsec+ , bytestring+ , containers+ , data-default+ , exceptions+ , hslogger+ , monad-control+ , network+ , random+ , resource-pool+ , text+ , transformers+ default-language: Haskell2010++executable "nats-client"+ hs-source-dirs: example+ main-is: ExampleClient.hs+ build-depends: base >=4.7 && < 5+ , bytestring+ , nats-client+ , network+ if flag(fail-on-warning) {+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall -Werror+ } else {+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall+ }+ default-language: Haskell2010++test-suite nats-client-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ other-modules: Network.Nats.Protocol.Tests+ , Network.Nats.Test.Generators+ if flag(fail-on-warning) {+ ghc-options: -Wall -Werror -Wredundant-constraints+ } else {+ ghc-options: -Wall -Wredundant-constraints+ }+ build-depends: base >=4.7 && < 5+ , aeson+ , bytestring+ , mtl+ , nats-client+ , test-framework+ , hedgehog+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++benchmark bench-nats-client+ type: exitcode-stdio-1.0+ hs-source-dirs: bench+ main-is: Bench.hs+ build-depends: base >=4.7 && < 5+ , criterion+ , time+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++source-repository head+ type: git+ location: https://bitbucket.com/jpgneves/nats-client
+ src/Network/Nats/Client.hs view
@@ -0,0 +1,204 @@+{-|+Module : Network.Nats.Client+Description : Main interface to the NATS client library+-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}++module Network.Nats.Client (+ defaultConnectionSettings+ , defaultNatsHost+ , defaultNatsPort+ , connect+ , publish+ , subscribe+ , unsubscribe+ , withNats+ , createSubject+ , ConnectionSettings(..)+ , MessageHandler+ , NatsClient+ , Subject+ , Message (..)+ ) where++import Control.Concurrent (forkFinally)+import Control.Concurrent.MVar+import Control.Exception hiding (catch, bracket)+import Control.Monad.Catch+import Control.Monad.IO.Class+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.IORef+import Data.Pool+import Data.Typeable+import Network+import Network.Nats.Protocol+import System.IO (Handle, BufferMode(LineBuffering), hClose, hSetBuffering)+import System.Log.Logger+import System.Random+import System.Timeout+import qualified Data.Map as M+import qualified Data.Set as S+import qualified Data.ByteString.Char8 as BS++-- | A NATS client. See 'connect'.+data NatsClient = NatsClient { connections :: Pool NatsServerConnection+ , settings :: ConnectionSettings+ , subscriptions :: MVar (M.Map SubscriptionId NatsServerConnection)+ , servers :: MVar (S.Set (HostName, PortNumber))+ }++data NatsServerConnection = NatsServerConnection { natsHandle :: Handle+ , natsInfo :: NatsServerInfo+ , maxMessages :: IORef (Maybe Int)+ }++-- | NATS server connection settings+data ConnectionSettings = ConnectionSettings { host :: HostName+ , port :: PortNumber+ } deriving (Show)++-- |'Message' handling function+type MessageHandler = (Message -> IO ())++data NatsError = ConnectionFailure+ | ConnectionTimeout+ | InvalidServerBanner String+ | PayloadTooLarge String+ deriving (Show, Typeable)++instance Exception NatsError++-- | Convenience connection defaults using 'defaultNatsHost' and 'defaultNatsPort'+defaultConnectionSettings :: ConnectionSettings+defaultConnectionSettings = ConnectionSettings defaultNatsHost defaultNatsPort++-- | Default NATS host to connect to+defaultNatsHost :: HostName+defaultNatsHost = "127.0.0.1"++-- | Default port of the NATS server to connect to+defaultNatsPort :: PortNumber+defaultNatsPort = 4222 :: PortNumber++makeNatsServerConnection :: (MonadThrow m, Connection m) => ConnectionSettings -> MVar (S.Set (HostName, PortNumber)) -> m NatsServerConnection+makeNatsServerConnection (ConnectionSettings ho po) srvs = do+ mh <- liftIO $ timeout defaultTimeout $ connectTo ho $ PortNumber po+ case mh of+ Nothing -> do+ liftIO $ warningM "Network.Nats.Client" $ "Timed out connecting to server: " ++ (show ho) ++ ":" ++ (show po)+ throwM ConnectionTimeout+ Just h -> do+ liftIO $ hSetBuffering h LineBuffering+ nInfo <- liftIO $ receiveServerBanner h+ liftIO $ infoM "Network.Nats.Client" $ "Received server info " ++ show nInfo+ case nInfo of+ Right info -> do+ sendConnect h defaultConnectionOptions+ maxMsgs <- liftIO $ newIORef Nothing+ _subsc <- liftIO $ newIORef Nothing+ liftIO $ modifyMVarMasked_ srvs $ \ss -> return $ S.insert (ho, po) ss+ return $ NatsServerConnection { natsHandle = h, natsInfo = info, maxMessages = maxMsgs }+ Left err -> throwM $ InvalidServerBanner err++destroyNatsServerConnection :: Connection m => NatsServerConnection -> m ()+destroyNatsServerConnection conn = liftIO $ hClose (natsHandle conn)++-- | Connect to a NATS server+connect :: (MonadThrow m, MonadIO m) => ConnectionSettings -> Int -> m NatsClient+connect s max_connections = do+ srvs <- liftIO $ newMVar S.empty+ connpool <- liftIO $ createPool (makeNatsServerConnection s srvs) destroyNatsServerConnection 1 300 max_connections+ subs <- liftIO $ newMVar M.empty+ return $ NatsClient { connections = connpool, settings = s, subscriptions = subs, servers = srvs }++-- | Disconnect from a NATS server+disconnect :: (MonadIO m) => NatsClient -> m ()+disconnect conn = liftIO $ destroyAllResources (connections conn)++-- | Perform a computation with a NATS connection+withNats :: (MonadMask m, MonadIO m) => ConnectionSettings -> (NatsClient -> m b) -> m b+withNats connectionSettings f = bracket (connect connectionSettings 10) disconnect f++-- | Publish a 'BS.ByteString' to 'Subject'+publish :: (MonadThrow m, MonadIO m, MonadBaseControl IO m) => NatsClient -> Subject -> BS.ByteString -> m ()+publish conn subj msg = withResource (connections conn) $ doPublish subj msg++doPublish :: (MonadThrow m, MonadIO m) => Subject -> BS.ByteString -> NatsServerConnection -> m ()+doPublish subj msg conn = do+ case payload_length > (maxPayloadSize (natsInfo conn)) of+ True -> throwM $ PayloadTooLarge $ "Size: " ++ show payload_length ++ ", Max: " ++ show (maxPayloadSize (natsInfo conn))+ False -> liftIO $ sendPub sock subj payload_length msg+ where sock = (natsHandle conn)+ payload_length = BS.length msg++-- | Subscribe to a 'Subject' processing 'Message's via a 'MessageHandler'. Returns a 'SubscriptionId' used to cancel subscriptions+subscribe :: MonadIO m => NatsClient -> Subject -> MessageHandler -> Maybe QueueGroup -> m SubscriptionId+subscribe conn subj callback _qgroup = do+ (c, pool) <- liftIO $ takeResource (connections conn)+ subId <- liftIO $ generateSubscriptionId 5+ let sock = (natsHandle c)+ max_payload = (maxPayloadSize (natsInfo c))+ maxMsgs = (maxMessages c)+ liftIO $ sendSub sock subj subId Nothing+ _ <- liftIO $ forkFinally (connectionLoop sock max_payload callback maxMsgs) $ handleCompletion sock (connections conn) pool c+ liftIO $ modifyMVarMasked_ (subscriptions conn) $ \m -> return $ M.insert subId c m+ return subId++-- | Unsubscribe to a 'SubjectId' (returned by 'subscribe'), with an optional max amount of additional messages to listen to+unsubscribe :: (MonadIO m, MonadBaseControl IO m) => NatsClient -> SubscriptionId -> Maybe Int -> m ()+unsubscribe conn subId msgs@(Just maxMsgs) = do+ liftIO $ withMVarMasked (subscriptions conn) $ \m -> doUnsubscribe m subId maxMsgs+ withResource (connections conn) $ \s -> liftIO $ sendUnsub (natsHandle s) subId msgs+unsubscribe conn subId msgs@Nothing = do+ liftIO $ withMVarMasked (subscriptions conn) $ \m -> doUnsubscribe m subId 0+ withResource (connections conn) $ \s -> liftIO $ sendUnsub (natsHandle s) subId msgs++doUnsubscribe :: M.Map SubscriptionId NatsServerConnection -> SubscriptionId -> Int -> IO ()+doUnsubscribe m subId maxMsgs = do+ case M.lookup subId m of+ Nothing ->+ warningM "Network.Nats.Client" $ "Could not find subscription " ++ (show subId)+ Just c -> do+ atomicWriteIORef (maxMessages c) (Just maxMsgs)+++handleCompletion :: Handle -> Pool NatsServerConnection -> LocalPool NatsServerConnection -> NatsServerConnection -> Either SomeException b -> IO ()+handleCompletion h pool lpool conn (Left exn) = do+ warningM "Network.Nats.Client" $ "Connection closed: " ++ (show exn)+ hClose h+ destroyResource pool lpool conn+handleCompletion _ _ lpool conn _ = do+ debugM "Network.Nats.Client" "Subscription finished"+ atomicWriteIORef (maxMessages conn) Nothing+ putResource lpool conn++connectionLoop :: Handle -> Int -> (Message -> IO ()) -> IORef (Maybe Int) -> IO ()+connectionLoop h max_payload f maxMsgsRef = do+ maxMsgs <- readIORef maxMsgsRef+ case maxMsgs of+ Just 0 -> return ()+ _ ->+ receiveMessage h max_payload >>= (\m -> handleMessage h f m maxMsgs) >>= atomicWriteIORef maxMsgsRef >> connectionLoop h max_payload f maxMsgsRef++-- | Attempt to create a 'Subject' from a 'BS.ByteString'+createSubject :: BS.ByteString -> Either String Subject+createSubject = parseSubject++generateSubscriptionId :: Int -> IO SubscriptionId+generateSubscriptionId idLength = do+ gen <- getStdGen+ return $ SubscriptionId $ BS.pack $ take idLength $ (randoms gen :: [Char])++handleMessage :: Handle -> (Message -> IO ()) -> Message -> Maybe Int -> IO (Maybe Int)+handleMessage h _ Ping m = do+ sendPong h+ return m+handleMessage _ f msg@(Message _) maxMsgs = do+ f msg+ case maxMsgs of+ Nothing -> return Nothing+ (Just n) -> return $ Just (n - 1)+handleMessage _ _ msg m = do+ warningM "Network.Nats.Client" $ "Received " ++ (show msg)+ return m
+ src/Network/Nats/Protocol.hs view
@@ -0,0 +1,143 @@+{-|+Module : Network.Nats.Protocol+Description : Implementation of the NATS client protocol+-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedStrings #-}++module Network.Nats.Protocol ( Connection (..)+ , Subject(..)+ , SubscriptionId(..)+ , QueueGroup+ , defaultConnectionOptions+ , defaultTimeout+ , receiveMessage+ , receiveServerBanner+ , sendConnect+ , sendPong+ , sendPub+ , sendSub+ , sendUnsub+ , module Network.Nats.Protocol.Message+ , module Network.Nats.Protocol.Types+ ) where++import Control.Monad.Catch+import Control.Monad.IO.Class (MonadIO)+import Data.Aeson (encode)+import Data.ByteString.Builder+import Data.Default (def)+import Data.Monoid+import Network.Nats.Protocol.Message+import Network.Nats.Protocol.Types+import System.IO (Handle)+import qualified Data.ByteString.Char8 as BS++-- | Type for representing Queue Groups, used to implement round-robin receivers+newtype QueueGroup = QueueGroup BS.ByteString deriving Show++-- | Connection context for abstracting away IO+class (Monad m, MonadIO m) => Connection m where+ receiveRawMessage :: Handle -> Int -> m BS.ByteString+ sendRawMessage :: Handle -> Builder -> m ()++-- | Default IO implementation using bytestrings+instance Connection IO where+ receiveRawMessage = BS.hGetSome+ sendRawMessage = hPutBuilder++-- | Default client connection options, for convenience.+defaultConnectionOptions :: NatsConnectionOptions+defaultConnectionOptions = def NatsConnectionOptions++-- | Default client timeout, in millisseconds+defaultTimeout :: Int+defaultTimeout = 1000000++-- | Sendable commands+data Command where+ Connect :: NatsConnectionOptions -> Command+ Publish :: Subject -> BS.ByteString -> Command+ Subscribe :: Subject -> SubscriptionId -> Maybe QueueGroup -> Command+ Unsubscribe :: SubscriptionId -> Maybe Int -> Command+ Pong :: Command+ deriving (Show)++-- | Render a Command into a bytestring builder+render :: Command -> Builder+render (Connect opts) =+ stringUtf8 "CONNECT "+ <> lazyByteString (encode opts)+ <> spaceBuilder+ <> byteString lineTerminator+render (Publish subj payload) =+ stringUtf8 "PUB "+ <> renderSubject subj+ <> spaceBuilder+ <> renderPayload payload+ <> byteString lineTerminator+render (Subscribe subj subId _qgroup) =+ stringUtf8 "SUB "+ <> renderSubject subj+ <> spaceBuilder+ <> renderSubscriptionId subId+ <> spaceBuilder+ <> byteString lineTerminator+render (Unsubscribe subId _maxMsgs) =+ stringUtf8 "UNSUB "+ <> renderSubscriptionId subId <> spaceBuilder+ <> byteString lineTerminator+render Pong =+ stringUtf8 "PONG "+ <> byteString lineTerminator++spaceBuilder :: Builder+spaceBuilder = charUtf8 ' '++lineTerminator :: BS.ByteString+lineTerminator = "\r\n"++renderSubject :: Subject -> Builder+renderSubject (Subject s) = byteString s++renderSubscriptionId :: SubscriptionId -> Builder+renderSubscriptionId (SubscriptionId i) = byteString i++renderPayload :: BS.ByteString -> Builder+renderPayload p = intDec (BS.length p) <> byteString lineTerminator <> byteString p++sendCommand :: Connection m => Handle -> Command -> m ()+sendCommand h cmd = sendRawMessage h $ render cmd++-- | Receive the initial server banner from an INFO message, or an error message if it cannot be parsed.+receiveServerBanner :: Connection m => Handle -> m (Either String NatsServerInfo)+receiveServerBanner h = do+ receiveRawMessage h maxBytes >>= return . parseServerBanner+ where maxBytes = 10240++-- | Receive a 'Message' from the server+receiveMessage :: (MonadThrow m, Connection m) => Handle -> Int -> m Message+receiveMessage h maxBytes = do+ m <- receiveRawMessage h maxBytes+ parseMessage m++-- | Send a CONNECT message to the server+sendConnect :: Connection m => Handle -> NatsConnectionOptions -> m ()+sendConnect h opts = sendCommand h $ Connect opts++-- | Send a publish request to the server+sendPub :: Connection m => Handle -> Subject -> Int -> BS.ByteString -> m ()+sendPub h subj _payloadLen payload = sendCommand h $ Publish subj payload++-- | Send a Subscription request to a 'Subject', with a 'SubscriptionId' and optionally a 'QueueGroup'.+sendSub :: Connection m => Handle -> Subject -> SubscriptionId -> Maybe QueueGroup -> m ()+sendSub h subj subId qgroup = sendCommand h $ Subscribe subj subId qgroup++-- | Send an unsubscription request with a 'SubscriptionId' and optionally a maximum number of messages that will still be listened to.+sendUnsub :: Connection m => Handle -> SubscriptionId -> Maybe Int -> m ()+sendUnsub h subId max_msgs = sendCommand h $ Unsubscribe subId max_msgs+ +-- | Send a PONG message to the server, typically in reply to a PING challenge.+sendPong :: Connection m => Handle -> m ()+sendPong h = sendCommand h $ Pong
+ src/Network/Nats/Protocol/Message.hs view
@@ -0,0 +1,102 @@+{-|+Module : Network.Nats.Protocol.Message+Description: Message definitions and utilities for the NATS protocol+-}+{-# LANGUAGE OverloadedStrings #-}++module Network.Nats.Protocol.Message ( Message(..)+ , parseMessage+ , parseServerBanner+ , parseSubject+ ) where++import Control.Applicative ((<|>))+import Control.Monad.Catch+import Data.Aeson (eitherDecodeStrict)+import Network.Nats.Protocol.Types+import qualified Data.Attoparsec.ByteString.Char8 as A+import qualified Data.ByteString.Char8 as BS++-- | Messages received from the NATS server+data Message = Message BS.ByteString -- ^ A published message, containing a payload+ | OKMsg -- ^ Acknowledgment from server after a client request+ | ErrorMsg BS.ByteString -- ^ Error message from server after a client request+ | Banner BS.ByteString -- ^ Server "banner" received via an INFO message+ | Ping -- ^ Server ping challenge+ deriving Show++-- | Specialized parsed to return a NatsServerInfo+parseServerBanner :: BS.ByteString -> Either String NatsServerInfo+parseServerBanner bannerBytes = do+ case A.parseOnly bannerParser bannerBytes of+ Left err -> Left err+ Right (Banner b) -> eitherDecodeStrict b+ Right a -> Left $ "Expected server banner, got " ++ (show a)++-- | Parses a Message from a ByteString+parseMessage :: MonadThrow m => BS.ByteString -> m Message+parseMessage m =+ case A.parseOnly messageParser m of+ Left err -> throwM $ makeMessageParseError err+ Right msg -> return msg+ +bannerParser :: A.Parser Message+bannerParser = do+ _ <- A.string "INFO"+ A.skipSpace+ banner <- A.takeByteString+ return $ Banner banner++-- | Parse a 'BS.ByteString' into a 'Subject' or return an error message. See <http://nats.io/documentation/internals/nats-protocol/>+parseSubject :: BS.ByteString -> Either String Subject+parseSubject = A.parseOnly $ subjectParser <* A.endOfInput++-- | The actual parser is quite dumb, it doesn't try to validate silly subjects.+subjectParser :: A.Parser Subject+subjectParser = do+ tokens <- (A.takeWhile1 $ not . A.isSpace) `A.sepBy` (A.char '.')+ return $ makeSubject $ BS.intercalate "." tokens++msgParser :: A.Parser Message+msgParser = do+ _ <- A.string "MSG"+ A.skipSpace+ _subject <- subjectParser+ A.skipSpace+ _subscriptionId <- A.takeTill A.isSpace+ _ <- A.option "" (A.takeTill A.isSpace)+ A.skipSpace+ msgLength <- A.decimal+ A.endOfLine+ payload <- A.take msgLength+ A.endOfLine+ return $ Message payload++okParser :: A.Parser Message+okParser = do+ _ <- A.string "+OK"+ A.endOfLine+ return OKMsg++singleQuoted :: A.Parser BS.ByteString+singleQuoted = do+ _ <- A.char '\''+ str <- A.takeWhile $ \c -> c /= '\''+ _ <- A.char '\''+ return str++errorParser :: A.Parser Message+errorParser = do+ _ <- A.string "-ERR"+ A.skipSpace+ err <- singleQuoted+ A.endOfLine+ return $ ErrorMsg err++pingParser :: A.Parser Message+pingParser = do+ A.string "PING" *> A.endOfLine+ return Ping++messageParser :: A.Parser Message+messageParser = bannerParser <|> msgParser <|> okParser <|> errorParser <|> pingParser
+ src/Network/Nats/Protocol/Types.hs view
@@ -0,0 +1,106 @@+{-|+Module : Network.Nats.Protocol.Types+Description: NATS protocol types+-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}++module Network.Nats.Protocol.Types ( makeMessageParseError+ , makeSubject+ , maxPayloadSize+ , NatsServerInfo (..)+ , NatsConnectionOptions (..)+ , Subject (..)+ , Subscription+ , SubscriptionId (..)+ , ProtocolError (..)+ )+where++import Control.Exception+import Data.Aeson hiding (Object)+import Data.Aeson.Types (Options(..))+import Data.Default+import Data.Typeable+import GHC.Generics+import qualified Data.ByteString.Char8 as BS+import qualified Data.Text as T++-- | Server information returned on connection+data NatsServerInfo = NatsServerInfo{ _srv_server_id :: T.Text -- ^ Server ID+ , _srv_version :: T.Text -- ^ Server version+ , _srv_go :: T.Text -- ^ Version of Go (<https://www.golang.org>) the server was compiled against+ , _srv_host :: T.Text -- ^ Server hostname+ , _srv_port :: Int -- ^ Server port+ , _srv_auth_required :: Maybe Bool -- ^ Server requires authentication+ , _srv_ssl_required :: Maybe Bool -- ^ Server requires SSL connections+ , _srv_max_payload :: Int -- ^ Maximum message payload size accepted by server+ }+ deriving (Generic, Show)++instance FromJSON NatsServerInfo where+ parseJSON = genericParseJSON defaultOptions{ fieldLabelModifier = stripPrefix "_srv_" }+instance ToJSON NatsServerInfo where+ toEncoding = genericToEncoding defaultOptions{ fieldLabelModifier = stripPrefix "_srv_" }++-- | Retrieve the maximum payload size, in bytes+maxPayloadSize :: NatsServerInfo -> Int+maxPayloadSize = _srv_max_payload++-- | Client connection options sent when issuing a CONNECT to the server+-- See <http://nats.io/documentation/internals/nats-protocol/>+data NatsConnectionOptions = NatsConnectionOptions{ clnt_verbose :: Bool -- ^ Should server reply with an acknowledgment to every message?+ , clnt_pedantic :: Bool -- ^ Turn on strict format checking+ , clnt_ssl_required :: Bool -- ^ Require an SSL connection+ , clnt_name :: T.Text -- ^ Client name+ , clnt_lang :: T.Text -- ^ Client implementation language+ , clnt_version :: T.Text -- ^ Client version+ , clnt_protocol :: Int -- ^ Client protocol+ }+ deriving (Generic, Show)++instance FromJSON NatsConnectionOptions where+ parseJSON = genericParseJSON defaultOptions{ fieldLabelModifier = stripPrefix "clnt_" }+instance ToJSON NatsConnectionOptions where+ toEncoding = genericToEncoding defaultOptions{ fieldLabelModifier = stripPrefix "clnt_" }++instance Default NatsConnectionOptions where+ def = NatsConnectionOptions { clnt_verbose = False+ , clnt_pedantic = False+ , clnt_ssl_required = False+ , clnt_name = ""+ , clnt_lang = "haskell"+ , clnt_version = "0.1.0"+ , clnt_protocol = 1+ }++-- | Name of a NATS subject. Must be a dot-separated alphanumeric string, with ">" and "." as wildcard characters. See <http://nats.io/documentation/internals/nats-protocol/>+newtype Subject = Subject BS.ByteString deriving (Show)++-- | Create a Subject with the given ByteString as message+makeSubject :: BS.ByteString -> Subject+makeSubject bs = Subject bs++-- | A subscription to a 'Subject'+data Subscription = Subscription { _subject :: Subject+ , _subscriptionId :: SubscriptionId+ }+ deriving (Show)++-- | A 'Subscription' identifier+newtype SubscriptionId = SubscriptionId BS.ByteString+ deriving (Show, Ord, Eq)++stripPrefix :: String -> String -> String+stripPrefix prefix = drop prefixLength+ where prefixLength = length prefix++-- | Type for unexpected protocol errors+data ProtocolError = MessageParseError String -- ^ The message from the server could not be parsed.+ deriving (Show, Typeable)++instance Exception ProtocolError++-- | Create a 'MessageParseError' with the given reason+makeMessageParseError :: String -> ProtocolError+makeMessageParseError reason = MessageParseError reason
+ test/Network/Nats/Protocol/Tests.hs view
@@ -0,0 +1,41 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE UndecidableInstances #-}+module Network.Nats.Protocol.Tests where++import Hedgehog+import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy.Char8 as LBS+import Data.Either+import Network.Nats.Protocol+import Network.Nats.Test.Generators++prop_parseSubject :: Property+prop_parseSubject =+ withTests 5000 . property $ do+ subjectBytes <- forAll genSubjectBytes+ assert $ case BS.length (BS.filter (\c -> c `elem` invalidBytes) subjectBytes) of+ 0 -> isRight (parseSubject subjectBytes)+ _ -> isLeft (parseSubject subjectBytes)++prop_parseServerBanner :: Property+prop_parseServerBanner =+ withTests 5000 . property $ do+ serverInfo <- forAll genNatsServerBannerBytes+ let result = parseServerBanner $ LBS.toStrict serverInfo+ assert $ isRight result++prop_parseMessage :: Property+prop_parseMessage =+ withTests 5000 . property $ do+ msg <- forAll genMessageBytes+ _ <- evalM $ parseMessage $ LBS.toStrict msg+ success++tests :: IO Bool+tests =+ checkSequential $ Group "Network.Nats.Protocol.Tests" [+ ("prop_parseSubject", prop_parseSubject)+ , ("prop_parseServerBanner", prop_parseServerBanner)+ , ("prop_parseMessage", prop_parseMessage)+ ]
+ test/Network/Nats/Test/Generators.hs view
@@ -0,0 +1,112 @@+{-# LANGUAGE OverloadedStrings #-}+module Network.Nats.Test.Generators where++import Hedgehog+import Data.Aeson+import Data.ByteString.Builder+import qualified Data.ByteString.Char8 as BS+import qualified Data.ByteString.Lazy.Char8 as LBS+import Data.Semigroup ((<>))+import Network.Nats.Protocol.Types+import qualified Hedgehog.Gen as Gen+import qualified Hedgehog.Range as Range++invalidBytes :: [Char]+invalidBytes = " \t\n\r"++genValidSubjectBytes :: (MonadGen m, Functor m) => m BS.ByteString+genValidSubjectBytes =+ BS.intercalate "." <$> (Gen.list (Range.constant 1 5) $ Gen.utf8 (Range.constant 1 10) $ Gen.choice [ Gen.alphaNum, Gen.element (">*") ])++genInvalidSubjectBytes :: (MonadGen m, Functor m) => m BS.ByteString+genInvalidSubjectBytes =+ BS.intercalate "." <$> (Gen.list (Range.constant 1 5) $ Gen.utf8 (Range.singleton 1) $ Gen.element (">*" ++ invalidBytes))++genSubjectBytes :: (MonadGen m, Functor m) => m BS.ByteString+genSubjectBytes =+ Gen.choice [ genValidSubjectBytes+ , genInvalidSubjectBytes+ ]++genSubject :: (MonadGen m, Functor m) => m Subject+genSubject =+ Subject . (BS.intercalate ".") <$> (Gen.list (Range.constant 1 5) $ Gen.choice [ Gen.utf8 (Range.constant 1 10) Gen.alphaNum+ , Gen.utf8 (Range.singleton 1) $ Gen.element ">*"+ ])++genSubscriptionIdBytes :: (MonadGen m, Functor m) => m BS.ByteString+genSubscriptionIdBytes =+ Gen.utf8 (Range.constant 1 100) Gen.alphaNum++genNatsServerInfo :: (MonadGen m, Functor m) => m NatsServerInfo+genNatsServerInfo =+ NatsServerInfo+ <$> Gen.text (Range.constant 1 10) Gen.alphaNum+ <*> Gen.text (Range.constant 1 10) Gen.alphaNum+ <*> Gen.text (Range.constant 1 10) Gen.alphaNum+ <*> Gen.text (Range.constant 1 10) Gen.alphaNum+ <*> Gen.int (Range.constant 0 65535)+ <*> Gen.maybe Gen.bool+ <*> Gen.maybe Gen.bool+ <*> Gen.int (Range.constant 0 1000000)++genNatsServerBannerBytes :: (MonadGen m, Functor m) => m LBS.ByteString+genNatsServerBannerBytes = do+ nsi <- genNatsServerInfo+ return $ LBS.append (LBS.pack "INFO ") (encode nsi)++spaceBuilder :: Builder+spaceBuilder = charUtf8 ' '++eolBuilder :: Builder+eolBuilder = byteString "\r\n"++singleQuoteBuilder :: Builder+singleQuoteBuilder = charUtf8 '\''++genMessageMsgBytes :: (MonadGen m, Functor m) => m LBS.ByteString+genMessageMsgBytes = do+ msg <- Gen.bytes (Range.constant 1 1024)+ subj <- genValidSubjectBytes+ subs <- genSubscriptionIdBytes+ return $ toLazyByteString $+ stringUtf8 "MSG"+ <> spaceBuilder+ <> byteString subj+ <> spaceBuilder+ <> byteString subs+ <> spaceBuilder+ <> intDec (BS.length msg)+ <> eolBuilder+ <> byteString msg+ <> eolBuilder++genOkMsgBytes :: (MonadGen m, Functor m) => m LBS.ByteString+genOkMsgBytes = return $ toLazyByteString $+ stringUtf8 "+OK"+ <> eolBuilder++genErrorMsgBytes :: (MonadGen m, Functor m) => m LBS.ByteString+genErrorMsgBytes = do+ err <- Gen.list (Range.constant 1 1024) Gen.alphaNum+ return $ toLazyByteString $+ stringUtf8 "-ERR"+ <> spaceBuilder+ <> singleQuoteBuilder+ <> stringUtf8 err+ <> singleQuoteBuilder+ <> eolBuilder++genPingMsgBytes :: (MonadGen m, Functor m) => m LBS.ByteString+genPingMsgBytes = return $ toLazyByteString $+ stringUtf8 "PING"+ <> eolBuilder++genMessageBytes :: (MonadGen m, Functor m) => m LBS.ByteString+genMessageBytes =+ Gen.choice [ genMessageMsgBytes+ , genOkMsgBytes+ , genErrorMsgBytes+ , genPingMsgBytes+ , genNatsServerBannerBytes+ ]
+ test/Spec.hs view
@@ -0,0 +1,16 @@+module Main where++import Control.Monad (unless)+import System.Exit (exitFailure)+import System.IO (BufferMode(..), hSetBuffering, stdout, stderr)+import qualified Network.Nats.Protocol.Tests++main :: IO ()+main = do+ hSetBuffering stdout LineBuffering+ hSetBuffering stderr LineBuffering++ results <- sequence [ Network.Nats.Protocol.Tests.tests+ ]++ unless (and results) exitFailure