diff --git a/Network/JsonRpc/Conduit.hs b/Network/JsonRpc/Conduit.hs --- a/Network/JsonRpc/Conduit.hs +++ b/Network/JsonRpc/Conduit.hs @@ -24,14 +24,17 @@ import Control.DeepSeq import Control.Monad import Control.Monad.Trans +import Control.Monad.Trans.State import Data.Aeson -import Data.Aeson.Types +import Data.Aeson.Types (parseEither) +import Data.Attoparsec.ByteString import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import qualified Data.ByteString.Char8 as B8 import qualified Data.ByteString.Lazy.Char8 as L8 import Data.HashMap.Strict (HashMap) import qualified Data.HashMap.Strict as M import Data.Conduit -import qualified Data.Conduit.Binary as CB import qualified Data.Conduit.List as CL import Data.Conduit.Network import Data.Conduit.TMChan @@ -79,7 +82,7 @@ -- | Conduit that serializes JSON documents for sending to the network. encodeConduit :: (ToJSON a, Monad m) => Conduit a m ByteString -encodeConduit = CL.map (L8.toStrict . flip L8.append "\n" . encode) +encodeConduit = CL.map (L8.toStrict . encode) -- | Conduit for outgoing JSON-RPC messages. -- Adds an id to requests whose id is 'IdNull'. @@ -125,35 +128,59 @@ -> Session qo -- ^ JSON-RPC session -> Conduit ByteString m (IncomingMsg qo qi ni ri) -- ^ Decoded incoming messages -decodeConduit ver c qs = CB.lines =$= f True where +decodeConduit ver c qs = evalStateT (f True) Nothing where f re = if c && re then do l <- liftIO . atomically $ readTQueue (isLast qs) - if l then return () else g - else g + if l then return () else loop + else loop - g = await >>= \bsM -> case bsM of - Nothing -> - return () - Just bs -> do - msg <- liftIO . atomically $ decodeSTM bs - yield msg >> f (re msg) + loop = lift await >>= maybe flush process + + process = runParser >=> handle + + flush = do + p <- get + case p of + Nothing -> return () + Just k -> handle (k B.empty) + + runParser chunk = do + p <- getPartialParser + return $ case p of + Nothing -> parse json' chunk + Just k -> k chunk + + getPartialParser = get <* put Nothing + + handle (Fail {}) = lift (yield . IncomingError $ errorParse ver Null) + handle (Partial k) = put (Just k) >> loop + handle (Done rest v) = do + msg <- liftIO . atomically $ decodeSTM v + lift $ yield msg + if B.null rest + then f (re msg) + else process rest where re (IncomingMsg _ (Just _)) = True re _ = False - decodeSTM bs = readTVar (sentRequests qs) >>= \h -> case decodeMsg h bs of - Right m@(MsgResponse rs) -> do - rq <- requestSTM h $ getResId rs - return $ IncomingMsg m (Just rq) - Right m@(MsgError re) -> case getErrId re of - IdNull -> - return $ IncomingMsg m Nothing - i -> do - rq <- requestSTM h i + decodeSTM v = do + h <- readTVar (sentRequests qs) + case parseEither (topParse h) v of + Left _ -> return . IncomingError $ errorParse ver Null + Right x -> case x of + Right m@(MsgResponse rs) -> do + rq <- requestSTM h $ getResId rs return $ IncomingMsg m (Just rq) - Right m -> return $ IncomingMsg m Nothing - Left e -> return $ IncomingError e + Right m@(MsgError re) -> case getErrId re of + IdNull -> + return $ IncomingMsg m Nothing + i -> do + rq <- requestSTM h i + return $ IncomingMsg m (Just rq) + Right m -> return $ IncomingMsg m Nothing + Left e -> return $ IncomingError e requestSTM h i = do let rq = fromJust $ i `M.lookup` h @@ -161,12 +188,6 @@ writeTVar (sentRequests qs) h' return rq - decodeMsg h bs = case eitherDecodeStrict' bs of - Left _ -> Left $ errorParse ver Null - Right v -> case parseEither (topParse h) v of - Left _ -> Left $ errorParse ver Null - Right x -> x - topParse h v = parseReq v <|> parseNot v <|> parseRes h v @@ -255,7 +276,10 @@ -- ^ JSON-RPC action -> IO a -- ^ Output of action tcpClient ver d cs f = runTCPClient cs $ \ad -> do - runConduits ver d (appSink ad) (appSource ad) f + runConduits ver d (cr =$ appSink ad) (appSource ad $= ln) f + where + cr = CL.map (flip B8.snoc '\n') + ln = CL.mapFoldable B8.lines -- JSON-RPC-over-TCP server. tcpServer :: ( FromRequest qi, FromNotif ni, FromResponse ri @@ -266,5 +290,7 @@ -- ^ JSON-RPC action to perform on connecting client thread -> IO () tcpServer ver ss f = runTCPServer ss $ \cl -> do - runConduits ver False (appSink cl) (appSource cl) f + runConduits ver False (cr =$ appSink cl) (appSource cl) f + where + cr = CL.map (flip B8.snoc '\n') 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.2.1.2 +version: 0.2.1.4 synopsis: Fully-featured JSON-RPC 2.0 library description: This JSON-RPC library is fully-compatible with JSON-RPC 2.0 and @@ -17,12 +17,22 @@ extra-source-files: README.md cabal-version: >= 1.10 +source-repository head + type: git + location: https://github.com/xenog/json-rpc.git + +source-repository this + type: git + location: https://github.com/xenog/json-rpc.git + tag: v0.2.1.4 + library exposed-modules: Network.JsonRpc other-modules: Network.JsonRpc.Data, Network.JsonRpc.Conduit build-depends: base >= 4.6 && < 5, aeson >= 0.7 && < 0.9, + attoparsec >= 0.11, async >= 2.0 && < 2.1, bytestring >= 0.10 && < 0.11, conduit >= 1.2 && < 1.3, @@ -33,6 +43,7 @@ stm >= 2.4 && < 2.5, stm-conduit >= 2.5 && < 2.6, text >= 0.11 && < 1.3, + transformers >= 0.3, unordered-containers >= 0.2 && < 0.3 default-language: Haskell2010 ghc-options: -Wall