diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Revision history for Z-MessagePack +## 0.4.1.0 -- 2021-04-25 + +* Adapt to `Z-IO` 0.8. Add stream error handling code. + ## 0.3.0.1 -- 2021-02-25 * Add `callStream` and `StreamHandler`. diff --git a/Z-MessagePack.cabal b/Z-MessagePack.cabal --- a/Z-MessagePack.cabal +++ b/Z-MessagePack.cabal @@ -1,6 +1,6 @@ cabal-version: >=1.10 name: Z-MessagePack -version: 0.4.0.1 +version: 0.4.1.0 synopsis: MessagePack description: MessagePack binary serialization format. license: BSD3 @@ -14,6 +14,10 @@ bug-reports: https://github.com/ZHaskell/z-messagepack/issues extra-source-files: CHANGELOG.md +source-repository head + type: git + location: git://github.com/ZHaskell/z-messagepack.git + library exposed-modules: Z.Data.MessagePack @@ -34,8 +38,8 @@ , tagged >=0.8 && <0.9 , time >=1.9 && <2.0 , unordered-containers >=0.2 && <0.3 - , Z-Data >=0.6 && <1.0 - , Z-IO >=0.6 && <1.0 + , Z-Data >=0.8 && <0.9 + , Z-IO >=0.8 && <0.9 default-language: Haskell2010 default-extensions: diff --git a/Z/Data/MessagePack.hs b/Z/Data/MessagePack.hs --- a/Z/Data/MessagePack.hs +++ b/Z/Data/MessagePack.hs @@ -67,10 +67,10 @@ MessagePack(..), Value(..), defaultSettings, Settings(..), JSON.snakeCase, JSON.trainCase -- * Encode & Decode , readMessagePackFile, writeMessagePackFile - , decode, decode', decodeChunks, encode, encodeChunks + , decode, decode', decodeChunk, decodeChunks, encode, encodeChunks , DecodeError, ParseError -- * parse into MessagePack Value - , parseValue, parseValue', parseValueChunks, parseValueChunks' + , parseValue, parseValue' -- * Generic FromValue, ToValue & EncodeMessagePack , gToValue, gFromValue, gEncodeMessagePack -- * Convert 'Value' to Haskell data diff --git a/Z/Data/MessagePack/Base.hs b/Z/Data/MessagePack/Base.hs --- a/Z/Data/MessagePack/Base.hs +++ b/Z/Data/MessagePack/Base.hs @@ -15,10 +15,10 @@ ( -- * MessagePack Class MessagePack(..), Value(..), defaultSettings, Settings(..) -- * Encode & Decode - , decode, decode', decodeChunks, encode, encodeChunks + , decode, decode', decodeChunk, decodeChunks, encode, encodeChunks , DecodeError, P.ParseError, P.ParseChunks -- * parse into MessagePack Value - , MV.parseValue, MV.parseValue', MV.parseValueChunks, MV.parseValueChunks' + , MV.parseValue, MV.parseValue' -- * Generic FromValue, ToValue & EncodeMessagePack , gToValue, gFromValue, gEncodeMessagePack -- * Convert 'Value' to Haskell data @@ -145,17 +145,24 @@ Left cErr -> (bs', Left (Right cErr)) Right r -> (bs', Right r) +-- | Decode a MessagePack doc chunk. +decodeChunk :: MessagePack a => V.Bytes -> P.Result DecodeError a +{-# INLINE decodeChunk #-} +decodeChunk bs = loop (P.parseChunk MV.value bs) + where + loop r = do + case r of + P.Success v rest -> + case convertValue v of + Left cErr -> P.Failure (Right cErr) rest + Right r' -> P.Success r' rest + P.Failure e rest -> P.Failure (Left e) rest + P.Partial f' -> P.Partial (loop . f') + -- | Decode MessagePack doc chunks, return trailing bytes. -decodeChunks :: (MessagePack a, Monad m) => m V.Bytes -> V.Bytes -> m (V.Bytes, Either DecodeError a) +decodeChunks :: (MessagePack a, Monad m) => P.ParseChunks m DecodeError a {-# INLINE decodeChunks #-} -decodeChunks mb bs = do - mr <- P.parseChunks MV.value mb bs - case mr of - (bs', Left pErr) -> pure (bs', Left (Left pErr)) - (bs', Right v) -> - case convertValue v of - Left cErr -> pure (bs', Left (Right cErr)) - Right r -> pure (bs', Right r) +decodeChunks = P.parseChunks decodeChunk -- | Directly encode data to MessagePack bytes. encode :: MessagePack a => a -> V.Bytes diff --git a/Z/Data/MessagePack/Value.hs b/Z/Data/MessagePack/Value.hs --- a/Z/Data/MessagePack/Value.hs +++ b/Z/Data/MessagePack/Value.hs @@ -11,8 +11,6 @@ -- * parse into MessagePack Value , parseValue , parseValue' - , parseValueChunks - , parseValueChunks' -- * Value Parsers , value ) where @@ -153,13 +151,3 @@ parseValue' :: V.Bytes -> Either P.ParseError Value {-# INLINE parseValue' #-} parseValue' = P.parse' (value <* P.endOfInput) - --- | Increamental parse 'Value' without consuming trailing bytes. -parseValueChunks :: Monad m => m V.Bytes -> V.Bytes -> m (V.Bytes, Either P.ParseError Value) -{-# INLINE parseValueChunks #-} -parseValueChunks = P.parseChunks value - --- | Increamental parse 'Value', if there're bytes left, parsing will fail. -parseValueChunks' :: Monad m => m V.Bytes -> V.Bytes -> m (Either P.ParseError Value) -{-# INLINE parseValueChunks' #-} -parseValueChunks' mi inp = snd <$> P.parseChunks (value <* P.endOfInput) mi inp diff --git a/Z/IO/RPC/MessagePack.hs b/Z/IO/RPC/MessagePack.hs --- a/Z/IO/RPC/MessagePack.hs +++ b/Z/IO/RPC/MessagePack.hs @@ -228,7 +228,7 @@ MP.encodeMessagePack req -- param flushBuffer bo return (sendEOF, sourceFromIO $ do - res <- pull (sourceParserFromBuffered (do + res <- readParser (do tag <- P.anyWord8 -- stream stop case tag of @@ -245,10 +245,10 @@ P.fail' $ "wrong response type: " <> T.toText typ return (Just (err, v)) _ -> P.fail' $ "wrong response tag: " <> T.toText tag - ) bi) + ) bi -- we take tcp disconnect as eof too - case join res of + case res of Just (err, v) -> do when (err /= MV.Nil) $ throwIO (RPCException err callStack) unwrap "EPARSE" (MP.convertValue v) @@ -292,8 +292,10 @@ => (SessionCtx a -> req -> IO res) -> ServerHandler a NotifyHandler :: MessagePack req => (SessionCtx a -> req -> IO ()) -> ServerHandler a + -- | 'StreamHandler' will receive an 'IORef' which get updated to 'True' + -- when client send stream end packet, stream should end up ASAP. StreamHandler :: (MessagePack req, MessagePack res) - => (SessionCtx a -> req -> IO (Source res)) -> ServerHandler a + => (SessionCtx a -> IORef Bool -> req -> IO (Source res)) -> ServerHandler a -- | Simple router using `FlatMap`, lookup name in /O(log(N))/. -- @@ -337,15 +339,15 @@ loop ctx bi bo where loop ctx bi bo = do - req <- pull $ sourceParserFromBuffered sourceParser bi + req <- readParser sourceParser bi case req of - Just (Notify (name, v)) -> do + Notify (name, v) -> do case handle name of Just (NotifyHandler f) -> do f ctx =<< unwrap "EPARSE" (MP.convertValue v) _ -> throwOtherError "ENOTFOUND" "notification method not found" loop ctx bi bo - Just (Call (msgid, name, v)) -> do + Call (msgid, name, v) -> do case handle name of Just (CallHandler f) -> do res <- try (f ctx =<< unwrap "EPARSE" (MP.convertValue v)) @@ -370,11 +372,11 @@ MB.nil flushBuffer bo loop ctx bi bo - Just (StreamStart (name, v)) -> do + StreamStart (name, v) -> do eofRef <- newIORef False -- fork new thread to get stream end notification forkIO $ do - pull (sourceParserFromBuffered (do + _ <- readParser (do tag <- P.anyWord8 -- stream stop when (tag /= 0x91) $ @@ -382,49 +384,44 @@ !typ <- MV.value when (typ /= MV.Int 5) $ P.fail' $ "wrong request type: " <> T.toText typ - ) bi) + ) bi atomicWriteIORef eofRef True case handle name of - Just (StreamHandler f) -> do - src <- f ctx =<< unwrap "EPARSE" (MP.convertValue v) - loopSend eofRef src bo - _ -> do - writeBuilder bo $ do - MB.arrayHeader 3 - MB.int 6 -- type response - MB.str $ "request method: " <> name <> " not found" - MB.nil - flushBuffer bo + Just (StreamHandler f) -> (do + src <- f ctx eofRef =<< unwrap "EPARSE" (MP.convertValue v) + src (writeItem bo) EOF) `catch` (\ (e :: SomeException) -> + writeErrorItem bo $ "error when stream: " <> T.toText e) + _ -> writeErrorItem bo $ "request method: " <> name <> " not found" loop ctx bi bo - _ -> return () - + writeItem bo = \ mx -> do + case mx of + Just x -> do + writeBuilder bo $ do + MB.arrayHeader 3 + MB.int 6 -- type stream item + MB.nil + MP.encodeMessagePack x + flushBuffer bo + _ -> do + writeBuilder bo $ do + MB.arrayHeader 1 + MB.int 7 -- type stream end + flushBuffer bo - loopSend eofRef src bo = do - eof <- readIORef eofRef - if eof - then do - writeBuilder bo $ do - MB.arrayHeader 1 - MB.int 7 -- type response - flushBuffer bo - else do - r <- pull src - case r of - Just r' -> do - writeBuilder bo $ do - MB.arrayHeader 3 - MB.int 6 -- type response - MB.nil - MP.encodeMessagePack r' - flushBuffer bo - _ -> atomicWriteIORef eofRef True - loopSend eofRef src bo + writeErrorItem bo msg = do + writeBuilder bo $ do + MB.arrayHeader 3 + MB.int 6 -- type stream item + MB.str msg + MB.nil + flushBuffer bo ------------------------------------------------------------------------------- sourceParser :: P.Parser (Request MV.Value) +{-# INLINE sourceParser #-} sourceParser = do tag <- P.anyWord8 case tag of @@ -455,7 +452,6 @@ _ -> P.fail' $ "wrong msgid: " <> T.toText seq_ _ -> P.fail' $ "wrong request type: " <> T.toText typ _ -> P.fail' $ "wrong request tag: " <> T.toText tag -{-# INLINE sourceParser #-} -- $server-example -- @@ -481,8 +477,16 @@ -- > , ("bar", CallHandler $ \ctx (req :: T.Text) -> do -- > counter . fromJust <$> readSessionCtx ctx -- > ) --- > , ("qux", StreamHandler $ \ctx (_ :: ()) -> do --- > withMVar stdinBuf (pure . sourceFromBuffered) +-- > , ("qux", StreamHandler $ \ctx eofRef (_ :: ()) -> do +-- > withMVar stdinBuf (\ stdin -> pure $ \ k _ -> do +-- eof <- readIORef eofRef +-- if eof +-- then k EOF +-- else do +-- r <- readBuffer stdin +-- if V.null r +-- then k EOF +-- else k (Just r)) -- > ) -- > ]