nats-queue 0.1.1.0 → 0.1.2.0

raw patch · 5 files changed

+363/−137 lines, 5 filesdep +asyncdep +hspecdep +nats-queuedep ~aesondep ~basedep ~containersPVP ok

version bump matches the API change (PVP)

Dependencies added: async, hspec, nats-queue

Dependency ranges changed: aeson, base, containers, dequeue, network, network-uri

API changes (from Hackage documentation)

Files

• Network/Nats.hs +271/−124
• Network/Nats/Json.hs +14/−0
• nats-queue.cabal +34/−13
• test/Network/NatsSpec.hs +43/−0
• test/Spec.hs +1/−0

@@ -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 ()

@@ -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

@@ -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 <http://nats.io>. @@ -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

@@ -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

@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}