diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -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.
diff --git a/README.md b/README.md
new file mode 100644
--- /dev/null
+++ b/README.md
@@ -0,0 +1,5 @@
+# nats-client
+A Haskell client library for [NATS](https://nats.io).
+
+## Usage
+See [example](./example) for a sample client.
diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain
diff --git a/bench/Bench.hs b/bench/Bench.hs
new file mode 100644
--- /dev/null
+++ b/bench/Bench.hs
@@ -0,0 +1,7 @@
+module Main where
+
+import Criterion.Main
+
+main = defaultMain benches
+
+benches = []
diff --git a/example/ExampleClient.hs b/example/ExampleClient.hs
new file mode 100644
--- /dev/null
+++ b/example/ExampleClient.hs
@@ -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" }
diff --git a/nats-client.cabal b/nats-client.cabal
new file mode 100644
--- /dev/null
+++ b/nats-client.cabal
@@ -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
diff --git a/src/Network/Nats/Client.hs b/src/Network/Nats/Client.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Nats/Client.hs
@@ -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
diff --git a/src/Network/Nats/Protocol.hs b/src/Network/Nats/Protocol.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Nats/Protocol.hs
@@ -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
diff --git a/src/Network/Nats/Protocol/Message.hs b/src/Network/Nats/Protocol/Message.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Nats/Protocol/Message.hs
@@ -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
+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
diff --git a/src/Network/Nats/Protocol/Types.hs b/src/Network/Nats/Protocol/Types.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Nats/Protocol/Types.hs
@@ -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 () 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
+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
+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
diff --git a/test/Network/Nats/Protocol/Tests.hs b/test/Network/Nats/Protocol/Tests.hs
new file mode 100644
--- /dev/null
+++ b/test/Network/Nats/Protocol/Tests.hs
@@ -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)
+ ]
diff --git a/test/Network/Nats/Test/Generators.hs b/test/Network/Nats/Test/Generators.hs
new file mode 100644
--- /dev/null
+++ b/test/Network/Nats/Test/Generators.hs
@@ -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
+ ]
diff --git a/test/Spec.hs b/test/Spec.hs
new file mode 100644
--- /dev/null
+++ b/test/Spec.hs
@@ -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