nats-queue 0.1.2.0 → 0.1.2.1

raw patch · 3 files changed

+238/−224 lines, 3 filesdep ~aesondep ~basedep ~dequeuePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: aeson, base, dequeue

API changes (from Hackage documentation)

Files

• Network/Nats.hs +210/−196
• Network/Nats/Json.hs +16/−16
• nats-queue.cabal +12/−12

@@ -1,4 +1,10 @@-{-# LANGUAGE TemplateHaskell,OverloadedStrings,RecordWildCards,GeneralizedNewtypeDeriving,PatternGuards,DeriveDataTypeable, ScopedTypeVariables #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE PatternGuards              #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE TemplateHaskell            #-}  module Network.Nats (     -- * How to use this module@@ -16,13 +22,13 @@     -- 'publish' nats \"news\" \"I got news for you\"     --     -- 'unsubscribe' nats sid-    -- +    --     -- 'subscribe' nats \"gift\" Nothing $ \\_ _ msg mreply -> do     --     putStrLn $ show msg     --     case mreply of     --        Nothing -> return ()     --        Just reply -> 'publish' nats reply \"I've got a gift for you.\"-    --  +    --     -- reply <- 'request' nats \"gift\" \"Do you have anything for me?\"     --     -- putStrLn $ show reply@@ -33,15 +39,15 @@     -- Client commands are generally acknowledged by the server with an +OK message,     -- the library waits for acknowledgment only for the 'subscribe' command. The NATS     -- 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 does     -- not implement timeouts and automatic unsubscribing, the 'request' call is implemented-    -- as a synchronous call. +    -- as a synchronous call.     --     -- The timeouts can be easily implemented using 'System.Timeout' module, automatic unsubscribing     -- 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.@@ -53,10 +59,10 @@     -- 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 a network failure, the nats commands 'subscribe' and 'request' +    -- 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 they +    -- and 'unsubscribe' are asynchronous, no confirmation from server is required and they     -- should not raise an exception.     --     Nats@@ -80,34 +86,42 @@ ) where  -import System.IO-import Control.Concurrent.MVar-import Control.Concurrent-import qualified Network.Socket as S-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, SomeException,-                          catches, Handler(..))-import System.Random (randomRIO)-import Data.IORef-import System.Timeout-import Control.Concurrent.Async (concurrently)+import           Control.Applicative        ((<$>), (<*>))+import           Control.Concurrent+import           Control.Concurrent.Async   (concurrently)+import           Control.Exception          (AsyncException, Exception,+                                             Handler (..), IOException,+                                             SomeException, bracket,+                                             bracketOnError, catch, catches,+                                             throwIO)+import           Control.Monad              (forever, replicateM, unless, void,+                                             when, mzero)+import           Data.Dequeue               as D+import qualified Data.Foldable              as FOLD+import           Data.IORef+import           Data.Maybe                 (fromMaybe)+import           Data.Typeable+import           Network.Socket             (SockAddr (..),+                                             SocketOption (KeepAlive, NoDelay),+                                             getAddrInfo, setSocketOption)+import qualified Network.Socket             as S+import           System.IO+import           System.Random              (randomRIO)+import           System.Timeout -import qualified Data.Map.Strict as Map+import qualified Data.ByteString.Char8      as BS import qualified Data.ByteString.Lazy.Char8 as BL-import qualified Data.ByteString.Char8 as BS-import Data.Char (toLower, isUpper)-import qualified Data.Text as T-import Data.Text.Encoding (decodeUtf8)+import           Data.Char                  (isUpper, toLower)+import qualified Data.Map.Strict            as Map+import qualified Data.Text                  as T+import           Data.Text.Encoding         (decodeUtf8) -import qualified Data.Aeson as AE-import Data.Aeson.TH (deriveJSON, defaultOptions, fieldLabelModifier)+import qualified Data.Aeson                 as AE+import Data.Aeson ((.:), (.!=))+import           Data.Aeson.TH              (defaultOptions, deriveJSON,+                                             fieldLabelModifier) -import qualified Network.URI as URI+import qualified Network.URI                as URI  -- | How often should we ping the server pingInterval :: Int@@ -123,25 +137,25 @@ instance Exception NatsException  data NatsConnectionOptions = NatsConnectionOptions {-        natsConnUser :: String-        , natsConnPass :: String-        , natsConnVerbose :: Bool-        , natsConnPedantic :: Bool+        natsConnUser          :: String+        , natsConnPass        :: String+        , natsConnVerbose     :: Bool+        , natsConnPedantic    :: Bool         , natsConnSslRequired :: Bool     } deriving (Show)  defaultConnectionOptions :: NatsConnectionOptions defaultConnectionOptions = NatsConnectionOptions{natsConnUser="nats",natsConnPass="nats", natsConnVerbose=True,                                                 natsConnPedantic=True, natsConnSslRequired=False}-    -$(deriveJSON defaultOptions{fieldLabelModifier =(++$(deriveJSON defaultOptions{fieldLabelModifier =             let insertUnderscore acc chr                     | isUpper chr = chr : '_' : acc                     | otherwise   = chr : acc-            in +            in                 map toLower . drop 1 . reverse . foldl insertUnderscore [] . drop 8-        )} ''NatsConnectionOptions)-    +        } ''NatsConnectionOptions)+ -- | Server information sent usually upon opening a connection to NATS server data NatsServerInfo = NatsServerInfo { --     natsSvrServerId :: T.Text@@ -149,14 +163,14 @@ --     , natsSvrMaxPayload :: Int       natsSvrAuthRequired :: Bool     } deriving (Show)-    -$(deriveJSON defaultOptions{fieldLabelModifier =(++$(deriveJSON defaultOptions{fieldLabelModifier =             let insertUnderscore acc chr                     | isUpper chr = chr : '_' : acc                     | otherwise   = chr : acc-            in +            in                 map toLower . drop 1 . reverse . foldl insertUnderscore [] . drop 7-        )} ''NatsServerInfo)+        } ''NatsServerInfo)   newtype NatsSID = NatsSID Int deriving (Num, Ord, Eq)@@ -171,16 +185,16 @@         -> BL.ByteString -- ^ Message         -> Maybe String -- ^ Reply subject         -> IO ()-    + data NatsSubscription = NatsSubscription {-        subSubject :: Subject-      , subQueue :: Maybe Subject+        subSubject  :: Subject+      , subQueue    :: Maybe Subject       , subCallback :: MsgCallback-      , subSid :: NatsSID+      , subSid      :: NatsSID     }-    + type FifoQueue = D.BankersDequeue (Maybe T.Text -> IO ())-    + -- | Control structure representing a connection to NATS server data Nats = Nats {           natsSettings :: NatsSettings@@ -200,25 +214,34 @@       , natsHPort :: Int       , natsHUser :: String      -- ^ Username for authentication       , natsHPass :: String  -- ^ Password for authentication-    }-    +    } deriving (Show)+instance AE.FromJSON NatsHost where+  parseJSON (AE.Object v) =+    NatsHost <$>+      v .: "host" <*>+      v .: "port" .!= 4222 <*>+      v .: "user" .!= "nats" <*>+      v .: "pass" .!= "pass"+  parseJSON _ = mzero+ -- | 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 () +        natsHosts        :: [NatsHost]+      , natsOnReconnect  :: Nats -> (String, Int) -> IO ()+        -- ^ Called when a client has successfully re-connected. This callback is called synchronously+        --   before the processing of incoming messages begins. It is not called when the client+        --   connects the first time, as such connection is synchronous.+      , natsOnDisconnect :: Nats -> String -> IO ()         -- ^ Called when a client is disconnected.     }  defaultSettings :: NatsSettings-defaultSettings = (NatsSettings {-        natsHosts = [(NatsHost "localhost" 4222 "nats" "nats")]-      , natsOnConnect = \_ _ -> (return ())+defaultSettings = NatsSettings {+        natsHosts = [NatsHost "localhost" 4222 "nats" "nats"]+      , natsOnReconnect = \_ _ -> (return ())       , natsOnDisconnect = \_ _ -> (return ())-    })-    +    }+ -- | Message received by the client from the server data NatsSvrMessage =     NatsSvrMsg { msgSubject::String, msgSid::NatsSID, msgText::BS.ByteString, msgReply::Maybe String}@@ -227,8 +250,8 @@     | NatsSvrPing     | NatsSvrPong     | NatsSvrInfo NatsServerInfo-    deriving (Show)    -    +    deriving (Show)+ newtype Subject = Subject String deriving (Show)  subjectToStr :: Subject -> String@@ -239,7 +262,7 @@ makeSubject str     | any (<=' ') str = error $ "Subject contains incorrect characters: " ++ str     | otherwise       = Subject str-    + -- | Message sent from the client to server data NatsClntMessage =     NatsClntPing@@ -248,7 +271,7 @@     | NatsClntUnsubscribe NatsSID     | NatsClntPublish Subject (Maybe Subject) BL.ByteString     | NatsClntConnect NatsConnectionOptions-    + -- | Encode NATS client message makeClntMsg :: NatsClntMessage -> BL.ByteString makeClntMsg = BL.fromChunks . _makeClntMsg@@ -256,19 +279,19 @@         _makeClntMsg :: NatsClntMessage -> [BS.ByteString]         _makeClntMsg NatsClntPing = ["PING"]         _makeClntMsg NatsClntPong = ["PONG"]-        _makeClntMsg (NatsClntSubscribe subject sid (Just queue)) = [BS.pack $ "SUB " ++ (subjectToStr subject) ++ " " ++ (subjectToStr queue) ++ " " ++ (show sid)]-        _makeClntMsg (NatsClntSubscribe subject sid Nothing) = [BS.pack $ "SUB " ++ (subjectToStr subject) ++ " " ++ (show sid)]-        _makeClntMsg (NatsClntUnsubscribe sid) = [ BS.pack $ "UNSUB " ++ (show sid) ]-        _makeClntMsg (NatsClntPublish subj Nothing msg) = -            (BS.pack $ "PUB " ++ (subjectToStr subj) ++ " " ++ (show $ BL.length msg) ++ "\r\n") : BL.toChunks msg-        _makeClntMsg (NatsClntPublish subj (Just reply) msg) = -            (BS.pack $ "PUB " ++ (subjectToStr subj) ++ " " ++ (subjectToStr reply) ++ " " ++ (show $ BL.length msg) ++ "\r\n") : BL.toChunks msg-        _makeClntMsg (NatsClntConnect info) = "CONNECT " : (BL.toChunks $ AE.encode info)-    +        _makeClntMsg (NatsClntSubscribe subject sid (Just queue)) = [BS.pack $ "SUB " ++ subjectToStr subject ++ " " ++ subjectToStr queue ++ " " ++ show sid]+        _makeClntMsg (NatsClntSubscribe subject sid Nothing) = [BS.pack $ "SUB " ++ subjectToStr subject ++ " " ++ show sid]+        _makeClntMsg (NatsClntUnsubscribe sid) = [ BS.pack $ "UNSUB " ++ show sid ]+        _makeClntMsg (NatsClntPublish subj Nothing msg) =+            BS.pack ("PUB " ++ subjectToStr subj ++ " " ++ show (BL.length msg) ++ "\r\n") : BL.toChunks msg+        _makeClntMsg (NatsClntPublish subj (Just reply) msg) =+            BS.pack ("PUB " ++ subjectToStr subj ++ " " ++ subjectToStr reply ++ " " ++ show (BL.length msg) ++ "\r\n") : BL.toChunks msg+        _makeClntMsg (NatsClntConnect info) = "CONNECT " : BL.toChunks (AE.encode info)+ -- | Decode NATS server message; result is message + payload (payload is 'undefined' in NatsSvrMsg) decodeMessage :: BS.ByteString -> Maybe (NatsSvrMessage, Maybe Int) decodeMessage line = decodeMessage_ mid (BS.drop 1 mrest)-    where +    where         (mid, mrest) = BS.span (\x -> x/=' ' && x/='\r') line         decodeMessage_ :: BS.ByteString -> BS.ByteString -> Maybe (NatsSvrMessage, Maybe Int)         decodeMessage_ "PING" _ = Just (NatsSvrPing, Nothing)@@ -277,15 +300,15 @@         decodeMessage_ "-ERR" msg = Just (NatsSvrError (decodeUtf8 msg), Nothing)         decodeMessage_ "INFO" msg = do             info <- AE.decode $ BL.fromChunks [msg]-            return $ (NatsSvrInfo info, Nothing)-        decodeMessage_ "MSG" msg = do-            case (map BS.unpack (BS.words msg)) of+            return (NatsSvrInfo info, Nothing)+        decodeMessage_ "MSG" msg =+            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)+                 [subj, sid, reply, len] -> return (NatsSvrMsg subj (read sid) undefined (Just reply), Just $ read len)                  _ -> fail ""         decodeMessage_ _ _ = Nothing -    + -- | Returns next sid and updates MVar newNatsSid :: Nats -> IO NatsSID newNatsSid nats = atomicModifyIORef' (natsNextSid nats) $ \sid -> (sid + 1, sid)@@ -295,21 +318,21 @@ newInbox = do     rnd <- replicateM 13 (randomRIO ('a', 'z'))     return $ "_INBOX." ++ rnd-    + -- | Create a TCP connection to the server connectToServer :: String -> Int -> IO Handle connectToServer hostname port = do     addrinfos <- getAddrInfo Nothing (Just hostname) Nothing-    let serveraddr = (head addrinfos)+    let serveraddr = head addrinfos     -- Create a socket     bracketOnError         (S.socket (S.addrFamily serveraddr) S.Stream S.defaultProtocol)-        (S.sClose)+        S.sClose         (\sock -> do-            +             setSocketOption sock KeepAlive 1             setSocketOption sock NoDelay 1-            let connaddr = case (S.addrAddress serveraddr) of+            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                     other -> other@@ -318,24 +341,24 @@             hSetBuffering h NoBuffering             return h         )-        -        -ensureConnection :: Nats +++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+ensureConnection nats True f =     bracketOnError         (takeMVar $ natsRuntime nats)         (putMVar $ natsRuntime nats)         (\r@(handle, _, x1, x2) -> do             result <- runAction r-            case result of +            case result of                  Just nqueue -> putMVar (natsRuntime nats) (handle, nqueue, x1, x2)                  Nothing -> return ()         )-    where +    where         -- Connected         runAction (handle, queue, True, _) = do             nqueue <- f (handle, queue)@@ -356,17 +379,17 @@         -- Disconnected, ignore         runAction (handle, queue, False, csig) =             return (handle, queue, False, csig)-        + -- | Send a message and register callback if possible sendMessage :: Nats -> Bool -> NatsClntMessage -> Maybe (Maybe T.Text -> IO ()) -> IO () sendMessage nats blockIfDisconnected msg mcb-    | Just cb <- mcb, supportsCallback msg = +    | Just cb <- mcb, supportsCallback msg =         ensureConnection nats blockIfDisconnected $ \(handle, queue) -> do             _sendMessage handle msg             return $ D.pushBack queue cb -- Append callback on the callback queue     | supportsCallback msg = sendMessage nats blockIfDisconnected msg (Just $ \_ -> return ())     | Just _ <- mcb, not (supportsCallback msg) = error "Callback not supported"-    | True = ensureConnection nats blockIfDisconnected $ \(handle, queue) -> do+    | otherwise = ensureConnection nats blockIfDisconnected $ \(handle, queue) -> do         _sendMessage handle msg         return queue     where@@ -383,14 +406,14 @@     case res of          Just x -> return x          Nothing -> throwIO $ NatsException "Reached timeout"-        + _sendMessage :: Handle -> NatsClntMessage -> IO () _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+                BS.hPut handle $ BS.concat $ BL.toChunks msg ++ ["\r\n"]+        | otherwise -> do                 BL.hPut handle msg                 BS.hPut handle "\r\n" @@ -398,46 +421,47 @@ authenticate :: Handle -> String -> String -> IO () authenticate handle user password = do     info <- BS.hGetLine handle-    case (decodeMessage info) of+    case decodeMessage info of         Just (NatsSvrInfo (NatsServerInfo {natsSvrAuthRequired=True}), Nothing) -> do             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+            case decodeMessage response of                  Just (NatsSvrOK, Nothing) -> return ()-                 Just (NatsSvrError err, Nothing)-> throwIO $ NatsException $ "Authentication error: " ++ (show err)-                 _ -> throwIO $ NatsException $ "Incorrect server response"+                 Just (NatsSvrError err, Nothing)-> throwIO $ NatsException $ "Authentication error: " ++ show err+                 _ -> throwIO $ NatsException "Incorrect server response"         Just (NatsSvrInfo _, Nothing) -> return ()         _ -> throwIO $ NatsException "Incorrect input from server"-            + -- | Open and authenticate a connection prepareConnection :: Nats -> NatsHost -> IO () prepareConnection nats nhost = timeoutThrow timeoutInterval $     bracketOnError         (connectToServer (natsHHost nhost) (natsHPort nhost))-        (hClose)+        hClose         (\handle -> do             authenticate handle (natsHUser nhost) (natsHPass nhost)             csig <- modifyMVar (natsRuntime nats) $ \(_,_,_, csig) ->-                return $ ((handle, D.empty, True, undefined), csig)+                return ((handle, D.empty, True, undefined), csig)             putMVar csig ()         )  -- | Main thread that reads events from NATS server and reconnects if necessary-connectionThread :: Nats+connectionThread :: Bool+                    -> 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...+connectionThread _ _ [] = error "Empty list of connections"+connectionThread firstTime nats (thisconn:nextconn) = do+    mnewconnlist <-+        (connectionHandler firstTime 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+         Just newconnlist -> connectionThread False nats newconnlist     where         finalize :: (Show e) => e -> IO ()         finalize e = do@@ -451,160 +475,151 @@             FOLD.mapM_ (\f -> f $ Just (T.pack $ show e)) queue             -- Call user supplied disconnect             (natsOnDisconnect $ natsSettings nats) nats (show e)-            +         errorHandler :: (Show e) => e -> IO [NatsHost]         errorHandler e = do             finalize e             tryToConnect nextconn             where                 tryToConnect connlist@(conn:rest) = do-                    res <- ((prepareConnection nats conn) >> (return $ Just connlist))+                    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 = do-            finalize e+        finalHandler = finalize  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 ()+    unless  ok $ throwIO (NatsException "Ping timeouted")     sendMessage nats True NatsClntPing Nothing-        + -- | Forever read input from a connection and process it-connectionHandler :: Nats -> NatsHost -> IO ()-connectionHandler nats (NatsHost host port _ _) = do+connectionHandler :: Bool -> Nats -> NatsHost -> IO ()+connectionHandler firstTime 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)+    unless firstTime $ (natsOnReconnect $ natsSettings nats) nats (host, port)     -- Allocate structures for PING, IORef is probably easiest to manage     pingStatus <- newIORef (0, 0)-    +     -- Perform the job-    _ <- concurrently+    void $ 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) -> +        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)-            +        _ ->+            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+                (item, newq) = case D.popFront queue of+                    Just inq -> inq+                    Nothing  -> (maybe (return ()) print, D.empty)         handleMessage NatsSvrPing = sendMessage nats True NatsClntPong Nothing-        handleMessage NatsSvrPong = -            atomicModifyIORef' pingStatus $ +        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+            cb <- modifyMVar (natsRuntime nats) popCb+            cb Nothing         handleMessage (NatsSvrError txt) = do-            cb <- modifyMVar (natsRuntime nats) $ popCb-            case cb of-                Just f -> f $ Just txt-                Nothing -> putStrLn $ show txt+            cb <- modifyMVar (natsRuntime nats) popCb+            cb $ Just 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+                Just subscription ->+                    subCallback subscription msgSid msgSubject (BL.fromChunks [msgText]) msgReply                     `catch`-                        (\(e :: SomeException) -> putStrLn $ (show e))+                        (\(e :: SomeException) -> print e)                 -- SID not found in map, force unsubscribe-                Nothing -> sendMessage nats True (NatsClntUnsubscribe msgSid) Nothing -                        --- | Connect to a NATS server    +                Nothing -> sendMessage nats True (NatsClntUnsubscribe msgSid) Nothing++-- | Connect to a NATS server connect :: String -- ^ URI with format: nats:\/\/user:password\@localhost:4222     -> IO Nats connect uri = do-    let parsedUri = case (URI.parseURI uri) of -            Just x -> x-            Nothing -> error ("Error parsing NATS url: " ++ uri)-    if URI.uriScheme parsedUri /= "nats:"-        then error "Incorrect URL scheme"-        else return ()-    -    let (host, port, user, password) = case (URI.uriAuthority parsedUri) of-                Just (URI.URIAuth {..}) -> (uriRegName, +    let parsedUri = fromMaybe (error ("Error parsing NATS url: " ++ uri))+                      (URI.parseURI uri)+    when (URI.uriScheme parsedUri /= "nats:") $ error "Incorrect URL scheme"++    let (host, port, user, password) = case URI.uriAuthority parsedUri of+                Just (URI.URIAuth {..}) -> (uriRegName,                                           read $ drop 1 uriPort,-                                          takeWhile (\x -> x /= ':') uriUserInfo,-                                          takeWhile (\x -> x /= '@') $ drop 1 $ dropWhile (\x -> x /= ':') uriUserInfo+                                          takeWhile (/= ':') uriUserInfo,+                                          takeWhile (/= '@') $ drop 1 $ dropWhile (/= ':') uriUserInfo                                           )                 Nothing -> error "Missing hostname section"     connectSettings defaultSettings{-            natsHosts=[(NatsHost host port user password)]+            natsHosts=[NatsHost host port user password]         }  -- | Connect to NATS server using custom settings-connectSettings :: NatsSettings -> IO Nats                +connectSettings :: NatsSettings -> IO Nats connectSettings settings = do     csig <- newEmptyMVar     mruntime <- newMVar (undefined, undefined, False, csig)-    mthreadid <- newEmptyMVar +    mthreadid <- newEmptyMVar     nextsid <- newIORef 1     submap <- newIORef Map.empty     let nats = Nats{               natsSettings=settings-            , natsRuntime=mruntime -            , natsThreadId=mthreadid -            , natsNextSid=nextsid +            , natsRuntime=mruntime+            , natsThreadId=mthreadid+            , natsNextSid=nextsid             , natsSubMap=submap         }-        hosts = (natsHosts settings)-    +        hosts = natsHosts settings+     -- Try to connect to all until one succeeds     connhost <- tryUntilSuccess hosts $ prepareConnection nats-    threadid <- forkIO $ connectionThread nats (connhost:(cycle hosts))+    threadid <- forkIO $ connectionThread True 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 ++-- | Subscribe to a channel, optionally specifying queue group+subscribe :: Nats     -> String -- ^ Subject-    -> (Maybe String) -- ^ Queue+    -> Maybe String -- ^ Queue     -> MsgCallback -- ^ Callback     -> IO NatsSID -- ^ SID of subscription-subscribe nats subject queue cb = +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, ()) +                (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@@ -617,11 +632,11 @@         merr <- takeMVar mvar         case merr of             Just err -> throwIO $ NatsException $ T.unpack err-            Nothing -> return $ sid+            Nothing -> return sid  -- | Unsubscribe from a channel-unsubscribe :: Nats -    -> NatsSID +unsubscribe :: Nats+    -> NatsSID     -> IO () unsubscribe nats sid = do     -- Remove from internal tables@@ -630,10 +645,10 @@     sendMessage nats False (NatsClntUnsubscribe sid) Nothing         `catches` [ Handler (\(_ :: IOException) -> return ()),                     Handler (\(_ :: NatsException) -> return ()) ]-             + -- | Synchronous request/response communication to obtain one message-request :: Nats +request :: Nats     -> String             -- ^ Subject     -> BL.ByteString      -- ^ Request     -> IO BL.ByteString   -- ^ Response@@ -641,20 +656,19 @@     mvar <- newEmptyMVar :: IO (MVar (Either String BL.ByteString))     inbox <- newInbox     bracket-            (subscribe nats inbox Nothing $ \_ _ response _ -> do-                _ <- tryPutMVar mvar (Right response)-                return ()-            ) +            (subscribe nats inbox Nothing $ \_ _ response _ ->+                void $ tryPutMVar mvar (Right response)+            )             (unsubscribe nats)             (\_ -> do-                sendMessage nats True (NatsClntPublish (makeSubject subject) (Just $ makeSubject inbox) body) $ Just $ \merr -> do+                sendMessage nats True (NatsClntPublish (makeSubject subject) (Just $ makeSubject inbox) body) $ Just $ \merr ->                     case merr of                          Nothing -> return ()-                         Just err -> tryPutMVar mvar (Left $ T.unpack err) >> return ()+                         Just err -> void $ tryPutMVar mvar (Left $ T.unpack err)                 result <- takeMVar mvar                 case result of                      Left err -> throwIO $ NatsException err-                     Right res -> return $ res+                     Right res -> return res             )  -- | Synchronous request/response for obtaining many messages in certain timespan@@ -666,7 +680,7 @@ requestMany nats subject body time = do     result <- newIORef []     inbox <- newInbox-    bracket +    bracket         (subscribe nats inbox Nothing $ \_ _ response _ ->                 atomicModifyIORef result $ \old -> (response:old, ())         )@@ -676,25 +690,25 @@             threadDelay time         )     reverse <$> readIORef result-            + -- | Publish a message-publish :: Nats +publish :: Nats     -> String -- ^ Subject     -> BL.ByteString -- ^ Data     -> IO ()-publish nats subject body = publish' nats subject Nothing body-    -publish' :: Nats +publish nats subject = publish' nats subject Nothing++publish' :: Nats     -> String -- ^ Subject     -> Maybe String     -> BL.ByteString -- ^ Data     -> IO ()-publish' nats subject inbox body = do+publish' nats subject inbox body =     -- Ignore errors - messages can get lost     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 () disconnect nats = do

@@ -1,5 +1,4 @@-{-# LANGUAGE RecordWildCards,PatternGuards,Rank2Types #-}-+{-# LANGUAGE Rank2Types    #-}  module Network.Nats.Json (     subscribe@@ -7,26 +6,26 @@   , 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 ((<$>))+import           Control.Applicative ((<$>))+import qualified Data.Aeson          as AE+import           Data.Maybe          (mapMaybe)+import           Network.Nats        (Nats, NatsSID)+import qualified Network.Nats        as N  -- | Publish a message publish :: AE.ToJSON a =>-    Nats +    Nats     -> String -- ^ Subject     -> a -- ^ Data     -> IO () publish nats subject body = N.publish nats subject (AE.encode body) --- | Subscribe to a channel, optionally specifying queue group +-- | Subscribe to a channel, optionally specifying queue group -- If the JSON cannot be properly parsed, the message is ignored subscribe :: AE.FromJSON a =>-    Nats +    Nats     -> String -- ^ Subject-    -> (Maybe String) -- ^ Queue+    -> Maybe String -- ^ Queue     -> (NatsSID         -> String         -> a@@ -37,9 +36,10 @@     -> IO NatsSID -- ^ SID of subscription subscribe nats subject queue jcallback = N.subscribe nats subject queue cb     where-        cb sid subj msg repl-            | Just body <- AE.decode msg = jcallback sid subj body repl-            | True                       = return () -- Ignore when there is an error decoding+        cb sid subj msg repl = case AE.eitherDecode msg of+            Right body -> jcallback sid subj body repl+            -- Ignore when there is an error decoding+            Left err -> putStrLn $ err ++ ": " ++ show msg  requestMany :: (AE.ToJSON a, AE.FromJSON b) =>     Nats@@ -47,7 +47,7 @@     -> a                   -- ^ Body     -> Int                 -- ^ Timeout in microseconds     -> IO [b]-requestMany nats subject body time = +requestMany nats subject body time =    decodeAndFilter <$> N.requestMany nats subject (AE.encode body) time     where-        decodeAndFilter = catMaybes . map AE.decode+        decodeAndFilter = mapMaybe AE.decode

@@ -1,13 +1,13 @@ name:                nats-queue-version:             0.1.2.0+version:             0.1.2.1 synopsis:            Haskell API for NATS messaging system-description:         -    This library is a Haskell driver for NATS <http://nats.io>. -    NATS is a fast and lightweight publish-subscribe messaging system. -    -    This version supports authentication but does not support SSL as the current +description:+    This library is a Haskell driver for NATS <http://nats.io>.+    NATS is a fast and lightweight publish-subscribe messaging system.++    This version supports authentication but does not support SSL as the current     version of gnatsd (go server 0.5.6) does not support it either.-    + license:             BSD3 license-file:        LICENSE author:              Ondrej Palkovsky@@ -22,19 +22,19 @@  library   exposed-modules:   Network.Nats, Network.Nats.Json-  build-depends:     base == 4.*+  build-depends:     base >= 4.8 && < 4.9                    , network >=2.6-                   , dequeue+                   , dequeue >= 0.1.7                    , random                    , network-uri >= 2.6                    , containers  >= 0.5                    , bytestring                    , text-                   , aeson >= 0.7+                   , aeson >= 0.8.0.1                    , async-                   +   ghc-options:       -Wall-  +   default-language:  Haskell2010  test-suite spec