diff --git a/Network/Nats.hs b/Network/Nats.hs --- a/Network/Nats.hs +++ b/Network/Nats.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE TemplateHaskell,OverloadedStrings,RecordWildCards,GeneralizedNewtypeDeriving,PatternGuards,DeriveDataTypeable #-} +{-# LANGUAGE TemplateHaskell,OverloadedStrings,RecordWildCards,GeneralizedNewtypeDeriving,PatternGuards,DeriveDataTypeable, ScopedTypeVariables #-} module Network.Nats ( -- * How to use this module @@ -35,28 +35,37 @@ -- server usually closes the connection when there is an error. -- * Comparison to API in other languages - -- |Compared to API in other languages, the Haskell binding is very sparse. It does + -- |Compared to API in other languages, the Haskell binding does -- not implement timeouts and automatic unsubscribing, the 'request' call is implemented -- as a synchronous call. -- -- The timeouts can be easily implemented using 'System.Timeout' module, automatic unsubscribing - -- can be easily done in the callback function. + -- can be done in the callback function. -- * Error behaviour -- |The 'connect' function tries to connect to the NATS server. In case of failure it immediately fails. -- If there is an error during operations, the NATS module tries to reconnect to the server. + -- When there are more servers, the client immediately tries to connect to the next server. If + -- that fails, it waits 1s before trying the next server in the NatsSettings list. + -- -- During the reconnection, the calls 'subscribe' and 'request' will block. The calls -- 'publish' and 'unsubscribe' silently fail (unsubscribe is handled locally, NATS is a messaging -- system without guarantees, 'publish' is not guaranteed to succeed anyway). -- After reconnecting to the server, the module automatically resubscribes to previously subscribed channels. -- - -- If there is network failure, the nats commands 'subscribe' and 'request' - -- may fail on a network exception. The 'subscribe' + -- If there is a network failure, the nats commands 'subscribe' and 'request' + -- may fail on an IOexception or NatsException. The 'subscribe' -- command is synchronous, it waits until the server responds with +OK. The commands 'publish' - -- and 'unsubscribe' are asynchronous, no confirmation from server is required. + -- and 'unsubscribe' are asynchronous, no confirmation from server is required and they + -- should not raise an exception. + -- Nats , NatsSID , connect + , connectSettings + , NatsHost(..) + , NatsSettings(..) + , defaultSettings -- * Exceptions , NatsException -- * Access @@ -65,6 +74,7 @@ , unsubscribe , publish , request + , requestMany -- * Termination , disconnect ) where @@ -74,15 +84,18 @@ import Control.Concurrent.MVar import Control.Concurrent import qualified Network.Socket as S -import Network.Socket (SocketOption(KeepAlive), setSocketOption, getAddrInfo, SockAddr(..)) +import Network.Socket (SocketOption(KeepAlive, NoDelay), setSocketOption, getAddrInfo, SockAddr(..)) import Control.Monad (forever, replicateM) import Data.Dequeue as D import Control.Applicative ((<$>)) import Data.Typeable import qualified Data.Foldable as FOLD -import Control.Exception (bracket, bracketOnError, throwIO, catch, IOException, AsyncException, Exception) +import Control.Exception (bracket, bracketOnError, throwIO, catch, IOException, AsyncException, Exception, SomeException, + catches, Handler(..)) import System.Random (randomRIO) import Data.IORef +import System.Timeout +import Control.Concurrent.Async (concurrently) import qualified Data.Map.Strict as Map import qualified Data.ByteString.Lazy.Char8 as BL @@ -96,14 +109,22 @@ import qualified Network.URI as URI +-- | How often should we ping the server +pingInterval :: Int +pingInterval = 3000000 + +-- | Timeout interval for connect operations +timeoutInterval :: Int +timeoutInterval = 1000000 + -- | NATS communication error data NatsException = NatsException String deriving (Show, Typeable) instance Exception NatsException - + data NatsConnectionOptions = NatsConnectionOptions { - natsConnUser :: T.Text - , natsConnPass :: T.Text + natsConnUser :: String + , natsConnPass :: String , natsConnVerbose :: Bool , natsConnPedantic :: Bool , natsConnSslRequired :: Bool @@ -123,10 +144,10 @@ -- | Server information sent usually upon opening a connection to NATS server data NatsServerInfo = NatsServerInfo { - natsSvrServerId :: T.Text - , natsSvrVersion :: T.Text - , natsSvrMaxPayload :: Int - , natsSvrAuthRequired :: Bool +-- natsSvrServerId :: T.Text +-- , natsSvrVersion :: T.Text +-- , natsSvrMaxPayload :: Int + natsSvrAuthRequired :: Bool } deriving (Show) $(deriveJSON defaultOptions{fieldLabelModifier =( @@ -162,9 +183,7 @@ -- | Control structure representing a connection to NATS server data Nats = Nats { - natsConnOptions :: NatsConnectionOptions - , natsHost :: String - , natsPort :: Int + natsSettings :: NatsSettings , natsRuntime :: MVar (Handle, -- Network socket FifoQueue, -- FIFO of sent commands waiting for ack Bool, -- False if we are disconnected @@ -175,7 +194,31 @@ , natsSubMap :: IORef (Map.Map NatsSID NatsSubscription) } +-- | Host settings; may have different username/password for each host +data NatsHost = NatsHost { + natsHHost :: String + , natsHPort :: Int + , natsHUser :: String -- ^ Username for authentication + , natsHPass :: String -- ^ Password for authentication + } + +-- | Advanced settings for connecting to NATS server +data NatsSettings = NatsSettings { + natsHosts :: [NatsHost] + , natsOnConnect :: Nats -> (String, Int) -> IO () + -- ^ Called when a client has successfully connected. This callback is called synchronously + -- before the processing of incoming messages begins. + , natsOnDisconnect :: Nats -> String -> IO () + -- ^ Called when a client is disconnected. + } +defaultSettings :: NatsSettings +defaultSettings = (NatsSettings { + natsHosts = [(NatsHost "localhost" 4222 "nats" "nats")] + , natsOnConnect = \_ _ -> (return ()) + , natsOnDisconnect = \_ _ -> (return ()) + }) + -- | Message received by the client from the server data NatsSvrMessage = NatsSvrMsg { msgSubject::String, msgSid::NatsSID, msgText::BS.ByteString, msgReply::Maybe String} @@ -192,6 +235,7 @@ subjectToStr (Subject str) = str makeSubject :: String -> Subject +makeSubject "" = error "Empty subject" makeSubject str | any (<=' ') str = error $ "Subject contains incorrect characters: " ++ str | otherwise = Subject str @@ -223,11 +267,9 @@ -- | Decode NATS server message; result is message + payload (payload is 'undefined' in NatsSvrMsg) decodeMessage :: BS.ByteString -> Maybe (NatsSvrMessage, Maybe Int) -decodeMessage line = decodeMessage_ mid mpayload +decodeMessage line = decodeMessage_ mid (BS.drop 1 mrest) where - (mid, mpayload) = (BS.takeWhile (\x -> x/=' ' && x/='\r') line, - BS.drop 1 $ BS.dropWhile (\x -> x/=' ' && x/='\r') line) - + (mid, mrest) = BS.span (\x -> x/=' ' && x/='\r') line decodeMessage_ :: BS.ByteString -> BS.ByteString -> Maybe (NatsSvrMessage, Maybe Int) decodeMessage_ "PING" _ = Just (NatsSvrPing, Nothing) decodeMessage_ "PONG" _ = Just (NatsSvrPong, Nothing) @@ -237,8 +279,7 @@ info <- AE.decode $ BL.fromChunks [msg] return $ (NatsSvrInfo info, Nothing) decodeMessage_ "MSG" msg = do - let fields = BS.split ' ' msg - case (map BS.unpack fields) of + case (map BS.unpack (BS.words msg)) of [subj, sid, len] -> return (NatsSvrMsg subj (read sid) undefined Nothing, Just $ read len) [subj, sid, reply, len] -> return (NatsSvrMsg subj (read sid) undefined (Just $ reply), Just $ read len) _ -> fail "" @@ -265,7 +306,9 @@ (S.socket (S.addrFamily serveraddr) S.Stream S.defaultProtocol) (S.sClose) (\sock -> do + setSocketOption sock KeepAlive 1 + setSocketOption sock NoDelay 1 let connaddr = case (S.addrAddress serveraddr) of SockAddrInet _ haddr -> SockAddrInet (fromInteger $ toInteger port) haddr SockAddrInet6 _ finfo haddr scopeid -> SockAddrInet6 (fromInteger $ toInteger port) finfo haddr scopeid @@ -277,7 +320,10 @@ ) -ensureConnection :: Nats -> Bool -> ((Handle, FifoQueue) -> IO FifoQueue) -> IO () +ensureConnection :: Nats + -> Bool -- ^ If true, wait for the connection to become available + -> ((Handle, FifoQueue) -> IO FifoQueue) -- ^ Action to do when the connection is available + -> IO () -- Block if we are disconnected ensureConnection nats True f = do bracketOnError @@ -330,18 +376,32 @@ supportsCallback (NatsClntUnsubscribe {}) = True supportsCallback _ = False +-- Throw exception if an action does not end in specified time +timeoutThrow :: Int -> IO a -> IO a +timeoutThrow t f = do + res <- timeout t f + case res of + Just x -> return x + Nothing -> throwIO $ NatsException "Reached timeout" + _sendMessage :: Handle -> NatsClntMessage -> IO () -_sendMessage handle cmsg = do - BL.hPut handle $ makeClntMsg cmsg - BS.hPut handle "\r\n" +_sendMessage handle cmsg = timeoutThrow timeoutInterval $ do + let msg = makeClntMsg cmsg + case () of + _| BL.length msg < 1024 -> + BS.hPut handle $ BS.concat $ (BL.toChunks msg) ++ ["\r\n"] + | True -> do + BL.hPut handle msg + BS.hPut handle "\r\n" -- | Do the authentication handshake if necessary -authenticate :: Nats -> Handle -> IO () -authenticate nats handle = do +authenticate :: Handle -> String -> String -> IO () +authenticate handle user password = do info <- BS.hGetLine handle case (decodeMessage info) of Just (NatsSvrInfo (NatsServerInfo {natsSvrAuthRequired=True}), Nothing) -> do - BL.hPut handle $ makeClntMsg (NatsClntConnect $ natsConnOptions nats) + let coptions = defaultConnectionOptions{natsConnUser=user, natsConnPass=password} + BL.hPut handle $ makeClntMsg (NatsClntConnect coptions) BS.hPut handle "\r\n" response <- BS.hGetLine handle case (decodeMessage response) of @@ -352,21 +412,34 @@ _ -> throwIO $ NatsException "Incorrect input from server" -- | Open and authenticate a connection -prepareConnection :: Nats -> IO () -prepareConnection nats = do - handle <- connectToServer (natsHost nats) (natsPort nats) - authenticate nats handle - (_, _, _, csig) <- takeMVar (natsRuntime nats) - putMVar (natsRuntime nats) (handle, D.empty, True, undefined) - putMVar csig () +prepareConnection :: Nats -> NatsHost -> IO () +prepareConnection nats nhost = timeoutThrow timeoutInterval $ + bracketOnError + (connectToServer (natsHHost nhost) (natsHPort nhost)) + (hClose) + (\handle -> do + authenticate handle (natsHUser nhost) (natsHPass nhost) + csig <- modifyMVar (natsRuntime nats) $ \(_,_,_, csig) -> + return $ ((handle, D.empty, True, undefined), csig) + putMVar csig () + ) -- | Main thread that reads events from NATS server and reconnects if necessary -connectionThread :: Nats -> IO () -connectionThread nats = do - connectionHandler nats - `catch` errorHandler - `catch` finalHandler +connectionThread :: Nats + -> [NatsHost] -- ^ inifinite list of connections to try + -> IO () +connectionThread _ [] = error "Empty list of connections" +connectionThread nats (thisconn:nextconn) = do + mnewconnlist <- + (connectionHandler nats thisconn >> return Nothing) -- connectionHandler never returns... + `catches` [Handler (\(e :: IOException) -> Just <$> errorHandler e), + Handler (\(e :: NatsException) -> Just <$> errorHandler e), + Handler (\e -> finalHandler e >> return Nothing)] + case mnewconnlist of + Nothing -> return () -- Never happens + Just newconnlist -> connectionThread nats newconnlist where + finalize :: (Show e) => e -> IO () finalize e = do -- Hide existing connection (handle, queue, _, _) <- takeMVar (natsRuntime nats) @@ -376,70 +449,100 @@ hClose handle -- Call appropriate callbacks on unfinished calls FOLD.mapM_ (\f -> f $ Just (T.pack $ show e)) queue + -- Call user supplied disconnect + (natsOnDisconnect $ natsSettings nats) nats (show e) - errorHandler :: IOException -> IO () + errorHandler :: (Show e) => e -> IO [NatsHost] errorHandler e = do finalize e - tryToConnect - -- Restart - connectionThread nats + tryToConnect nextconn where - tryToConnect = do - threadDelay 5000000 - prepareConnection nats - `catch` ((\_ -> tryToConnect) :: IOException -> IO ()) - `catch` ((\_ -> tryToConnect) :: NatsException -> IO ()) + tryToConnect connlist@(conn:rest) = do + res <- ((prepareConnection nats conn) >> (return $ Just connlist)) + `catches` [ Handler (\(_ :: IOException) -> return Nothing), + Handler (\(_ :: NatsException) -> return Nothing) ] + case res of + Just restlist -> return restlist + Nothing -> threadDelay timeoutInterval >> tryToConnect rest + tryToConnect [] = error "Empty list of connections" -- Handler for exiting the thread finalHandler :: AsyncException -> IO () - finalHandler e = finalize e + finalHandler e = do + finalize e -connectionHandler :: Nats -> IO () -connectionHandler nats = do +pingerThread :: Nats -> IORef (Int, Int) -> IO () +pingerThread nats pingStatus = forever $ do + threadDelay pingInterval + -- todo - kontrola pingu + ok <- atomicModifyIORef' pingStatus $ \(pings, pongs) -> ((pings+1, pongs), pings - pongs < 2) + if ok == False + then throwIO (NatsException "Ping timeouted") + else return () + sendMessage nats True NatsClntPing Nothing + +-- | Forever read input from a connection and process it +connectionHandler :: Nats -> NatsHost -> IO () +connectionHandler nats (NatsHost host port _ _) = do (handle, _, _, _) <- readMVar (natsRuntime nats) -- Subscribe channels that are supposed to be subscribed subscriptions <- readIORef (natsSubMap nats) FOLD.forM_ subscriptions $ \(NatsSubscription subject queue _ sid) -> sendMessage nats True (NatsClntSubscribe subject sid queue) Nothing + + -- Call user function that we are successfully connected + (natsOnConnect $ natsSettings nats) nats (host, port) + -- Allocate structures for PING, IORef is probably easiest to manage + pingStatus <- newIORef (0, 0) + -- Perform the job - forever $ - let - -- | Pull callback for OK/ERR status from FIFO queue - popCb (h, queue, x1, x2) = return ((h, newq, x1, x2), item) - where - (item, newq) = D.popFront queue - handleMessage NatsSvrPing = sendMessage nats True NatsClntPong Nothing - handleMessage NatsSvrPong = return () - handleMessage NatsSvrOK = do - cb <- modifyMVar (natsRuntime nats) $ popCb - case cb of - Just f -> f Nothing - Nothing -> return () -- This should not happen, spurious OK - handleMessage (NatsSvrError txt) = do - cb <- modifyMVar (natsRuntime nats) $ popCb - case cb of - Just f -> f $ Just txt - Nothing -> putStrLn $ show txt - handleMessage (NatsSvrInfo (NatsServerInfo {natsSvrAuthRequired=True})) = do - sendMessage nats True (NatsClntConnect $ natsConnOptions nats) Nothing - handleMessage (NatsSvrInfo _) = return () - handleMessage (NatsSvrMsg {..}) = do - msubscription <- Map.lookup msgSid <$> readIORef (natsSubMap nats) - case msubscription of - Just subscription -> (subCallback subscription) msgSid msgSubject (BL.fromChunks [msgText]) msgReply - -- SID not found in map, force unsubscribe - Nothing -> sendMessage nats True (NatsClntUnsubscribe msgSid) Nothing - in do - line <- BS.hGetLine handle - case (decodeMessage line) of - Just (msg, Nothing) -> do - handleMessage msg - Just (msg@(NatsSvrMsg {}), Just paylen) -> do - payload <- BS.hGet handle paylen - _ <- BS.hGet handle 2 -- CRLF - handleMessage msg{msgText=payload} - _ -> - putStrLn $ "Incorrect message: " ++ (show line) + _ <- concurrently + (pingerThread nats pingStatus) + (connectionHandler' handle nats pingStatus) + return () + +connectionHandler' :: Handle -> Nats -> IORef (Int, Int) -> IO () +connectionHandler' handle nats pingStatus = forever $ do + line <- BS.hGetLine handle + case decodeMessage line of + Just (msg, Nothing) -> + handleMessage msg + Just (msg@(NatsSvrMsg {}), Just paylen) -> do + payload <- BS.hGet handle (paylen + 2) -- +2 = CRLF + handleMessage msg{msgText=BS.take paylen payload} + _ -> + putStrLn $ "Incorrect message: " ++ (show line) + + where + -- | Pull callback for OK/ERR status from FIFO queue + popCb (h, queue, x1, x2) = return ((h, newq, x1, x2), item) + where + (item, newq) = D.popFront queue + handleMessage NatsSvrPing = sendMessage nats True NatsClntPong Nothing + handleMessage NatsSvrPong = + atomicModifyIORef' pingStatus $ + \(pings, pongs) -> ((pings, pongs + 1), ()) + + handleMessage NatsSvrOK = do + cb <- modifyMVar (natsRuntime nats) $ popCb + case cb of + Just f -> f Nothing + Nothing -> return () -- This should not happen, spurious OK + handleMessage (NatsSvrError txt) = do + cb <- modifyMVar (natsRuntime nats) $ popCb + case cb of + Just f -> f $ Just txt + Nothing -> putStrLn $ show txt + handleMessage (NatsSvrInfo _) = return () + handleMessage (NatsSvrMsg {..}) = do + msubscription <- Map.lookup msgSid <$> readIORef (natsSubMap nats) + case msubscription of + Just subscription -> + (subCallback subscription) msgSid msgSubject (BL.fromChunks [msgText]) msgReply + `catch` + (\(e :: SomeException) -> putStrLn $ (show e)) + -- SID not found in map, force unsubscribe + Nothing -> sendMessage nats True (NatsClntUnsubscribe msgSid) Nothing -- | Connect to a NATS server connect :: String -- ^ URI with format: nats:\/\/user:password\@localhost:4222 @@ -459,26 +562,36 @@ takeWhile (\x -> x /= '@') $ drop 1 $ dropWhile (\x -> x /= ':') uriUserInfo ) Nothing -> error "Missing hostname section" + connectSettings defaultSettings{ + natsHosts=[(NatsHost host port user password)] + } +-- | Connect to NATS server using custom settings +connectSettings :: NatsSettings -> IO Nats +connectSettings settings = do csig <- newEmptyMVar mruntime <- newMVar (undefined, undefined, False, csig) mthreadid <- newEmptyMVar nextsid <- newIORef 1 submap <- newIORef Map.empty - let opts = defaultConnectionOptions{natsConnUser=T.pack user, natsConnPass=T.pack password} let nats = Nats{ - natsConnOptions=opts - , natsHost=host - , natsPort=port - , natsRuntime=mruntime - , natsThreadId=mthreadid - , natsNextSid=nextsid - , natsSubMap=submap + natsSettings=settings + , natsRuntime=mruntime + , natsThreadId=mthreadid + , natsNextSid=nextsid + , natsSubMap=submap } - prepareConnection nats - threadid <- forkIO $ connectionThread nats + hosts = (natsHosts settings) + + -- Try to connect to all until one succeeds + connhost <- tryUntilSuccess hosts $ prepareConnection nats + threadid <- forkIO $ connectionThread nats (connhost:(cycle hosts)) putMVar mthreadid threadid return nats + where + tryUntilSuccess [a] f = f a >> return a + tryUntilSuccess (a:rest) f = (f a >> return a) `catch` (\(_ :: SomeException) -> tryUntilSuccess rest f) + tryUntilSuccess [] _ = error "Empty list" -- | Subscribe to a channel, optionally specifying queue group subscribe :: Nats @@ -486,22 +599,26 @@ -> (Maybe String) -- ^ Queue -> MsgCallback -- ^ Callback -> IO NatsSID -- ^ SID of subscription -subscribe nats subject queue cb = do - let ssubject = makeSubject subject - let squeue = makeSubject `fmap` queue - mvar <- newEmptyMVar :: IO (MVar (Maybe T.Text)) - sid <- newNatsSid nats - sendMessage nats True (NatsClntSubscribe ssubject sid squeue) $ Just $ \err -> do - case err of - Just _ -> return () - Nothing -> atomicModifyIORef' (natsSubMap nats) $ \ioref -> - (Map.insert sid (NatsSubscription{subSubject=ssubject, subQueue=squeue, subCallback=cb, subSid=sid}) ioref, ()) - putMVar mvar err - merr <- takeMVar mvar - case merr of - Just err -> throwIO $ NatsException $ T.unpack err - Nothing -> return $ sid +subscribe nats subject queue cb = + let + ssubject = makeSubject subject + squeue = makeSubject `fmap` queue + addToSubTable sid = atomicModifyIORef' (natsSubMap nats) $ \submap -> + (Map.insert sid (NatsSubscription{subSubject=ssubject, subQueue=squeue, subCallback=cb, subSid=sid}) submap, ()) + in do + mvar <- newEmptyMVar :: IO (MVar (Maybe T.Text)) + sid <- newNatsSid nats + sendMessage nats True (NatsClntSubscribe ssubject sid squeue) $ Just $ \err -> do + case err of + Nothing -> addToSubTable sid + Just _ -> return () + putMVar mvar err + merr <- takeMVar mvar + case merr of + Just err -> throwIO $ NatsException $ T.unpack err + Nothing -> return $ sid + -- | Unsubscribe from a channel unsubscribe :: Nats -> NatsSID @@ -511,9 +628,11 @@ atomicModifyIORef' (natsSubMap nats) $ \ioref -> (Map.delete sid ioref, ()) -- Unsubscribe from server, ignore errors sendMessage nats False (NatsClntUnsubscribe sid) Nothing - `catch` ((\_ -> return ()) :: IOException -> IO ()) + `catches` [ Handler (\(_ :: IOException) -> return ()), + Handler (\(_ :: NatsException) -> return ()) ] + --- | Synchronous request/response communication +-- | Synchronous request/response communication to obtain one message request :: Nats -> String -- ^ Subject -> BL.ByteString -- ^ Request @@ -526,7 +645,7 @@ _ <- tryPutMVar mvar (Right response) return () ) - (\sid -> unsubscribe nats sid) + (unsubscribe nats) (\_ -> do sendMessage nats True (NatsClntPublish (makeSubject subject) (Just $ makeSubject inbox) body) $ Just $ \merr -> do case merr of @@ -538,15 +657,43 @@ Right res -> return $ res ) +-- | Synchronous request/response for obtaining many messages in certain timespan +requestMany :: Nats + -> String -- ^ Subject + -> BL.ByteString -- ^ Body + -> Int -- ^ Timeout in microseconds + -> IO [BL.ByteString] +requestMany nats subject body time = do + result <- newIORef [] + inbox <- newInbox + bracket + (subscribe nats inbox Nothing $ \_ _ response _ -> + atomicModifyIORef result $ \old -> (response:old, ()) + ) + (unsubscribe nats) + (\_ -> do + publish' nats subject (Just inbox) body + threadDelay time + ) + reverse <$> readIORef result + -- | Publish a message publish :: Nats -> String -- ^ Subject -> BL.ByteString -- ^ Data -> IO () -publish nats subject body = do +publish nats subject body = publish' nats subject Nothing body + +publish' :: Nats + -> String -- ^ Subject + -> Maybe String + -> BL.ByteString -- ^ Data + -> IO () +publish' nats subject inbox body = do -- Ignore errors - messages can get lost - sendMessage nats False (NatsClntPublish (makeSubject subject) Nothing body) Nothing - `catch` ((\_ -> return()) :: IOException -> IO ()) + sendMessage nats False (NatsClntPublish (makeSubject subject) (makeSubject <$> inbox) body) Nothing + `catches` [ Handler (\(_ :: IOException) -> return ()), + Handler (\(_ :: NatsException) -> return ()) ] -- | Disconnect from a NATS server disconnect :: Nats -> IO () diff --git a/Network/Nats/Json.hs b/Network/Nats/Json.hs --- a/Network/Nats/Json.hs +++ b/Network/Nats/Json.hs @@ -4,11 +4,14 @@ module Network.Nats.Json ( subscribe , publish + , requestMany ) where import Network.Nats (Nats, NatsSID) import qualified Network.Nats as N import qualified Data.Aeson as AE +import Data.Maybe (catMaybes) +import Control.Applicative ((<$>)) -- | Publish a message publish :: AE.ToJSON a => @@ -37,3 +40,14 @@ cb sid subj msg repl | Just body <- AE.decode msg = jcallback sid subj body repl | True = return () -- Ignore when there is an error decoding + +requestMany :: (AE.ToJSON a, AE.FromJSON b) => + Nats + -> String -- ^ Subject + -> a -- ^ Body + -> Int -- ^ Timeout in microseconds + -> IO [b] +requestMany nats subject body time = + decodeAndFilter <$> N.requestMany nats subject (AE.encode body) time + where + decodeAndFilter = catMaybes . map AE.decode diff --git a/nats-queue.cabal b/nats-queue.cabal --- a/nats-queue.cabal +++ b/nats-queue.cabal @@ -1,5 +1,5 @@ name: nats-queue -version: 0.1.1.0 +version: 0.1.2.0 synopsis: Haskell API for NATS messaging system description: This library is a Haskell driver for NATS . @@ -14,8 +14,7 @@ maintainer: palkovsky.ondrej@gmail.com category: Network build-type: Simple -cabal-version: >=1.8 - +cabal-version: >=1.10 source-repository head type: git @@ -23,14 +22,36 @@ library exposed-modules: Network.Nats, Network.Nats.Json - build-depends: base==4.*, network>=2.6, dequeue, random, network-uri>=2.6, - containers >= 0.5, bytestring, text, aeson >= 0.7 + build-depends: base == 4.* + , network >=2.6 + , dequeue + , random + , network-uri >= 2.6 + , containers >= 0.5 + , bytestring + , text + , aeson >= 0.7 + , async + ghc-options: -Wall - extensions: TemplateHaskell - OverloadedStrings - RecordWildCards - GeneralizedNewtypeDeriving - PatternGuards - DeriveDataTypeable - Rank2Types - + + default-language: Haskell2010 + +test-suite spec + main-is: Spec.hs + other-modules: Network.NatsSpec + type: exitcode-stdio-1.0 + hs-source-dirs: test + default-language: Haskell2010 + + build-depends: base + , nats-queue + , hspec + , network + , dequeue + , random + , network-uri + , containers + , bytestring + , text + , aeson diff --git a/test/Network/NatsSpec.hs b/test/Network/NatsSpec.hs new file mode 100644 --- /dev/null +++ b/test/Network/NatsSpec.hs @@ -0,0 +1,43 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Network.NatsSpec where + +import Network.Nats +import Test.Hspec +import Data.IORef +import Control.Concurrent (threadDelay) +import Data.ByteString.Lazy () + +natsUrl = "nats://nats:nats@localhost:4222" + +-- fakeSettings = defaultSettings { +-- natsHosts=[(NatsHost "127.0.0.1" 5222 "nats" "nats"), +-- (NatsHost "127.0.0.1" 5223 "nats" "nats"), +-- (NatsHost "127.0.0.1" 5224 "nats" "nats")], +-- natsOnConnect=onconnect, +-- natsOnDisconnect=ondisconnect +-- } + +spec :: Spec +spec = describe "Client" $ do + it ("connects and disconnects to " ++ natsUrl) $ do + nats <- connect natsUrl + disconnect nats + + it "sends and receives message" $ do + nats <- connect natsUrl + mcount <- newIORef 0 + sid <- subscribe nats "nats.test" Nothing $ \_ _ _ _ -> do + atomicModifyIORef mcount (\x -> (x+1, ())) + publish nats "nats.test" "Hello" + threadDelay 100000 + unsubscribe nats sid + publish nats "nats.test" "Hello2" + threadDelay 100000 + disconnect nats + newcount <- readIORef mcount + newcount `shouldBe` 1 + + +main :: IO () +main = hspec spec diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,1 @@ +{-# OPTIONS_GHC -F -pgmF hspec-discover #-}