diff --git a/Setup.hs b/Setup.hs --- a/Setup.hs +++ b/Setup.hs @@ -1,2 +1,2 @@ -import Distribution.Simple +import Distribution.Simple main = defaultMain diff --git a/mu-grpc-client.cabal b/mu-grpc-client.cabal --- a/mu-grpc-client.cabal +++ b/mu-grpc-client.cabal @@ -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 diff --git a/src/Mu/GRpc/Client/Internal.hs b/src/Mu/GRpc/Client/Internal.hs --- a/src/Mu/GRpc/Client/Internal.hs +++ b/src/Mu/GRpc/Client/Internal.hs @@ -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))