mu-grpc-client 0.4.0.1 → 0.4.0.2
raw patch · 3 files changed
+48/−44 lines, 3 filesdep +tracing-controldep −tracingsetup-changedPVP ok
version bump matches the API change (PVP)
Dependencies added: tracing-control
Dependencies removed: tracing
API changes (from Hackage documentation)
Files
- Setup.hs +1/−1
- mu-grpc-client.cabal +8/−8
- src/Mu/GRpc/Client/Internal.hs +39/−35
Setup.hs view
@@ -1,2 +1,2 @@-import Distribution.Simple+import Distribution.Simple main = defaultMain
mu-grpc-client.cabal view
@@ -1,5 +1,5 @@ name: mu-grpc-client-version: 0.4.0.1+version: 0.4.0.2 synopsis: gRPC clients from Mu definitions description: With @mu-grpc-client@ you can easily build gRPC clients for mu-haskell!@@ -29,7 +29,7 @@ other-modules: Mu.GRpc.Client.Internal build-depends:- async >=2.2 && < 3+ async >=2.2 && <3 , avro >=0.5.1 && <0.6 , base >=4.12 && <5 , bytestring >=0.10 && <0.11@@ -38,11 +38,11 @@ , http2-client >=0.9 && <1 , http2-client-grpc >=0.8 && <0.9 , http2-grpc-types >=0.5 && <0.6- , mu-grpc-common ==0.4.*- , mu-optics ==0.3.*- , mu-protobuf ==0.4.*- , mu-rpc ==0.4.*- , mu-schema ==0.3.*+ , mu-grpc-common >=0.4 && <0.5+ , mu-optics >=0.3 && <0.4+ , mu-protobuf >=0.4 && <0.5+ , mu-rpc >=0.4 && <0.6+ , mu-schema >=0.3 && <0.4 , optics-core >=0.2 && <0.4 , sop-core >=0.5 && <0.6 , stm >=2.5 && <3@@ -51,7 +51,7 @@ , template-haskell >=2.14 && <2.17 , text >=1.2 && <2 , th-abstraction >=0.3.2 && <0.5- , tracing >=0.0.5+ , tracing-control >=0.0.7 hs-source-dirs: src default-language: Haskell2010
src/Mu/GRpc/Client/Internal.hs view
@@ -21,6 +21,7 @@ import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TMChan import Control.Concurrent.STM.TMVar+import Control.Exception (throwIO) import Control.Monad.IO.Class import Data.Avro import qualified Data.ByteString.Char8 as BS@@ -38,7 +39,7 @@ import Network.GRPC.HTTP2.Encoding (GRPCInput, GRPCOutput) import Network.HTTP2 (ErrorCode) import Network.HTTP2.Client (ClientError, ClientIO, TooMuchConcurrency,- runExceptT)+ runExceptT, ExceptT) import Mu.Adapter.ProtoBuf.Via import Mu.GRpc.Avro@@ -304,47 +305,50 @@ instance ( KnownName name , GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r- , handler ~ (CompressMode -> IO (ConduitT v (GRpcReply r) IO ())) )+ , handler ~ (CompressMode -> IO (ConduitT v Void IO (), ConduitT () r IO (GRpcReply ())))) => GRpcMethodCall p ('Method name '[ 'ArgStream aname vref ] ('RetStream rref)) handler where gRpcMethodCall rpc _ client compress- = do -- Create a new TMChan- inchan <- newTMChanIO :: IO (TMChan (GRpcReply r))- outchan <- newTMChanIO :: IO (TMChan v)- var <- newEmptyTMVarIO -- if full, this means an error+ = do serverChan <- newTMChanIO :: IO (TMChan r)+ clientChan <- newTMChanIO :: IO (TMChan v)+ finalReply <- newEmptyTMVarIO :: IO (TMVar (GRpcReply ())) -- Start executing the client in another thread+ -- TODO: Is there anything that makes sure that this thread doesn't keep running forever? _ <- async $ do v <- simplifyResponse $ buildGRpcReply3 <$> rawGeneralStream @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) rpc client- () (\_ ievent -> do -- on the first iteration, say that everything is OK- _ <- liftIO $ atomically $ tryPutTMVar var (GRpcOk ())- case ievent of- RecvMessage o -> liftIO $ atomically $ writeTMChan inchan (GRpcOk $ unGRpcOWTy(Proxy @p) (Proxy @rref) o)- Invalid e -> liftIO $ atomically $ writeTMChan inchan (GRpcErrorString (show e))- _ -> pure () )- () (\_ -> do- nextVal <- liftIO $ atomically $ readTMChan outchan- case nextVal of- Nothing -> pure ((), Finalize)- Just v -> pure ((), SendMessage compress (buildGRpcIWTy (Proxy @p) (Proxy @vref) v)))- case v of- GRpcOk () -> liftIO $ atomically $ closeTMChan inchan- _ -> liftIO $ atomically $ putTMVar var v- -- This conduit feeds information to the other thread- let go = do err <- liftIO $ atomically $ takeTMVar var- case err of- GRpcOk _ -> go2- e -> yield $ (\_ -> error "this should never happen") <$> e- go2 = do nextOut <- await- case nextOut of- Just v -> do liftIO $ atomically $ writeTMChan outchan v- go2- Nothing -> do r <- liftIO $ atomically $ tryReadTMChan inchan- case r of- Nothing -> pure () -- both are empty, end- Just Nothing -> go2- Just (Just nextIn) -> yield nextIn >> go2- pure go+ () (incomingEventConsumer serverChan)+ () (outgoingEventProducer clientChan)+ liftIO $ atomically $ putTMVar finalReply v+ let clientConduit = do+ sinkTMChan clientChan+ liftIO . atomically . closeTMChan $ clientChan+ serverConduit = do+ sourceTMChan serverChan+ liftIO . atomically . readTMVar $ finalReply+ pure (clientConduit, serverConduit)+ where+ incomingEventConsumer :: TMChan r -> () -> IncomingEvent (GRpcOWTy p rref r) () -> ExceptT ClientError IO ()+ incomingEventConsumer serverChan _ ievent =+ case ievent of+ RecvMessage o -> do+ liftIO $ atomically $ writeTMChan serverChan (unGRpcOWTy (Proxy @p) (Proxy @rref) o)+ Invalid e -> liftIO $ do+ atomically $ closeTMChan serverChan+ throwIO e+ Trailers _ ->+ -- TODO: Read the trailers and use them to make the 'finalReply'+ liftIO $ atomically $ closeTMChan serverChan+ Headers _ ->+ -- TODO: Read the headers and use them to make the 'finalReply'+ pure ()++ outgoingEventProducer :: TMChan v -> () -> ExceptT ClientError IO ((), OutgoingEvent (GRpcIWTy p vref v) ())+ outgoingEventProducer clientChan _ = do+ nextVal <- liftIO $ atomically $ readTMChan clientChan+ case nextVal of+ Nothing -> pure ((), Finalize)+ Just v -> pure ((), SendMessage compress (buildGRpcIWTy (Proxy @p) (Proxy @vref) v))