diff --git a/mu-grpc-server.cabal b/mu-grpc-server.cabal --- a/mu-grpc-server.cabal +++ b/mu-grpc-server.cabal @@ -1,5 +1,5 @@ name: mu-grpc-server -version: 0.2.0.0 +version: 0.3.0.0 synopsis: gRPC servers for Mu definitions description: With @mu-grpc-server@ you can easily build gRPC servers for mu-haskell! @@ -24,22 +24,23 @@ exposed-modules: Mu.GRpc.Server build-depends: async - , base >=4.12 && <5 + , avro >=0.5 + , base >=4.12 && <5 , binary , bytestring , conduit , http2-grpc-types , mtl - , mu-grpc-common >=0.2.0 - , mu-protobuf >=0.2.0 - , mu-rpc >=0.2.0 - , mu-schema >=0.2.0 + , mu-grpc-common >=0.3.0 + , mu-protobuf >=0.3.0 + , mu-rpc >=0.3.0 + , mu-schema >=0.3.0 , sop-core , stm , stm-conduit , wai , warp - , warp-grpc + , warp-grpc >=0.4.0.1 , warp-tls hs-source-dirs: src @@ -51,22 +52,23 @@ other-modules: Mu.GRpc.Server build-depends: async - , base >=4.12 && <5 + , avro >=0.4.7 + , base >=4.12 && <5 , binary , bytestring , conduit , http2-grpc-types , mtl - , mu-grpc-common >=0.2.0 - , mu-protobuf >=0.2.0 - , mu-rpc >=0.2.0 - , mu-schema >=0.2.0 + , mu-grpc-common >=0.3.0 + , mu-protobuf >=0.3.0 + , mu-rpc >=0.3.0 + , mu-schema >=0.3.0 , sop-core , stm , stm-conduit , wai , warp - , warp-grpc + , warp-grpc >=0.4.0.1 , warp-tls hs-source-dirs: src diff --git a/src/ExampleServer.hs b/src/ExampleServer.hs --- a/src/ExampleServer.hs +++ b/src/ExampleServer.hs @@ -1,6 +1,5 @@ {-# language DataKinds #-} {-# language OverloadedStrings #-} -{-# language TypeApplications #-} {-# language TypeFamilies #-} module Main where @@ -10,11 +9,11 @@ import Mu.Schema type instance AnnotatedSchema ProtoBufAnnotation QuickstartSchema - = '[ 'AnnField "HelloRequest" "name" ('ProtoBufId 1) - , 'AnnField "HelloResponse" "message" ('ProtoBufId 1) - , 'AnnField "HiRequest" "number" ('ProtoBufId 1) ] + = '[ 'AnnField "HelloRequest" "name" ('ProtoBufId 1 'True) + , 'AnnField "HelloResponse" "message" ('ProtoBufId 1 'True) + , 'AnnField "HiRequest" "number" ('ProtoBufId 1 'True) ] main :: IO () main = do putStrLn "running quickstart application" - runGRpcApp msgProtoBuf 8080 (quickstartServer @_ @Maybe) + runGRpcApp msgProtoBuf 8080 quickstartServer diff --git a/src/Mu/GRpc/Server.hs b/src/Mu/GRpc/Server.hs --- a/src/Mu/GRpc/Server.hs +++ b/src/Mu/GRpc/Server.hs @@ -28,7 +28,7 @@ , runGRpcAppSettings, Settings , runGRpcAppTLS, TLSSettings -- * Convert a 'Server' into a WAI application -, gRpcApp +, gRpcApp, gRpcAppTrans -- * Raise errors as exceptions in IO , raiseErrors, liftServerConduit -- * Re-export useful instances @@ -36,27 +36,28 @@ ) where import Control.Concurrent.Async -import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TMVar import Control.Exception import Control.Monad.Except -import Data.ByteString (ByteString) -import qualified Data.ByteString.Char8 as BS +import Data.Avro +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as BS import Data.Conduit import Data.Conduit.TMChan import Data.Kind import Data.Proxy -import Network.GRPC.HTTP2.Encoding (GRPCInput, GRPCOutput, gzip, uncompressed) -import Network.GRPC.HTTP2.Types (GRPCStatus (..), GRPCStatusCode (..)) -import Network.GRPC.Server.Handlers -import Network.GRPC.Server.Wai as Wai -import Network.Wai (Application) -import Network.Wai.Handler.Warp (Port, Settings, run, runSettings) -import Network.Wai.Handler.WarpTLS (TLSSettings, runTLS) +import Network.GRPC.HTTP2.Encoding (GRPCInput, GRPCOutput, gzip, uncompressed) +import Network.GRPC.HTTP2.Types (GRPCStatus (..), GRPCStatusCode (..)) +import Network.GRPC.Server.Handlers.Trans +import Network.GRPC.Server.Wai as Wai +import Network.Wai (Application) +import Network.Wai.Handler.Warp (Port, Settings, run, runSettings) +import Network.Wai.Handler.WarpTLS (TLSSettings, runTLS) import Mu.Adapter.ProtoBuf.Via import Mu.GRpc.Avro -import qualified Mu.GRpc.Avro as Avro +import qualified Mu.GRpc.Avro as Avro import Mu.GRpc.Bridge import Mu.Rpc import Mu.Schema @@ -64,22 +65,20 @@ -- | Run a Mu 'Server' on the given port. runGRpcApp - :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers protocol ServerErrorIO methods handlers ) + :: ( KnownName name, GRpcServiceHandlers protocol ServerErrorIO chn services handlers ) => Proxy protocol -> Port - -> ServerT f ('Service name anns methods) ServerErrorIO handlers + -> ServerT chn ('Package ('Just name) services) ServerErrorIO handlers -> IO () runGRpcApp protocol port = runGRpcAppTrans protocol port id -- | Run a Mu 'Server' on the given port. runGRpcAppTrans - :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers protocol m methods handlers ) + :: ( KnownName name, GRpcServiceHandlers protocol m chn services handlers ) => Proxy protocol -> Port -> (forall a. m a -> ServerErrorIO a) - -> ServerT f ('Service name anns methods) m handlers + -> ServerT chn ('Package ('Just name) services) m handlers -> IO () runGRpcAppTrans protocol port f svr = run port (gRpcAppTrans protocol f svr) @@ -87,12 +86,11 @@ -- -- Go to 'Network.Wai.Handler.Warp' to declare 'Settings'. runGRpcAppSettings - :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers protocol m methods handlers ) + :: ( KnownName name, GRpcServiceHandlers protocol m chn services handlers ) => Proxy protocol -> Settings -> (forall a. m a -> ServerErrorIO a) - -> ServerT f ('Service name anns methods) m handlers + -> ServerT chn ('Package ('Just name) services) m handlers -> IO () runGRpcAppSettings protocol st f svr = runSettings st (gRpcAppTrans protocol f svr) @@ -101,12 +99,11 @@ -- Go to 'Network.Wai.Handler.WarpTLS' to declare 'TLSSettings' -- and to 'Network.Wai.Handler.Warp' to declare 'Settings'. runGRpcAppTLS - :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers protocol m methods handlers ) + :: ( KnownName name, GRpcServiceHandlers protocol m chn services handlers ) => Proxy protocol -> TLSSettings -> Settings -> (forall a. m a -> ServerErrorIO a) - -> ServerT f ('Service name anns methods) m handlers + -> ServerT chn ('Package ('Just name) services) m handlers -> IO () runGRpcAppTLS protocol tls st f svr = runTLS tls st (gRpcAppTrans protocol f svr) @@ -116,10 +113,9 @@ -- for example, @wai-routes@, or you can add middleware -- from @wai-extra@, among others. gRpcApp - :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers protocol ServerErrorIO methods handlers ) + :: ( KnownName name, GRpcServiceHandlers protocol ServerErrorIO chn services handlers ) => Proxy protocol - -> ServerT f ('Service name anns methods) ServerErrorIO handlers + -> ServerT chn ('Package ('Just name) services) ServerErrorIO handlers -> Application gRpcApp protocol = gRpcAppTrans protocol id @@ -129,40 +125,58 @@ -- for example, @wai-routes@, or you can add middleware -- from @wai-extra@, among others. gRpcAppTrans - :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers protocol m methods handlers ) + :: ( KnownName name, GRpcServiceHandlers protocol m chn services handlers ) => Proxy protocol -> (forall a. m a -> ServerErrorIO a) - -> ServerT f ('Service name anns methods) m handlers + -> ServerT chn ('Package ('Just name) services) m handlers -> Application gRpcAppTrans protocol f svr = Wai.grpcApp [uncompressed, gzip] - (gRpcServiceHandlers protocol f svr) + (gRpcServerHandlers protocol f svr) -gRpcServiceHandlers - :: forall name anns methods handlers m protocol w. - ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers protocol m methods handlers ) +gRpcServerHandlers + :: forall name services handlers m protocol chn. + ( KnownName name, GRpcServiceHandlers protocol m chn services handlers ) => Proxy protocol -> (forall a. m a -> ServerErrorIO a) - -> ServerT w ('Service name anns methods) m handlers + -> ServerT chn ('Package ('Just name) services) m handlers -> [ServiceHandler] -gRpcServiceHandlers pr f (Server svr) = gRpcMethodHandlers f pr packageName serviceName svr - where packageName = BS.pack (nameVal (Proxy @(FindPackageName anns))) - serviceName = BS.pack (nameVal (Proxy @name)) +gRpcServerHandlers pr f (Services svr) = gRpcServiceHandlers f pr packageName svr + where packageName = BS.pack (nameVal (Proxy @name)) +class GRpcServiceHandlers (p :: GRpcMessageProtocol) (m :: Type -> Type) + (chn :: ServiceChain snm) + (ss :: [Service snm mnm anm]) (hs :: [[Type]]) where + gRpcServiceHandlers :: (forall a. m a -> ServerErrorIO a) + -> Proxy p -> ByteString + -> ServicesT chn ss m hs -> [ServiceHandler] + +instance GRpcServiceHandlers p m chn '[] '[] where + gRpcServiceHandlers _ _ _ S0 = [] +instance ( KnownName name, GRpcMethodHandlers p m chn (MappingRight chn name) methods h + , GRpcServiceHandlers p m chn rest hs ) + => GRpcServiceHandlers p m chn ('Service name anns methods ': rest) (h ': hs) where + gRpcServiceHandlers f pr packageName (svr :<&>: rest) + = gRpcMethodHandlers f pr packageName serviceName svr + ++ gRpcServiceHandlers f pr packageName rest + where serviceName = BS.pack (nameVal (Proxy @name)) + + class GRpcMethodHandlers (p :: GRpcMessageProtocol) (m :: Type -> Type) - (ms :: [Method mnm]) (hs :: [Type]) where + (chn :: ServiceChain snm) (inh :: Type) + (ms :: [Method snm mnm anm]) (hs :: [Type]) where gRpcMethodHandlers :: (forall a. m a -> ServerErrorIO a) -> Proxy p -> ByteString -> ByteString - -> HandlersT f ms m hs -> [ServiceHandler] + -> HandlersT chn inh ms m hs -> [ServiceHandler] -instance GRpcMethodHandlers p m '[] '[] where +instance GRpcMethodHandlers p m chn inh '[] '[] where gRpcMethodHandlers _ _ _ _ H0 = [] -instance (KnownName name, GRpcMethodHandler p m args r h, GRpcMethodHandlers p m rest hs, MkRPC p) - => GRpcMethodHandlers p m ('Method name anns args r ': rest) (h ': hs) where - gRpcMethodHandlers f pr p s (h :<|>: rest) - = gRpcMethodHandler f pr (Proxy @args) (Proxy @r) (mkRPC pr p s methodName) h +instance ( KnownName name, MkRPC p + , GRpcMethodHandler p m args r h + , GRpcMethodHandlers p m chn () rest hs) + => GRpcMethodHandlers p m chn () ('Method name anns args r ': rest) (h ': hs) where + gRpcMethodHandlers f pr p s (h :<||>: rest) + = gRpcMethodHandler f pr (Proxy @args) (Proxy @r) (mkRPC pr p s methodName) (h ()) : gRpcMethodHandlers f pr p s rest where methodName = BS.pack (nameVal (Proxy @name)) @@ -195,7 +209,7 @@ = liftIO $ do h' <- runExceptT h case h' of - Right r -> return r + Right r -> pure r Left (ServerError code msg) -> closeEarly $ GRPCStatus (serverErrorToGRpcError code) (BS.pack msg) @@ -222,7 +236,7 @@ -- the choice of message protocol (PB or Avro) class GRPCOutput (RPCTy p) (GRpcOWTy p ref r) - => GRpcOutputWrapper (p :: GRpcMessageProtocol) (ref :: TypeRef) (r :: Type) where + => GRpcOutputWrapper (p :: GRpcMessageProtocol) (ref :: TypeRef snm) (r :: Type) where type GRpcOWTy p ref r :: Type buildGRpcOWTy :: Proxy p -> Proxy ref -> r -> GRpcOWTy p ref r @@ -231,13 +245,16 @@ type GRpcOWTy 'MsgProtoBuf ref r = ViaToProtoBufTypeRef ref r buildGRpcOWTy _ _ = ViaToProtoBufTypeRef -instance (GRPCOutput AvroRPC (ViaToAvroTypeRef ('ViaSchema sch sty) r)) - => GRpcOutputWrapper 'MsgAvro ('ViaSchema sch sty) r where - type GRpcOWTy 'MsgAvro ('ViaSchema sch sty) r = ViaToAvroTypeRef ('ViaSchema sch sty) r +instance forall (sch :: Schema') sty (r :: Type). + ( ToSchema sch sty r + , ToAvro (WithSchema sch sty r) + , HasAvroSchema (WithSchema sch sty r) ) + => GRpcOutputWrapper 'MsgAvro ('SchemaRef sch sty) r where + type GRpcOWTy 'MsgAvro ('SchemaRef sch sty) r = ViaToAvroTypeRef ('SchemaRef sch sty) r buildGRpcOWTy _ _ = ViaToAvroTypeRef class GRPCInput (RPCTy p) (GRpcIWTy p ref r) - => GRpcInputWrapper (p :: GRpcMessageProtocol) (ref :: TypeRef) (r :: Type) where + => GRpcInputWrapper (p :: GRpcMessageProtocol) (ref :: TypeRef snm) (r :: Type) where type GRpcIWTy p ref r :: Type unGRpcIWTy :: Proxy p -> Proxy ref -> GRpcIWTy p ref r -> r @@ -246,150 +263,181 @@ type GRpcIWTy 'MsgProtoBuf ref r = ViaFromProtoBufTypeRef ref r unGRpcIWTy _ _ = unViaFromProtoBufTypeRef -instance (GRPCInput AvroRPC (ViaFromAvroTypeRef ('ViaSchema sch sty) r)) - => GRpcInputWrapper 'MsgAvro ('ViaSchema sch sty) r where - type GRpcIWTy 'MsgAvro ('ViaSchema sch sty) r = ViaFromAvroTypeRef ('ViaSchema sch sty) r +instance forall (sch :: Schema') sty (r :: Type). + ( FromSchema sch sty r + , FromAvro (WithSchema sch sty r) + , HasAvroSchema (WithSchema sch sty r) ) + => GRpcInputWrapper 'MsgAvro ('SchemaRef sch sty) r where + type GRpcIWTy 'MsgAvro ('SchemaRef sch sty) r = ViaFromAvroTypeRef ('SchemaRef sch sty) r unGRpcIWTy _ _ = unViaFromAvroTypeRef --- -instance (GRPCInput (RPCTy p) (), GRPCOutput (RPCTy p) ()) +instance (MonadIO m, GRPCInput (RPCTy p) (), GRPCOutput (RPCTy p) ()) => GRpcMethodHandler p m '[ ] 'RetNothing (m ()) where gRpcMethodHandler f _ _ _ rpc h - = unary @_ @() @() rpc (\_ _ -> raiseErrors (f h)) + = unary @m @_ @() @() (raiseErrors . f) rpc (\_ _ -> h) ----- -instance (GRPCInput (RPCTy p) (), GRpcOutputWrapper p rref r) +instance (MonadIO m, GRPCInput (RPCTy p) (), GRpcOutputWrapper p rref r) => GRpcMethodHandler p m '[ ] ('RetSingle rref) (m r) where gRpcMethodHandler f _ _ _ rpc h - = unary @_ @() @(GRpcOWTy p rref r) - rpc (\_ _ -> buildGRpcOWTy (Proxy @p) (Proxy @rref) <$> raiseErrors (f h)) + = unary @m @_ @() @(GRpcOWTy p rref r) + (raiseErrors . f) rpc (\_ _ -> buildGRpcOWTy (Proxy @p) (Proxy @rref) <$> h) ----- -instance (GRPCInput (RPCTy p) (), GRpcOutputWrapper p rref r, MonadIO m) +instance (MonadIO m, GRPCInput (RPCTy p) (), GRpcOutputWrapper p rref r, MonadIO m) => GRpcMethodHandler p m '[ ] ('RetStream rref) (ConduitT r Void m () -> m ()) where gRpcMethodHandler f _ _ _ rpc h - = serverStream @_ @() @(GRpcOWTy p rref r) rpc sstream + = serverStream @m @_ @() @(GRpcOWTy p rref r) (raiseErrors . f) rpc sstream where sstream :: req -> () - -> IO ((), ServerStream (GRpcOWTy p rref r) ()) + -> m ((), ServerStream m (GRpcOWTy p rref r) ()) sstream _ _ = do -- Variable to connect input and output - var <- newEmptyTMVarIO :: IO (TMVar (Maybe r)) + var <- liftIO newEmptyTMVarIO :: m (TMVar (Maybe r)) -- Start executing the handler - promise <- async (raiseErrors $ f (h (toTMVarConduit var))) - -- Return the information + promise <- liftIO $ async (raiseErrors $ f (h (toTMVarConduit var))) + -- Return the information let readNext _ - = do nextOutput <- atomically $ takeTMVar var + = do nextOutput <- liftIO $ atomically $ takeTMVar var case nextOutput of - Just o -> return $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) - Nothing -> do cancel promise - return Nothing - return ((), ServerStream readNext) + Just o -> pure $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) + Nothing -> do liftIO $ cancel promise + pure Nothing + pure ((), ServerStream readNext) ----- -instance (GRpcInputWrapper p vref v, GRPCOutput (RPCTy p) ()) - => GRpcMethodHandler p m '[ 'ArgSingle vref ] 'RetNothing (v -> m ()) where +instance (MonadIO m, GRpcInputWrapper p vref v, GRPCOutput (RPCTy p) ()) + => GRpcMethodHandler p m '[ 'ArgSingle aname anns vref ] 'RetNothing (v -> m ()) where gRpcMethodHandler f _ _ _ rpc h - = unary @_ @(GRpcIWTy p vref v) @() - rpc (\_ -> raiseErrors . f . h . unGRpcIWTy (Proxy @p) (Proxy @vref)) + = unary @m @_ @(GRpcIWTy p vref v) @() + (raiseErrors . f) rpc (\_ -> h . unGRpcIWTy (Proxy @p) (Proxy @vref)) ----- -instance (GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r) - => GRpcMethodHandler p m '[ 'ArgSingle vref ] ('RetSingle rref) (v -> m r) where +instance (MonadIO m, GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r) + => GRpcMethodHandler p m '[ 'ArgSingle aname anns vref ] ('RetSingle rref) (v -> m r) where gRpcMethodHandler f _ _ _ rpc h - = unary @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) - rpc (\_ -> (buildGRpcOWTy (Proxy @p) (Proxy @rref) <$>) - . raiseErrors . f . h - . unGRpcIWTy (Proxy @p) (Proxy @vref)) + = unary @m @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) + (raiseErrors . f) rpc + (\_ -> (buildGRpcOWTy (Proxy @p) (Proxy @rref) <$>) + . h + . unGRpcIWTy (Proxy @p) (Proxy @vref)) ----- instance (GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r, MonadIO m) - => GRpcMethodHandler p m '[ 'ArgStream vref ] ('RetSingle rref) - (ConduitT () v m () -> m r) where + => GRpcMethodHandler p m '[ 'ArgSingle aname anns vref ] ('RetStream rref) + (v -> ConduitT r Void m () -> m ()) where gRpcMethodHandler f _ _ _ rpc h - = clientStream @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) - rpc cstream + = serverStream @m @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) + (raiseErrors . f) rpc sstream + where sstream :: req -> GRpcIWTy p vref v + -> m ((), ServerStream m (GRpcOWTy p rref r) ()) + sstream _ v = do + -- Variable to connect input and output + var <- liftIO newEmptyTMVarIO :: m (TMVar (Maybe r)) + -- Start executing the handler + let v' = unGRpcIWTy (Proxy @p) (Proxy @vref) v + promise <- liftIO $ async (raiseErrors $ f (h v' (toTMVarConduit var))) + -- Return the information + let readNext _ + = do nextOutput <- liftIO $ atomically $ takeTMVar var + case nextOutput of + Just o -> pure $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) + Nothing -> do liftIO $ cancel promise + pure Nothing + pure ((), ServerStream readNext) + +----- + +instance (MonadIO m, GRpcInputWrapper p vref v, GRPCOutput (RPCTy p) (), MonadIO m) + => GRpcMethodHandler p m '[ 'ArgStream aname anns vref ] 'RetNothing + (ConduitT () v m () -> m ()) where + gRpcMethodHandler f _ _ _ rpc h + = clientStream @m @_ @(GRpcIWTy p vref v) @() + (raiseErrors . f) rpc cstream where cstream :: req - -> IO ((), ClientStream (GRpcIWTy p vref v) - (GRpcOWTy p rref r) ()) + -> m ((), ClientStream m (GRpcIWTy p vref v) () ()) cstream _ = do -- Create a new TMChan - chan <- newTMChanIO :: IO (TMChan v) + chan <- liftIO newTMChanIO :: m (TMChan v) let producer = sourceTMChan @m chan -- Start executing the handler in another thread - promise <- async (raiseErrors $ buildGRpcOWTy (Proxy @p) (Proxy @rref) <$> f (h producer)) + promise <- liftIO $ async (raiseErrors $ f (h producer)) -- Build the actual handler let cstreamHandler _ newInput - = atomically (writeTMChan chan (unGRpcIWTy (Proxy @p) (Proxy @vref) newInput)) + = liftIO $ atomically $ + writeTMChan chan (unGRpcIWTy (Proxy @p) (Proxy @vref) newInput) cstreamFinalizer _ - = atomically (closeTMChan chan) >> wait promise + = liftIO $ atomically (closeTMChan chan) >> wait promise -- Return the information - return ((), ClientStream cstreamHandler cstreamFinalizer) + pure ((), ClientStream cstreamHandler cstreamFinalizer) ----- -instance (GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r, MonadIO m) - => GRpcMethodHandler p m '[ 'ArgSingle vref ] ('RetStream rref) - (v -> ConduitT r Void m () -> m ()) where +instance (MonadIO m, GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r, MonadIO m) + => GRpcMethodHandler p m '[ 'ArgStream aname anns vref ] ('RetSingle rref) + (ConduitT () v m () -> m r) where gRpcMethodHandler f _ _ _ rpc h - = serverStream @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) - rpc sstream - where sstream :: req -> GRpcIWTy p vref v - -> IO ((), ServerStream (GRpcOWTy p rref r) ()) - sstream _ v = do - -- Variable to connect input and output - var <- newEmptyTMVarIO :: IO (TMVar (Maybe r)) - -- Start executing the handler - let v' = unGRpcIWTy (Proxy @p) (Proxy @vref) v - promise <- async (raiseErrors $ f (h v' (toTMVarConduit var))) - -- Return the information - let readNext _ - = do nextOutput <- atomically $ takeTMVar var - case nextOutput of - Just o -> return $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) - Nothing -> do cancel promise - return Nothing - return ((), ServerStream readNext) + = clientStream @m @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) + (raiseErrors . f) rpc cstream + where cstream :: req + -> m ((), ClientStream m (GRpcIWTy p vref v) + (GRpcOWTy p rref r) ()) + cstream _ = do + -- Create a new TMChan + chan <- liftIO newTMChanIO :: m (TMChan v) + let producer = sourceTMChan @m chan + -- Start executing the handler in another thread + promise <- liftIO $ async (raiseErrors $ buildGRpcOWTy (Proxy @p) (Proxy @rref) <$> f (h producer)) + -- Build the actual handler + let cstreamHandler _ newInput + = liftIO $ atomically $ + writeTMChan chan (unGRpcIWTy (Proxy @p) (Proxy @vref) newInput) + cstreamFinalizer _ + = liftIO $ atomically (closeTMChan chan) >> wait promise + -- Return the information + pure ((), ClientStream cstreamHandler cstreamFinalizer) ----- instance (GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r, MonadIO m) - => GRpcMethodHandler p m '[ 'ArgStream vref ] ('RetStream rref) + => GRpcMethodHandler p m '[ 'ArgStream aname anns vref ] ('RetStream rref) (ConduitT () v m () -> ConduitT r Void m () -> m ()) where gRpcMethodHandler f _ _ _ rpc h - = generalStream @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) - rpc bdstream - where bdstream :: req -> IO ( (), IncomingStream (GRpcIWTy p vref v) () - , (), OutgoingStream (GRpcOWTy p rref r) () ) + = generalStream @m @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) + (raiseErrors . f) rpc bdstream + where bdstream :: req -> m ( (), IncomingStream m (GRpcIWTy p vref v) () + , (), OutgoingStream m (GRpcOWTy p rref r) () ) bdstream _ = do -- Create a new TMChan and a new variable - chan <- newTMChanIO :: IO (TMChan v) + chan <- liftIO newTMChanIO :: m (TMChan v) let producer = sourceTMChan @m chan - var <- newEmptyTMVarIO :: IO (TMVar (Maybe r)) + var <- liftIO newEmptyTMVarIO :: m (TMVar (Maybe r)) -- Start executing the handler - promise <- async (raiseErrors $ f $ h producer (toTMVarConduit var)) + promise <- liftIO $ async (raiseErrors $ f $ h producer (toTMVarConduit var)) -- Build the actual handler let cstreamHandler _ newInput - = atomically (writeTMChan chan (unGRpcIWTy (Proxy @p) (Proxy @vref) newInput)) + = liftIO $ atomically $ + writeTMChan chan (unGRpcIWTy (Proxy @p) (Proxy @vref) newInput) cstreamFinalizer _ - = atomically (closeTMChan chan) >> wait promise + = liftIO $ atomically (closeTMChan chan) >> wait promise readNext _ - = do nextOutput <- atomically $ tryTakeTMVar var + = do nextOutput <- liftIO $ atomically $ tryTakeTMVar var case nextOutput of Just (Just o) -> - return $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) + pure $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) Just Nothing -> do - cancel promise - return Nothing + liftIO $ cancel promise + pure Nothing Nothing -> -- no new elements to output readNext () - return ((), IncomingStream cstreamHandler cstreamFinalizer, (), OutgoingStream readNext) + pure ((), IncomingStream cstreamHandler cstreamFinalizer, (), OutgoingStream readNext) -----