diff --git a/Network/JsonRpc/Arbitrary.hs b/Network/JsonRpc/Arbitrary.hs --- a/Network/JsonRpc/Arbitrary.hs +++ b/Network/JsonRpc/Arbitrary.hs @@ -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] diff --git a/Network/JsonRpc/Data.hs b/Network/JsonRpc/Data.hs --- a/Network/JsonRpc/Data.hs +++ b/Network/JsonRpc/Data.hs @@ -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) diff --git a/Network/JsonRpc/Interface.hs b/Network/JsonRpc/Interface.hs --- a/Network/JsonRpc/Interface.hs +++ b/Network/JsonRpc/Interface.hs @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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 ``` diff --git a/json-rpc.cabal b/json-rpc.cabal --- a/json-rpc.cabal +++ b/json-rpc.cabal @@ -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 diff --git a/test/Network/JsonRpc/Tests.hs b/test/Network/JsonRpc/Tests.hs --- a/test/Network/JsonRpc/Tests.hs +++ b/test/Network/JsonRpc/Tests.hs @@ -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