json-rpc 0.6.2.1 → 0.7.0.0

raw patch · 6 files changed

+507/−356 lines, 6 filesdep +vectorPVP ok

version bump matches the API change (PVP)

Dependencies added: vector

API changes (from Hackage documentation)

Files

• Network/JsonRpc/Arbitrary.hs +17/−0
• Network/JsonRpc/Data.hs +60/−16
• Network/JsonRpc/Interface.hs +198/−153
• README.md +30/−15
• json-rpc.cabal +3/−2
• test/Network/JsonRpc/Tests.hs +199/−170

@@ -37,11 +37,28 @@         , ErrorVal <$> arbitrary         ] +instance Arbitrary BatchRequest where+    arbitrary = oneof+        [ BatchRequest <$> arbitrary+        , SingleRequest <$> arbitrary+        ]++instance Arbitrary BatchResponse where+    arbitrary = oneof+        [ BatchResponse <$> arbitrary+        , SingleResponse <$> arbitrary+        ]+ instance Arbitrary Message where     arbitrary = oneof         [ MsgRequest  <$> arbitrary         , MsgResponse <$> arbitrary+        , MsgBatch    <$> batch         ]+      where+        batch = listOf $ oneof [ MsgRequest  <$> arbitrary+                               , MsgResponse <$> arbitrary+                               ]  instance Arbitrary Id where     arbitrary = oneof [IdInt <$> arbitrary, IdTxt <$> arbitrary]

@@ -4,6 +4,7 @@ module Network.JsonRpc.Data ( -- * Requests   Request(..)+, BatchRequest(..)   -- ** Parsing , FromRequest(..) , fromRequest@@ -13,6 +14,7 @@    -- * Responses , Response(..)+, BatchResponse(..)   -- ** Parsing , FromResponse(..) , fromResponse@@ -66,7 +68,7 @@                        , getReqMethod   :: !Method                        , getReqParams   :: !Value                        }-             deriving (Eq, Show)+             deriving (Eq, Show, Generic)  instance NFData Request where     rnf (Request v m p i) = rnf v `seq` rnf m `seq` rnf p `seq` rnf i@@ -118,7 +120,7 @@ parseVerIdMethParams :: Object -> Parser (Ver, Maybe Id, Method, Value) parseVerIdMethParams o = do     v <- parseVer o-    i <- parseId o+    i <- o .:? "id"     m <- o .: "method"     p <- o .:? "params" .!= Null     return (v, i, m, p)@@ -162,7 +164,7 @@               | OrphanError   { getResVer :: !Ver                               , getError  :: !ErrorObj                               }-              deriving (Eq, Show)+              deriving (Eq, Show, Generic)                instance NFData Response where@@ -189,6 +191,7 @@     -- Method corresponds to request to which this response answers.     parseResult :: Method -> Maybe (Value -> Parser r) +-- | Parse a response knowing the method of the corresponding request. fromResponse :: FromResponse r => Method -> Response -> Maybe r fromResponse m (Response _ r _) = parseResult m >>= flip parseMaybe r fromResponse _ _ = Nothing@@ -215,11 +218,12 @@                       -> Parser (Ver, Maybe Id, Either ErrorObj Value) parseVerIdResultError o = do     v <- parseVer o-    i <- parseId o+    i <- o .:? "id"     r <- o .:? "result" .!= Null     p <- if r == Null then Left <$> o .: "error" else return $ Right r     return (v, i, p) +-- | Create a response from a request. Use in servers. buildResponse :: (Monad m, FromRequest q, ToJSON r)               => Respond q m r               -> Request@@ -234,15 +238,17 @@                 Right r -> return . Just $ Response v (toJSON r) i buildResponse _ _ = return Nothing +-- | Type of function to make it easy to create a response from a request.+-- Meant to be used in servers. type Respond q m r = q -> m (Either ErrorObj r) --- Error object from JSON-RPC 2.0. ErrorVal for backwards compatibility.+-- | Error object from JSON-RPC 2.0. ErrorVal for backwards compatibility. data ErrorObj = ErrorObj  { getErrMsg  :: !String                           , getErrCode :: !Int                           , getErrData :: !Value                           }               | ErrorVal  { getErrData :: !Value }-              deriving (Show, Eq)+              deriving (Show, Eq, Generic)  instance NFData ErrorObj where     rnf (ErrorObj m c d) = rnf m `seq` rnf c `seq` rnf d@@ -264,6 +270,7 @@         ++ if d == Null then [] else ["data" .= d]     toJSON (ErrorVal v) = v +-- | Get a user-friendly string with the error information. fromError :: ErrorObj -> String fromError (ErrorObj m c v) = show c ++ ": " ++ m ++ ": " ++ valueAsString v fromError (ErrorVal (String t)) = T.unpack t@@ -297,22 +304,62 @@ -- Messages -- +data BatchRequest+    = BatchRequest     { getBatchRequest  :: ![Request] }+    | SingleRequest    { getSingleRequest ::  !Request  }+    deriving (Eq, Show, Generic)++instance NFData BatchRequest where+    rnf (BatchRequest qs) = rnf qs+    rnf (SingleRequest q) = rnf q++instance FromJSON BatchRequest where+    parseJSON qs@Array{} = BatchRequest  <$> parseJSON qs+    parseJSON q@Object{} = SingleRequest <$> parseJSON q+    parseJSON _ = mzero++instance ToJSON BatchRequest where+    toJSON (BatchRequest qs) = toJSON qs+    toJSON (SingleRequest q) = toJSON q++data BatchResponse+    = BatchResponse    { getBatchResponse :: ![Response] }+    | SingleResponse   { getSingleResponse :: !Response  }+    deriving (Eq, Show, Generic)++instance NFData BatchResponse where+    rnf (BatchResponse qs) = rnf qs+    rnf (SingleResponse q) = rnf q++instance FromJSON BatchResponse where+    parseJSON qs@Array{} = BatchResponse  <$> parseJSON qs+    parseJSON q@Object{} = SingleResponse <$> parseJSON q+    parseJSON _ = mzero++instance ToJSON BatchResponse where+    toJSON (BatchResponse qs) = toJSON qs+    toJSON (SingleResponse q) = toJSON q+ data Message-    = MsgRequest   { getMsgRequest  :: !Request  }-    | MsgResponse  { getMsgResponse :: !Response }-    deriving (Eq, Show)+    = MsgRequest  { getMsgRequest  :: !Request   }+    | MsgResponse { getMsgResponse :: !Response  }+    | MsgBatch    { getBatch       :: ![Message] }+    deriving (Eq, Show, Generic)  instance NFData Message where     rnf (MsgRequest  q) = rnf q     rnf (MsgResponse r) = rnf r+    rnf (MsgBatch    b) = rnf b  instance ToJSON Message where     toJSON (MsgRequest  q) = toJSON q     toJSON (MsgResponse r) = toJSON r+    toJSON (MsgBatch    b) = toJSON b  instance FromJSON Message where-    parseJSON v = (MsgRequest  <$> parseJSON v)-              <|> (MsgResponse <$> parseJSON v)+    parseJSON v = (MsgRequest   <$> parseJSON v)+              <|> (MsgResponse  <$> parseJSON v)+              <|> (MsgBatch     <$> parseJSON v)  -- -- Types@@ -344,15 +391,12 @@     toJSON (IdTxt s) = toJSON s     toJSON (IdInt n) = toJSON n -parseId :: Object -> Parser (Maybe Id)-parseId o = do-    d <- o .:? "id" .!= Null-    if d == Null then return Nothing else parseJSON d-+-- | Pretty display a message id. Meant for logs. fromId :: Id -> String fromId (IdInt i) = show i fromId (IdTxt t) = T.unpack t +-- | JSON-RPC version. data Ver = V1 -- ^ JSON-RPC 1.0          | V2 -- ^ JSON-RPC 2.0          deriving (Eq, Show, Read, Generic)

@@ -13,8 +13,11 @@    -- * Communicate with remote party , receiveRequest+, receiveBatchRequest , sendResponse+, sendBatchResponse , sendRequest+, sendBatchRequest    -- * Transports   -- ** Client@@ -49,22 +52,25 @@ import qualified Data.HashMap.Strict as M import Data.Conduit import qualified Data.Conduit.List as CL+import Data.Maybe import Data.Conduit.Network import Data.Conduit.TMChan-import qualified Data.Text as T+import qualified Data.Foldable as F+import qualified Data.Vector as V import Network.JsonRpc.Data  type SentRequests = HashMap Id (TMVar (Maybe Response)) -data Session = Session { inCh     :: TBMChan (Either Response Message)+data Session = Session { inCh     :: TBMChan (Either Response Value)                        , outCh    :: TBMChan Message-                       , reqCh    :: Maybe (TBMChan Request)+                       , reqCh    :: Maybe (TBMChan BatchRequest)                        , lastId   :: TVar Id                        , sentReqs :: TVar SentRequests                        , rpcVer   :: Ver+                       , dead     :: TVar Bool                        } --- Context for JSON-RPC connection. Connection will remain active as long+-- Context for JSON-RPC connection.  Connection will remain active as long -- as context is maintaned. type JsonRpcT = ReaderT Session @@ -76,24 +82,16 @@             <*> newTVar (IdInt 0)             <*> newTVar M.empty             <*> return v+            <*> newTVar False -encodeConduit :: MonadLogger m => Conduit Message m ByteString-encodeConduit = CL.mapM $ \m -> do-    $(logDebug) $ T.pack $ unwords $ case m of-        MsgRequest Request{getReqId = i} ->-            [ "encoding request id:", fromId i ]-        MsgRequest Notif{} ->-            [ "encoding notification" ]-        MsgResponse Response{getResId = i} ->-            [ "encoding response id:", fromId i ]-        MsgResponse ResponseError{getResId = i} ->-            [ "encoding error id:", fromId i ]-        MsgResponse OrphanError{} ->-            [ "encoding error without id" ]-    return . L8.toStrict $ encode m+-- Conduit to encode JSON to ByteString.+encodeConduit :: (ToJSON j, MonadLogger m) => Conduit j m ByteString+encodeConduit = CL.mapM $ \m -> return . L8.toStrict $ encode m +-- | Conduit to decode incoming messages.  Left Response indicates+-- a response to send back to sender if parsing JSON fails. decodeConduit :: MonadLogger m-              => Ver -> Conduit ByteString m (Either Response Message)+              => Ver -> Conduit ByteString m (Either Response Value) decodeConduit ver = evalStateT loop Nothing where     loop = lift await >>= maybe flush process     flush = get >>= maybe (return ()) (handle True . ($ B8.empty))@@ -108,139 +106,178 @@         loop     handle _ (Partial k) = put (Just k) >> loop     handle _ (Done rest v) = do-        let msg = decod v-        when (isLeft msg) $ $(logError) "received invalid message"-        lift $ yield msg+        lift $ yield $ Right v         if B8.null rest then loop else process rest -    decod v = case parseMaybe parseJSON v of-        Just msg -> Right msg-        Nothing -> Left $ OrphanError ver (errorInvalid v)-+-- | Process incoming messages. Do not use this directly unless you know+-- what you are doing. This is an internal function. processIncoming :: (Functor m, MonadLoggerIO m) => JsonRpcT m ()-processIncoming = do-    i  <- reader inCh-    o  <- reader outCh-    qM <- reader reqCh-    s  <- reader sentReqs-    join . liftIO . atomically $ readTBMChan i >>= \inc ->-        case inc of-            Nothing -> do-                m <- readTVar s-                mapM_ ((`putTMVar` Nothing) . snd) $ M.toList m-                return $ do-                    $(logDebug) "closed incoming channel"-                    unless (M.null m) $-                        $(logError) "some requests did not get responses"-                    return ()-            Just (Left e) -> do-                writeTBMChan o (MsgResponse e)-                return $ do-                    $(logError) "replied to sender with error"-                    processIncoming-            Just (Right (MsgRequest req)) ->-                case qM of-                    Just q -> do-                        writeTBMChan q req-                        return $ do-                            $(logDebug) "received request"-                            processIncoming-                    Nothing ->-                        case req of-                            Request v m _ d -> do-                                let e = ResponseError v (errorMethod m) d-                                writeTBMChan o (MsgResponse e)-                                return $ do-                                    $(logError) $ T.pack $ unwords-                                        [ "rejected incoming request id:"-                                        , fromId d-                                        ]-                                    processIncoming-                            Notif{} -> return $ do-                                $(logError) $ "ignored incoming notification"-                                processIncoming-            Just (Right (MsgResponse res)) -> do-                let hasId = case res of-                                Response{}      -> True-                                ResponseError{} -> True-                                OrphanError{}   -> False-                if hasId-                    then do-                        let x = getResId res-                        m <- readTVar s-                        let pM = x `M.lookup` m-                        case pM of-                            Nothing -> do-                                let v = getResVer res-                                    e = errorId x-                                    err = OrphanError v e-                                writeTBMChan o $ MsgResponse err-                                return $ do-                                    $(logError) $ T.pack $ unwords-                                        [ "got response with unknown id:"-                                        , fromId x-                                        ]-                                    processIncoming-                            Just p -> do-                                writeTVar s $ M.delete x m-                                putTMVar p $ Just res-                                return $ do-                                    $(logDebug) $ T.pack $ unwords-                                        [ "received response id:"-                                        , fromId x-                                        ]-                                    processIncoming-                    else return $ do-                        $(logError) $ T.pack $ unwords-                            [ "ignoring orhpan error:"-                            , fromError (getError res)-                            ]+processIncoming = ask >>= \qs -> join . liftIO . atomically $ do+    vEM <- readTBMChan $ inCh qs+    case vEM of+        Nothing -> flush qs+        Just vE ->+            case vE of+                Right v@Object{} -> do+                    single qs v+                    return $ do+                        $(logDebug) "received message"                         processIncoming+                Right v@(Array a) -> do+                    if V.null a+                        then do+                            let e = OrphanError (rpcVer qs) (errorInvalid v)+                            writeTBMChan (outCh qs) $ MsgResponse e+                        else batch qs (V.toList a)+                    return $ do+                        $(logDebug) "received batch"+                        processIncoming+                Right v -> do+                    let e = OrphanError (rpcVer qs) (errorInvalid v)+                    writeTBMChan (outCh qs) $ MsgResponse e+                    return $ do+                        $(logWarn) "got invalid message"+                        processIncoming+                Left e -> do+                    writeTBMChan (outCh qs) $ MsgResponse e+                    return $ do+                        $(logWarn) "error parsing JSON"+                        processIncoming+  where+    flush qs = do+        m <- readTVar $ sentReqs qs+        F.forM_ (reqCh qs) closeTBMChan+        closeTBMChan $ outCh qs+        writeTVar (dead qs) True+        mapM_ ((`putTMVar` Nothing) . snd) $ M.toList m+        return $ do+            $(logDebug) "session is now dead"+            unless (M.null m) $ $(logError) "requests remained unfulfilled"+    batch qs vs = do+        ts <- catMaybes <$> forM vs (process qs)+        unless (null ts) $+            if any isRight ts+                then do+                    let ch = fromJust $ reqCh qs+                    writeTBMChan ch $ BatchRequest $ rights ts+                else writeTBMChan (outCh qs) $ MsgBatch $ lefts ts+    single qs v = do+        tM <- process qs v+        case tM of+            Nothing -> return ()+            Just (Right t) -> do+                let ch = fromJust $ reqCh qs+                writeTBMChan ch $ SingleRequest t+            Just (Left e) -> writeTBMChan (outCh qs) e+    process qs v = do+        let qM = parseMaybe parseJSON v+        case qM of+            Just q -> request qs q+            Nothing -> do+                let rM = parseMaybe parseJSON v+                case rM of+                    Just r -> response qs r >> return Nothing+                    Nothing -> do+                        let e = OrphanError (rpcVer qs) (errorInvalid v)+                            m = MsgResponse e+                        return $ Just $ Left m+    request qs t =+        case reqCh qs of+            Just _ -> return $ Just $ Right t+            Nothing ->+                case t of+                    Notif{} -> return Nothing+                    Request{} -> do+                        let e = errorMethod (getReqMethod t)+                            v = getReqVer t+                            i = getReqId t+                            m = MsgResponse $ ResponseError v e i+                        return $ Just $ Left m+    response qs r = do+        let hasid = case r of+                        Response{}      -> True+                        ResponseError{} -> True+                        OrphanError{}   -> False -- Ignore orphan errors+        when hasid $ do+            let x = getResId r+            m <- readTVar (sentReqs qs)+            case x `M.lookup` m of+                Nothing -> return () -- Ignore orphan responses+                Just p -> do+                    writeTVar (sentReqs qs) $ M.delete x m+                    putTMVar p $ Just r  -- | Returns Nothing if did not receive response, could not parse it, or--- request was a notification. Just Left contains the error object returned+-- request is a notification. Just Left contains the error object returned -- by server if any. Just Right means response was received just right.-sendRequest :: (MonadLoggerIO m, ToJSON q, ToRequest q, FromResponse r)+sendRequest :: (MonadLoggerIO m , ToJSON q, ToRequest q, FromResponse r)             => q -> JsonRpcT m (Maybe (Either ErrorObj r))-sendRequest q = do+sendRequest q = head `liftM` sendBatchRequest [q]++-- | Send multiple requests in a batch. If only a single request, do not+-- put it in a batch.+sendBatchRequest :: (MonadLoggerIO m, ToJSON q, ToRequest q, FromResponse r)+                 => [q] -> JsonRpcT m [Maybe (Either ErrorObj r)]+sendBatchRequest qs = do     v <- reader rpcVer     l <- reader lastId     s <- reader sentReqs     o <- reader outCh-    if requestIsNotif q-        then do-            $(logDebug) "sending notification"-            liftIO . atomically $ do-                let req = buildRequest v q undefined-                writeTBMChan o $ MsgRequest req-            $(logDebug) "notification sent"-            return Nothing-        else do-            $(logDebug) "sending request"-            p <- liftIO . atomically $ do-                p <- newEmptyTMVar -                i <- succ <$> readTVar l-                m <- readTVar s-                let req = buildRequest v q i-                writeTVar s $ M.insert i p m-                writeTBMChan o $ MsgRequest req -                writeTVar l i-                return p-            $(logDebug) "request sent, awaiting for response"-            liftIO . atomically $ takeTMVar p >>= \rM -> case rM of-                Nothing -> return Nothing-                Just y@Response{} ->-                    case fromResponse (requestMethod q) y of-                        Nothing -> return Nothing-                        Just x -> return . Just $ Right x-                Just e@ResponseError{} ->-                    return . Just $ Left $ getError e-                _ -> undefined+    k <- reader dead+    aps <- liftIO . atomically $ do+        d <- readTVar k+        aps <- forM qs $ \q ->+            if requestIsNotif q+                then return (buildRequest v q undefined, Nothing)+                else do+                    p <- newEmptyTMVar +                    i <- succ <$> readTVar l+                    m <- readTVar s+                    unless d $ writeTVar s $ M.insert i p m+                    unless d $ writeTVar l i+                    if d+                        then return (buildRequest v q i, Nothing)+                        else return (buildRequest v q i, Just p)+        case map fst aps of+            []  -> return ()+            [a] -> unless d $ writeTBMChan o $ MsgRequest a+            as  -> unless d $ writeTBMChan o $ MsgBatch $ map MsgRequest as+        return aps+    if null aps+        then $(logDebug) "no responses pending"+        else $(logDebug) "listening for responses if pending"+    liftIO . atomically $ forM aps $ \(a, pM) ->+        case pM of+            Nothing -> return Nothing+            Just  p -> do+                rM <- takeTMVar p+                case rM of+                    Nothing -> return Nothing+                    Just r@Response{} ->+                        case fromResponse (getReqMethod a) r of+                            Nothing -> return Nothing+                            Just  x -> return $ Just $ Right x+                    Just e -> return $ Just $ Left $ getError e  -- | Receive requests from remote endpoint. Returns Nothing if incoming--- channel is closed or has never been opened.+-- channel is closed or has never been opened. Will reject incoming request+-- if sent in a batch. receiveRequest :: MonadLoggerIO m => JsonRpcT m (Maybe Request) receiveRequest = do+    bt <- receiveBatchRequest+    case bt of+        Nothing -> return Nothing+        Just (SingleRequest q) -> return $ Just q+        Just BatchRequest{} -> do+            v <- reader rpcVer+            let e = errorInvalid $ String "not accepting batches"+                m = OrphanError v e+            sendResponse m+            return Nothing++-- | Receive batch of requests. Will also accept single requests.+receiveBatchRequest :: MonadLoggerIO m => JsonRpcT m (Maybe BatchRequest)+receiveBatchRequest = do     chM <- reader reqCh     case chM of         Just ch -> do@@ -250,30 +287,42 @@             $(logError) "ignoring requests from remote endpoint"             return Nothing +-- | Send response message. Do not use to respond to a batch of requests. sendResponse :: MonadLoggerIO m => Response -> JsonRpcT m () sendResponse r = do     o <- reader outCh     liftIO . atomically . writeTBMChan o $ MsgResponse r +-- | Send batch of responses. Use to respond to a batch of requests.+sendBatchResponse :: MonadLoggerIO m => BatchResponse -> JsonRpcT m ()+sendBatchResponse (BatchResponse rs) = do+    o <- reader outCh+    liftIO . atomically . writeTBMChan o $ MsgBatch $ map MsgResponse rs+sendBatchResponse (SingleResponse r) = do+    o <- reader outCh+    liftIO . atomically . writeTBMChan o $ MsgResponse r++-- | Send any message. Do not use this. Use the other high-level functions+-- instead. Will not track request ids. Incoming responses to requests sent+-- using this method will be ignored. sendMessage :: MonadLoggerIO m => Message -> JsonRpcT m () sendMessage msg = reader outCh >>= liftIO . atomically . (`writeTBMChan` msg) --- | Create JSON-RPC session around conduits from transport--- layer. When context exits session disappears.+-- | Create JSON-RPC session around conduits from transport layer.  When+-- context exits session disappears. runJsonRpcT :: (MonadLoggerIO m, MonadBaseControl IO m)-            => Ver               -- ^ JSON-RPC version-            -> Bool              -- ^ Ignore incoming requests or notifications-            -> Sink Message m () -- ^ Sink to send messages-            -> Source m (Either Response Message)-            -- ^ Incoming messages or error responses to be returned to sender-            -> JsonRpcT m a      -- ^ JSON-RPC action-            -> m a               -- ^ Output of action+            => Ver                  -- ^ JSON-RPC version+            -> Bool                 -- ^ Ignore incoming requests/notifs+            -> Sink ByteString m () -- ^ Sink to send messages+            -> Source m ByteString  -- ^ Source to receive messages from+            -> JsonRpcT m a         -- ^ JSON-RPC action+            -> m a                  -- ^ Output of action runJsonRpcT ver ignore snk src f = do     qs <- liftIO . atomically $ initSession ver ignore     let inSnk  = sinkTBMChan (inCh qs) True         outSrc = sourceTBMChan (outCh qs)-    withAsync (src $$ inSnk) $ const $-        withAsync (outSrc $$ snk) $ \o ->+    withAsync (src $$ decodeConduit ver $= inSnk) $ const $+        withAsync (outSrc =$ encodeConduit $$ snk) $ \o ->             withAsync (runReaderT processIncoming qs) $ const $ do                 a <- runReaderT f qs                 liftIO $ do@@ -298,9 +347,7 @@     -> JsonRpcT m a   -- ^ JSON-RPC action     -> m a            -- ^ Output of action jsonRpcTcpClient ver ignore cs f = runGeneralTCPClient cs $ \ad ->-    runJsonRpcT ver ignore-        (encodeConduit =$ cr =$ appSink ad)-        (appSource ad $= decodeConduit ver) f+    runJsonRpcT ver ignore (cr =$ appSink ad) (appSource ad) f  -- | TCP server transport for JSON-RPC. jsonRpcTcpServer@@ -311,6 +358,4 @@     -> JsonRpcT m ()   -- ^ Action to perform on connecting client thread     -> m a jsonRpcTcpServer ver ignore ss f = runGeneralTCPServer ss $ \cl ->-    runJsonRpcT ver ignore-        (encodeConduit =$ cr =$ appSink cl)-        (appSource cl $= decodeConduit ver) f+    runJsonRpcT ver ignore (cr =$ appSink cl) (appSource cl) f

@@ -24,10 +24,13 @@ ``` haskell {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-}+import Control.Monad import Control.Monad.Trans import Control.Monad.Logger import Data.Aeson.Types hiding (Error) import Data.Conduit.Network+import qualified Data.Foldable as F+import Data.Maybe import Data.Time.Clock import Data.Time.Format import Network.JsonRpc@@ -56,21 +59,21 @@ srv :: MonadLoggerIO m => JsonRpcT m () srv = do     $(logDebug) "listening for new request"-    qM <- receiveRequest+    qM <- receiveBatchRequest     case qM of         Nothing -> do             $(logDebug) "closed request channel, exting"             return ()-        Just q -> do+        Just (SingleRequest q) -> do             $(logDebug) "got request"             rM <- buildResponse respond q-            case rM of-                Nothing -> do-                    $(logDebug) "no response for this request"-                    srv-                Just r -> do-                    $(logDebug) "sending response"-                    sendResponse r >> srv+            F.forM_ rM sendResponse+            srv+        Just (BatchRequest qs) -> do+            $(logDebug) "got request batch"+            rs <- catMaybes `liftM` forM qs (buildResponse respond)+            sendBatchResponse $ BatchResponse rs+            srv ```  Client Example@@ -112,21 +115,33 @@         f t = parseTime defaultTimeLocale "%c" (T.unpack t)     parseResult _ = Nothing -req :: MonadLoggerIO m => JsonRpcT m (Either String UTCTime)+req :: MonadLoggerIO m => JsonRpcT m UTCTime req = do     $(logDebug) "sending time request"     ts <- sendRequest TimeReq-    $(logDebug) "received response"     case ts of-        Nothing -> return $ Left "could not parse response"-        Just (Left e) -> return . Left $ fromError e-        Just (Right (TimeRes r)) -> return $ Right r+        Nothing -> error "could not parse response"+        Just (Left e) -> error $ fromError e+        Just (Right (TimeRes r)) -> return r +reqBatch :: MonadLoggerIO m => JsonRpcT m [UTCTime]+reqBatch = do+    $(logDebug) "sending time requests"+    ts <- sendBatchRequest $ replicate 4 TimeReq+    forM ts $ \t ->+        case t of+            Nothing -> error "could not parse response"+            Just (Left e) -> error $ fromError e+            Just (Right (TimeRes r)) -> return r++ main :: IO () main = runStderrLoggingT $     jsonRpcTcpClient V2 True (clientSettings 31337 "::1") $ do         $(logDebug) "sending four time requests one second apart"         replicateM_ 4 $ do-            req >>= liftIO . print+            req >>= $(logDebug) . T.pack . ("response: "++) . show             liftIO (threadDelay 1000000)+        $(logDebug) "sending four time requests in a batch"+        reqBatch >>= $(logDebug) . T.pack . ("response: "++) . show ```

@@ -1,5 +1,5 @@ name:                   json-rpc-version:                0.6.2.1+version:                0.7.0.0 synopsis:               Fully-featured JSON-RPC 2.0 library description:   This JSON-RPC library is fully-compatible with JSON-RPC 2.0 and 1.0. It@@ -24,7 +24,7 @@ source-repository this   type:                 git   location:             https://github.com/xenog/json-rpc.git-  tag:                  0.6.2.1+  tag:                  0.7.0.0  library   exposed-modules:      Network.JsonRpc@@ -48,6 +48,7 @@                         text                        >= 0.11     && < 1.3,                         transformers                >= 0.3,                         unordered-containers        >= 0.2      && < 0.3,+                        vector                      >= 0.10     && < 0.11,                         QuickCheck                  >= 2.6      && < 2.9   default-language:     Haskell2010   ghc-options:          -Wall

@@ -4,6 +4,7 @@ module Network.JsonRpc.Tests (tests) where  import Control.Applicative+import Control.Arrow import Control.Concurrent.Async.Lifted import Control.Concurrent.STM import Control.Monad@@ -32,38 +33,44 @@  tests :: [Test] tests =-    [ testGroup "JSON-RPC Requests"-        [ testProperty "Check fields"+    [ testGroup "Messages"+        [ testProperty "Check request fields"             (reqFields :: Request -> Bool)-        , testProperty "Encode/decode"+        , testProperty "Encode/decode request"             (testEncodeDecode :: Request -> Bool)-        ]-    , testGroup "JSON-RPC Responses"-        [ testProperty "Check fields"+        , testProperty "Check response fields"             (resFields :: Response -> Bool)-        , testProperty "Encode/decode"+        , testProperty "Encode/decode response"             (testEncodeDecode :: Response -> Bool)+        , testProperty "Encode/decode message"+            (testEncodeDecode :: Message -> Bool)         ]     , testGroup "Conduits"         [ testProperty "Encoding" testEncodeConduit         , testProperty "Decoding" testDecodeConduit         , testProperty "Decode Garbage" testDecodeGarbageConduit-        , testProperty "Decode Invalid" testDecodeInvalidConduit         ]-    , testGroup "JSON-RPC monad"+    , testGroup "Functions"         [ testProperty "Processing incoming messages" testProcessIncoming         , testProperty "Processing incoming responses" testIncomingResponse         , testProperty "Receiving incoming requests" testReceiveRequest+        , testProperty "Receiving incoming batch requests"+            testReceiveBatchRequest         , testProperty "Sending responses" testSendResponse+        , testProperty "Sending batch responses" testSendBatchResponse+        , testProperty "Sending request batch" testSendBatchRequest         ]     , testGroup "Network"-        [ testProperty "Test server" serverTest-        , testProperty "Test client" clientTest-        , testProperty "Test notifications" notifTest-        , testProperty "Test send request" testSendRequest+        [ testProperty "Client/server talk" clientTest+        , testProperty "Notifications" notifTest         ]     ] +data ReqList = ReqList { getReqList :: [Req] } deriving (Show, Eq)++instance Arbitrary ReqList where+    arbitrary = resize 3 $ ReqList <$> arbitrary+ data Req = Req { getReq :: Request } deriving (Show, Eq)  instance Arbitrary Req where@@ -87,7 +94,7 @@     requestIsNotif ReqNotReq{} = False     requestIsNotif ReqNotNot{} = True -data Struct = Struct { getValue :: Value } deriving (Show, Eq)+data Struct = Struct Value deriving (Show, Eq)  instance Arbitrary Struct where     arbitrary = resize 10 $ Struct <$> oneof [lsn, objn] where@@ -105,7 +112,16 @@         objn = toJSON . M.fromList <$> listOf psn         psn  = (,) <$> (arbitrary :: Gen String) <*> oneof [val, ls, obj] +reqFields :: Request -> Bool+reqFields rq = testFields (checkFieldsReq rq) rq +resFields :: Response -> Bool+resFields rs = testFields (checkFieldsRes rs) rs++fromRight :: Either a b -> b+fromRight (Right x) = x+fromRight _ = undefined+ checkVerId :: Ver -> Maybe Id -> Object -> Parser Bool checkVerId ver i o = do     j <- o .:? "jsonrpc"@@ -157,7 +173,7 @@ testDecodeConduit (msgs, ver) = monadicIO $ do     rt <- run $ runNoLoggingT $         CL.sourceList (buffer encoded) =$ decodeConduit ver $$ CL.consume-    assert $ rt == map Right msgs+    assert $ rt == map (Right . toJSON) msgs   where     encoded = B.concat $ map (L.toStrict . encode) msgs     buffer bs | B.null bs = []@@ -169,27 +185,15 @@         CL.sourceList (map B.pack ss) =$ decodeConduit ver $$ CL.consume     assert $ all isLeft rt -testDecodeInvalidConduit :: ([Struct], Ver) -> Property-testDecodeInvalidConduit (structs, ver) = monadicIO $ do-    rt <- run $ runNoLoggingT $-        CL.sourceList (buffer encoded) =$ decodeConduit ver $$ CL.consume-    assert $ all invalid rt-  where-    encoded = B.concat $ map (L.toStrict . encode) msgs-    buffer bs | B.null bs = []-              | otherwise = B.take 16 bs : buffer (B.drop 16 bs)-    msgs = map getValue structs-    invalid (Left OrphanError{getError = ErrorObj{getErrCode = (-32600)}}) =-        True-    invalid _ = False--testProcessIncoming :: ([Either Response Message], Ver, Bool) -> Property+testProcessIncoming :: ([Either Response Message], Ver, Bool)+                    -> Property testProcessIncoming (msgs, ver, ignore) = monadicIO $ do     rt <- run $ runNoLoggingT $ do         expect <- liftIO . atomically $ newTMChan         qs <- liftIO . atomically $ initSession ver ignore-        withAsync (cli qs expect msgs) $ const $+        withAsync (cli qs expect msgs) $ \c -> do             runReaderT processIncoming qs+            wait c         validate expect True     assert rt   where@@ -202,56 +206,81 @@         closeTMChan expect         closeTBMChan $ inCh qs     cli qs expect (x:xs) = do-        liftIO . atomically $ writeTBMChan (inCh qs) x+        liftIO . atomically $ writeTBMChan (inCh qs) (toJSON <$> x)         liftIO . atomically $ case x of-            Left e -> do-                m <- readTBMChan $ outCh qs-                case m of-                    Just (MsgResponse err) ->-                        if err == e-                            then writeTMChan expect True-                            else writeTMChan expect False-                    _ -> writeTMChan expect False+            Left e ->+                caseLeftResponse qs e >>= writeTMChan expect             Right (MsgRequest req@Request{}) ->+                caseSingleRequest req qs >>= writeTMChan expect+            Right (MsgRequest nt@Notif{}) ->+                caseSingleNotif nt qs >>= writeTMChan expect+            Right MsgResponse{} ->+                writeTMChan expect True+            Right (MsgBatch []) ->+                caseEmptyBatch qs >>= writeTMChan expect+            Right (MsgBatch ms) ->+                caseBatch qs ms >>= writeTMChan expect+        cli qs expect xs+    caseLeftResponse qs e = do+        m <- readTBMChan $ outCh qs+        case m of+            Just (MsgResponse err) -> return $ err == e+            _ -> return False+    caseEmptyBatch qs = do+        m <- readTBMChan $ outCh qs+        case m of+            Just (MsgResponse (OrphanError v (ErrorObj _ i _))) ->+                return $ v == rpcVer qs && i == (-32600)+            _ -> return False+    caseBatch qs ms = do+        let isrq (MsgRequest Request{}) = True+            isrq _ = False+            isnotif (MsgRequest Notif{}) = True+            isnotif _ = False+        if any isrq ms+            then                 if ignore                     then do                         m <- readTBMChan $ outCh qs                         case m of-                            Just (MsgResponse (ResponseError _ e _)) ->-                                case e of-                                    ErrorObj{} ->-                                        if getErrCode e == (-32601)-                                            then writeTMChan expect True-                                            else writeTMChan expect False-                                    _ -> writeTMChan expect False-                            _ -> writeTMChan expect False-                    else do-                        m <- readTBMChan (fromJust $ reqCh qs)-                        if m == Just req-                            then writeTMChan expect True-                            else writeTMChan expect False-            Right (MsgRequest nt@Notif{}) ->-                if ignore-                    then writeTMChan expect True+                            Just MsgBatch{} -> return True+                            _ -> return False                     else do-                        m <- readTBMChan (fromJust $ reqCh qs)-                        if m == Just nt-                            then writeTMChan expect True-                            else writeTMChan expect False-            Right (MsgResponse OrphanError{}) ->-                writeTMChan expect True-            Right MsgResponse{} -> do+                        m <- readTBMChan $ fromJust $ reqCh qs+                        case m of+                            Just BatchRequest{} -> return True+                            _ -> return False+            else+                if any isnotif ms+                    then+                        if ignore+                            then return True+                            else do+                                m <- readTBMChan $ fromJust $ reqCh qs+                                case m of+                                    Just BatchRequest{} -> return True+                                    _ -> return False+                    else return True+    caseSingleRequest req qs =+        if ignore+            then do                 m <- readTBMChan $ outCh qs                 case m of-                    Just (MsgResponse (OrphanError _ e@ErrorObj{})) ->-                        if getErrCode e == (-32000)-                            then writeTMChan expect True-                            else writeTMChan expect False-                    _ -> writeTMChan expect False-        cli qs expect xs+                    Just (MsgResponse (ResponseError _ e@ErrorObj{} _)) ->+                        return $ getErrCode e == (-32601)+                    _ -> return False+            else do+                m <- readTBMChan (fromJust $ reqCh qs)+                return $ m == Just (SingleRequest req)+    caseSingleNotif nt qs =+        if ignore+            then return True+            else do+                m <- readTBMChan (fromJust $ reqCh qs)+                return $ m == Just (SingleRequest nt) -testIncomingResponse :: ([Req], Ver, Bool) -> Property-testIncomingResponse (reqss, ver, ignore) = nodup ==> monadicIO $ do+testIncomingResponse :: ([ReqList], Ver, Bool) -> Property+testIncomingResponse (reqss', ver, ignore) = nodup ==> monadicIO $ do     rt <- run $ runNoLoggingT $ do         expect <- liftIO . atomically $ newTMChan         qs <- liftIO . atomically $ do@@ -267,14 +296,20 @@         return $ valid && mapempty     assert rt   where-    reqs = map getReq reqss-    nodup = length (nubBy ((==) `on` getReqId) reqs) == length reqs+    reqss = map getReqList reqss'+    reqs = let f [q] = SingleRequest (getReq q)+               f qs  = BatchRequest (map getReq qs)+           in map f reqss+    flatreqs = let f (BatchRequest bt) = bt+                   f (SingleRequest rt) = [rt]+               in concatMap f reqs+    nodup = length (nubBy ((==) `on` getReqId) flatreqs) == length flatreqs     respond (Request v _ p i) = Response v p i     respond _ = undefined-    responses = map respond reqs+    responses = map respond flatreqs     msgs = map MsgResponse responses     sent = do-        promises <- forM reqs $+        promises <- forM flatreqs $             \Request{getReqId = i} -> (,) i <$> newEmptyTMVar         return $ M.fromList promises     validate _ [] state = return state@@ -286,72 +321,93 @@     cli qs expect [] = liftIO . atomically $ do         closeTMChan expect         closeTBMChan $ inCh qs-    cli qs expect (x:xs) = do-        let i = getResId $ getMsgResponse x-        p <- liftIO . atomically $ do+    cli qs expect (x:xs) =+        case x of+            MsgResponse res -> do+                p <- getpromise qs res+                liftIO . atomically $ writeTBMChan (inCh qs) (Right $ toJSON x)+                liftIO . atomically $ do+                    rE <- readTMVar p+                    F.forM_ rE $ writeTMChan expect+                cli qs expect xs+            MsgBatch resps -> do+                ps <- mapM (getpromise qs . getMsgResponse) resps+                liftIO . atomically $ writeTBMChan (inCh qs) (Right $ toJSON x)+                forM_ ps $ \p -> liftIO . atomically $ do+                    rE <- readTMVar p+                    F.forM_ rE $ writeTMChan expect+            _ -> undefined+    getpromise qs res = do+        let i = getResId res+        liftIO . atomically $ do             snt <- readTVar (sentReqs qs)             return . fromJust $ M.lookup i snt-        liftIO . atomically $ writeTBMChan (inCh qs) (Right x)-        liftIO . atomically $ do-            rE <- readTMVar p-            F.forM_ rE $ writeTMChan expect-        cli qs expect xs -testSendRequest :: ([(ReqNot, Either ErrorObj Value)], Ver, Bool) -> Property-testSendRequest (reqnotres, ver, ignore) = monadicIO $ do-    (ex, rs) <- run $ runNoLoggingT $ do-        expect <- liftIO . atomically $ newTMChan+testSendBatchRequest :: ([(ReqNot, Either ErrorObj Value)], Ver, Bool)+                     -> Property+testSendBatchRequest (reqnotres, ver, ignore) = nonull ==> monadicIO $ do+    rs <- run $ runNoLoggingT $ do         qs <- liftIO . atomically $ initSession ver ignore-        rs <- withAsync (cli qs expect reqnotres) $ \c -> do-            rs <- runReaderT (mapM (sendRequest . fst) reqnotres) qs+        withAsync (cli qs) $ \c -> do+            rs <- runReaderT (sendBatchRequest $ map fst reqnotres) qs             wait c             return (rs :: [Maybe (Either ErrorObj Value)])-        ex <- liftIO . atomically $ getexpect expect []-        return (ex, rs)     assert $ length rs == length reqnotres-    assert $ and $ zipWith matchresult reqnotres rs-    assert $ and $ zipWith matchnotif (map fst reqnotres) ex   where-    matchnotif q r =-        case q of-            ReqNotReq v ->-                case r of-                    Just (MsgRequest Request{getReqParams = v'}) -> v == v'-                    _ -> False-            ReqNotNot v ->-                case r of-                    Just (MsgRequest Notif{getReqParams = v'}) -> v == v'-                    _ -> False-    matchresult (q, e) r =-        case q of-            ReqNotReq{} -> Just e == r-            _ -> isNothing r-    getexpect expect acc = do-        valM <- readTMChan expect-        case valM of-            Nothing -> return $ reverse acc-            Just val -> getexpect expect (val:acc)+    nonull = not $ null reqnotres+    resmap = M.fromList $ map (first toJSON) reqnotres     respond (Request v _ _ i) (Left  y) = Just $ ResponseError v y i     respond (Request v _ _ i) (Right y) = Just $ Response      v y i     respond _ _ = undefined-    cli _ expect [] = liftIO . atomically $ closeTMChan expect-    cli qs expect ((x,y):xs) = join . liftIO . atomically $ do+    fulfill req sent = F.forM_ (getReqId req `M.lookup` sent) $ \p ->+        putTMVar p $ respond req $ fromJust $ getReqParams req `M.lookup` resmap+    cli qs = liftIO . atomically $ do         msg <- readTBMChan (outCh qs)-        writeTMChan expect msg-        unless (requestIsNotif x) $-            case msg of-                Just (MsgRequest req@Request{}) -> do-                    sent <- readTVar (sentReqs qs)-                    F.forM_ (getReqId req `M.lookup` sent) $ \p ->-                        putTMVar p $ respond req y-                _ -> return ()-        return $ cli qs expect xs+        sent <- readTVar (sentReqs qs)+        case msg of+            Just (MsgRequest req@Request{}) -> fulfill req sent+            Just (MsgBatch bt) ->+                forM_ bt $ \m ->+                    case m of+                        MsgRequest req@Request{} -> fulfill req sent+                        _ -> return ()+            _ -> return () -testReceiveRequest :: ([Request], Ver, Bool) -> Property+testReceiveRequest :: ([BatchRequest], Ver, Bool) -> Property testReceiveRequest (reqs, ver, ignore) = monadicIO $ do     rt <- run $ runNoLoggingT $ do         qs <- liftIO . atomically $ initSession ver ignore         withAsync (send qs) $ \c -> do+            res <- runReaderT receive qs+            wait c+            return res+    assert $ and rt+  where+    send qs = F.forM_ (reqCh qs) $ \qch -> do+        forM_ reqs $ liftIO . atomically . writeTBMChan qch+        liftIO . atomically $ closeTBMChan qch+    receive = forM reqs $ \q -> do+        t <- receiveRequest+        if ignore+            then return $ isNothing t+            else+                case q of+                    SingleRequest{} ->+                        return $ Just q == fmap SingleRequest t+                    BatchRequest{} -> do+                        ch <- reader outCh+                        m <- liftIO . atomically $ readTBMChan ch+                        case m of+                            Just (MsgResponse OrphanError{}) -> +                                return $ isNothing t+                            _ -> return $ isNothing t+++testReceiveBatchRequest :: ([BatchRequest], Ver, Bool) -> Property+testReceiveBatchRequest (reqs, ver, ignore) = monadicIO $ do+    rt <- run $ runNoLoggingT $ do+        qs <- liftIO . atomically $ initSession ver ignore+        withAsync (send qs) $ \c -> do             rt <- runReaderT receive qs             wait c             return rt@@ -362,7 +418,7 @@     send qs = F.forM_ (reqCh qs) $ \qch -> do         forM_ reqs $ liftIO . atomically . writeTBMChan qch         liftIO . atomically $ closeTBMChan qch-    receive = forM reqs $ const receiveRequest+    receive = forM reqs $ const receiveBatchRequest  testSendResponse :: ([Response], Ver, Bool) -> Property testSendResponse (responses, ver, ignore) = monadicIO $ do@@ -378,11 +434,21 @@     receive qs = forM responses $ const $ liftIO . atomically $         readTBMChan (outCh qs) -reqFields :: Request -> Bool-reqFields rq = testFields (checkFieldsReq rq) rq--resFields :: Response -> Bool-resFields rs = testFields (checkFieldsRes rs) rs+testSendBatchResponse :: ([BatchResponse], Ver, Bool) -> Property+testSendBatchResponse (responses, ver, ignore) = monadicIO $ do+    ex <- run $ runNoLoggingT $ do+        qs <- liftIO . atomically $ initSession ver ignore+        withAsync (receive qs) $ \c -> do+            runReaderT send qs+            wait c+    assert $ length ex == length responses+    assert $ all isJust ex && map fromJust ex == map msg responses+  where+    msg (BatchResponse bt) = MsgBatch $ map MsgResponse bt+    msg (SingleResponse r) = MsgResponse r+    send = forM_ responses sendBatchResponse+    receive qs = forM responses $ const $ liftIO . atomically $+        readTBMChan (outCh qs)  createChans :: MonadIO m             => m ((TBMChan a, TBMChan b), (Sink a m (), Source m b))@@ -391,43 +457,13 @@     let (snk, src) = (sinkTBMChan bso False, sourceTBMChan bsi)     return ((bso, bsi), (snk, src)) -serverTest :: ([Request], Ver) -> Property-serverTest (reqs, ver) = monadicIO $ do-    rt <- run $ runNoLoggingT $ do-        ((bso, bsi), (snk, src)) <- createChans-        withAsync (server snk src) $ const $-            withAsync (sender bsi) $ const $ receiver bso []-    assert $ length rt == length nonotif-    assert $ null rt || all isJust rt-    assert $ params == reverse (results rt)-  where-    respond q = return $ Right (q :: Value)-    server snk src = runJsonRpcT ver False-        (encodeConduit =$ snk) (src =$ decodeConduit ver) (srv respond)-    sender bsi = forM_ reqs $ liftIO . atomically .-        writeTBMChan bsi . L.toStrict . encode . MsgRequest-    receiver bso xs =-        if length xs == length nonotif-            then return xs-            else liftIO (atomically $ readTBMChan bso) >>= \b -> case b of-                Just x -> do-                    let res = decodeStrict' x :: Maybe Response-                    receiver bso (res:xs)-                Nothing -> undefined-    params = map getReqParams nonotif-    results = map $ getResult . fromJust-    nonotif = flip filter reqs $ \q -> case q of Request{} -> True-                                                 Notif{}   -> False- clientTest :: ([Value], Ver) -> Property clientTest (qs, ver) = monadicIO $ do     rt <- run $ runNoLoggingT $ do         ((bso, bsi), (snk, src)) <- createChans         let csnk = sinkTBMChan bsi False             csrc = sourceTBMChan bso-        withAsync (server snk src) $ const $ cli-            (CL.map Right =$ csnk)-            (csrc $= CL.map Right)+        withAsync (server snk src) $ const $ cli csnk csrc     assert $ length rt == length qs     assert $ null rt || all correct rt     assert $ qs == results rt@@ -447,16 +483,14 @@             csrc = sourceTBMChan bso         (sig, notifs) <- liftIO . atomically $             (,) <$> newEmptyTMVar <*> newTVar []-        withAsync (server snk src sig notifs) $ const $ cli sig-            (CL.map Right =$ csnk)-            (csrc $= CL.map Right)+        withAsync (server snk src sig notifs) $ const $ cli sig csnk csrc         liftIO . atomically $ readTVar notifs     assert $ length nt == length ntfs     assert $ reverse nt == ntfs   where     respond q = return $ Right (q :: Value)     server snk src sig notifs =-        runJsonRpcT ver False snk src $ process sig notifs+        runJsonRpcT ver False snk src (process sig notifs)     process sig notifs = do         qM <- receiveRequest         case qM of@@ -491,8 +525,3 @@             case rM of                 Nothing -> srv respond                 Just r -> sendResponse r >> srv respond---fromRight :: Either a b -> b-fromRight (Right x) = x-fromRight _ = undefined