diff --git a/mu-grpc-server.cabal b/mu-grpc-server.cabal --- a/mu-grpc-server.cabal +++ b/mu-grpc-server.cabal @@ -1,67 +1,74 @@ -cabal-version: >=1.10 -name: mu-grpc-server -version: 0.1.0.1 -synopsis: gRPC servers for Mu definitions -description: With @mu-grpc-server@ you can easily build gRPC servers for mu-haskell! -license: Apache-2.0 -license-file: LICENSE -author: Alejandro Serrano, Flavio Corpa -maintainer: alejandro.serrano@47deg.com -copyright: Copyright © 2019-2020 -category: Network -build-type: Simple -extra-source-files: CHANGELOG.md -homepage: https://higherkindness.io/mu-haskell/ -bug-reports: https://github.com/higherkindness/mu-haskell/issues +name: mu-grpc-server +version: 0.2.0.0 +synopsis: gRPC servers for Mu definitions +description: + With @mu-grpc-server@ you can easily build gRPC servers for mu-haskell! +license: Apache-2.0 +license-file: LICENSE +author: Alejandro Serrano, Flavio Corpa +maintainer: alejandro.serrano@47deg.com +copyright: Copyright © 2019-2020 +cabal-version: >=1.10 +category: Network +build-type: Simple +extra-source-files: CHANGELOG.md +homepage: https://higherkindness.io/mu-haskell/ +bug-reports: https://github.com/higherkindness/mu-haskell/issues + source-repository head type: git location: https://github.com/higherkindness/mu-haskell library - exposed-modules: Mu.GRpc.Server - build-depends: base >=4.12 && <5 - , async - , bytestring - , conduit - , http2-grpc-proto3-wire - , http2-grpc-types - , mtl - , mu-protobuf - , mu-rpc - , mu-schema - , sop-core - , stm - , stm-conduit - , wai - , warp - , warp-grpc - , warp-tls - hs-source-dirs: src - default-language: Haskell2010 - ghc-options: -Wall - -fprint-potential-instances + exposed-modules: Mu.GRpc.Server + build-depends: + async + , 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 + , sop-core + , stm + , stm-conduit + , wai + , warp + , warp-grpc + , warp-tls + hs-source-dirs: src + default-language: Haskell2010 + ghc-options: -Wall -fprint-potential-instances + executable grpc-example-server - main-is: ExampleServer.hs - other-modules: Mu.GRpc.Server - build-depends: base >=4.12 && <5 - , async - , bytestring - , conduit - , http2-grpc-proto3-wire - , http2-grpc-types - , mtl - , mu-protobuf - , mu-rpc - , mu-schema - , sop-core - , stm - , stm-conduit - , wai - , warp - , warp-grpc - , warp-tls - hs-source-dirs: src - default-language: Haskell2010 - ghc-options: -Wall + main-is: ExampleServer.hs + other-modules: Mu.GRpc.Server + build-depends: + async + , 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 + , sop-core + , stm + , stm-conduit + , wai + , warp + , warp-grpc + , warp-tls + + hs-source-dirs: src + default-language: Haskell2010 + ghc-options: -Wall diff --git a/src/ExampleServer.hs b/src/ExampleServer.hs --- a/src/ExampleServer.hs +++ b/src/ExampleServer.hs @@ -1,5 +1,6 @@ {-# language DataKinds #-} {-# language OverloadedStrings #-} +{-# language TypeApplications #-} {-# language TypeFamilies #-} module Main where @@ -16,4 +17,4 @@ main :: IO () main = do putStrLn "running quickstart application" - runGRpcApp 8080 quickstartServer + runGRpcApp msgProtoBuf 8080 (quickstartServer @_ @Maybe) 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 @@ -7,6 +7,7 @@ {-# language RankNTypes #-} {-# language ScopedTypeVariables #-} {-# language TypeApplications #-} +{-# language TypeFamilies #-} {-# language TypeOperators #-} {-# language UndecidableInstances #-} {-| @@ -19,37 +20,44 @@ variants provide more control over the settings. -} module Mu.GRpc.Server -( -- * Run a 'Server' directly - runGRpcApp, runGRpcAppTrans +( -- * Supported messaging formats + GRpcMessageProtocol(..) +, msgProtoBuf, msgAvro + -- * Run a 'Server' directly +, runGRpcApp, runGRpcAppTrans , runGRpcAppSettings, Settings , runGRpcAppTLS, TLSSettings -- * Convert a 'Server' into a WAI application , gRpcApp -- * Raise errors as exceptions in IO , raiseErrors, liftServerConduit + -- * Re-export useful instances +, module Avro ) 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.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 (gzip, uncompressed) -import Network.GRPC.HTTP2.Proto3Wire -import Network.GRPC.HTTP2.Types (GRPCStatus (..), GRPCStatusCode (..)) +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.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 Mu.GRpc.Bridge import Mu.Rpc import Mu.Schema import Mu.Server @@ -57,33 +65,36 @@ -- | Run a Mu 'Server' on the given port. runGRpcApp :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers ServerErrorIO methods handlers ) - => Port - -> ServerT Maybe ('Service name anns methods) ServerErrorIO handlers + , GRpcMethodHandlers protocol ServerErrorIO methods handlers ) + => Proxy protocol + -> Port + -> ServerT f ('Service name anns methods) ServerErrorIO handlers -> IO () -runGRpcApp port = runGRpcAppTrans port id +runGRpcApp protocol port = runGRpcAppTrans protocol port id -- | Run a Mu 'Server' on the given port. runGRpcAppTrans :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers m methods handlers ) - => Port + , GRpcMethodHandlers protocol m methods handlers ) + => Proxy protocol + -> Port -> (forall a. m a -> ServerErrorIO a) - -> ServerT Maybe ('Service name anns methods) m handlers + -> ServerT f ('Service name anns methods) m handlers -> IO () -runGRpcAppTrans port f svr = run port (gRpcAppTrans f svr) +runGRpcAppTrans protocol port f svr = run port (gRpcAppTrans protocol f svr) -- | Run a Mu 'Server' using the given 'Settings'. -- -- Go to 'Network.Wai.Handler.Warp' to declare 'Settings'. runGRpcAppSettings :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers m methods handlers ) - => Settings + , GRpcMethodHandlers protocol m methods handlers ) + => Proxy protocol + -> Settings -> (forall a. m a -> ServerErrorIO a) - -> ServerT Maybe ('Service name anns methods) m handlers + -> ServerT f ('Service name anns methods) m handlers -> IO () -runGRpcAppSettings st f svr = runSettings st (gRpcAppTrans f svr) +runGRpcAppSettings protocol st f svr = runSettings st (gRpcAppTrans protocol f svr) -- | Run a Mu 'Server' using the given 'TLSSettings' and 'Settings'. -- @@ -91,12 +102,13 @@ -- and to 'Network.Wai.Handler.Warp' to declare 'Settings'. runGRpcAppTLS :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers m methods handlers ) - => TLSSettings -> Settings + , GRpcMethodHandlers protocol m methods handlers ) + => Proxy protocol + -> TLSSettings -> Settings -> (forall a. m a -> ServerErrorIO a) - -> ServerT Maybe ('Service name anns methods) m handlers + -> ServerT f ('Service name anns methods) m handlers -> IO () -runGRpcAppTLS tls st f svr = runTLS tls st (gRpcAppTrans f svr) +runGRpcAppTLS protocol tls st f svr = runTLS tls st (gRpcAppTrans protocol f svr) -- | Turn a Mu 'Server' into a WAI 'Application'. -- @@ -105,10 +117,11 @@ -- from @wai-extra@, among others. gRpcApp :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers ServerErrorIO methods handlers ) - => ServerT Maybe ('Service name anns methods) ServerErrorIO handlers + , GRpcMethodHandlers protocol ServerErrorIO methods handlers ) + => Proxy protocol + -> ServerT f ('Service name anns methods) ServerErrorIO handlers -> Application -gRpcApp = gRpcAppTrans id +gRpcApp protocol = gRpcAppTrans protocol id -- | Turn a Mu 'Server' into a WAI 'Application'. -- @@ -117,40 +130,46 @@ -- from @wai-extra@, among others. gRpcAppTrans :: ( KnownName name, KnownName (FindPackageName anns) - , GRpcMethodHandlers m methods handlers ) - => (forall a. m a -> ServerErrorIO a) - -> ServerT Maybe ('Service name anns methods) m handlers + , GRpcMethodHandlers protocol m methods handlers ) + => Proxy protocol + -> (forall a. m a -> ServerErrorIO a) + -> ServerT f ('Service name anns methods) m handlers -> Application -gRpcAppTrans f svr = Wai.grpcApp [uncompressed, gzip] - (gRpcServiceHandlers f svr) +gRpcAppTrans protocol f svr + = Wai.grpcApp [uncompressed, gzip] + (gRpcServiceHandlers protocol f svr) gRpcServiceHandlers - :: forall name anns methods handlers m. - (KnownName name, KnownName (FindPackageName anns), GRpcMethodHandlers m methods handlers) - => (forall a. m a -> ServerErrorIO a) - -> ServerT Maybe ('Service name anns methods) m handlers + :: forall name anns methods handlers m protocol w. + ( KnownName name, KnownName (FindPackageName anns) + , GRpcMethodHandlers protocol m methods handlers ) + => Proxy protocol + -> (forall a. m a -> ServerErrorIO a) + -> ServerT w ('Service name anns methods) m handlers -> [ServiceHandler] -gRpcServiceHandlers f (Server svr) = gRpcMethodHandlers f packageName serviceName svr +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)) -class GRpcMethodHandlers (m :: Type -> Type) (ms :: [Method mnm]) (hs :: [Type]) where +class GRpcMethodHandlers (p :: GRpcMessageProtocol) (m :: Type -> Type) + (ms :: [Method mnm]) (hs :: [Type]) where gRpcMethodHandlers :: (forall a. m a -> ServerErrorIO a) - -> ByteString -> ByteString - -> HandlersT Maybe ms m hs -> [ServiceHandler] + -> Proxy p -> ByteString -> ByteString + -> HandlersT f ms m hs -> [ServiceHandler] -instance GRpcMethodHandlers m '[] '[] where - gRpcMethodHandlers _ _ _ H0 = [] -instance (KnownName name, GRpcMethodHandler m args r h, GRpcMethodHandlers m rest hs) - => GRpcMethodHandlers m ('Method name anns args r ': rest) (h ': hs) where - gRpcMethodHandlers f p s (h :<|>: rest) - = gRpcMethodHandler f (Proxy @args) (Proxy @r) (RPC p s methodName) h - : gRpcMethodHandlers f p s rest +instance GRpcMethodHandlers p m '[] '[] 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 + : gRpcMethodHandlers f pr p s rest where methodName = BS.pack (nameVal (Proxy @name)) -class GRpcMethodHandler m args r h where +class GRpcMethodHandler p m args r h where gRpcMethodHandler :: (forall a. m a -> ServerErrorIO a) - -> Proxy args -> Proxy r -> RPC -> h -> ServiceHandler + -> Proxy p -> Proxy args -> Proxy r + -> RPCTy p -> h -> ServiceHandler -- | Turns a 'Conduit' working on 'ServerErrorIO' -- into any other base monad which supports 'IO', @@ -195,102 +214,159 @@ serverErrorToGRpcError NotFound = NOT_FOUND serverErrorToGRpcError Invalid = INVALID_ARGUMENT -instance GRpcMethodHandler m '[ ] 'RetNothing (m ()) where - gRpcMethodHandler f _ _ rpc h +----- +-- IMPLEMENTATION OF THE METHODS +----- + +-- These type classes allow us to abstract over +-- the choice of message protocol (PB or Avro) + +class GRPCOutput (RPCTy p) (GRpcOWTy p ref r) + => GRpcOutputWrapper (p :: GRpcMessageProtocol) (ref :: TypeRef) (r :: Type) where + type GRpcOWTy p ref r :: Type + buildGRpcOWTy :: Proxy p -> Proxy ref -> r -> GRpcOWTy p ref r + +instance ToProtoBufTypeRef ref r + => GRpcOutputWrapper 'MsgProtoBuf ref r where + 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 + buildGRpcOWTy _ _ = ViaToAvroTypeRef + +class GRPCInput (RPCTy p) (GRpcIWTy p ref r) + => GRpcInputWrapper (p :: GRpcMessageProtocol) (ref :: TypeRef) (r :: Type) where + type GRpcIWTy p ref r :: Type + unGRpcIWTy :: Proxy p -> Proxy ref -> GRpcIWTy p ref r -> r + +instance FromProtoBufTypeRef ref r + => GRpcInputWrapper 'MsgProtoBuf ref r where + 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 + unGRpcIWTy _ _ = unViaFromAvroTypeRef + +--- + +instance (GRPCInput (RPCTy p) (), GRPCOutput (RPCTy p) ()) + => GRpcMethodHandler p m '[ ] 'RetNothing (m ()) where + gRpcMethodHandler f _ _ _ rpc h = unary @_ @() @() rpc (\_ _ -> raiseErrors (f h)) -instance (ToProtoBufTypeRef rref r) - => GRpcMethodHandler m '[ ] ('RetSingle rref) (m r) where - gRpcMethodHandler f _ _ rpc h - = unary @_ @() @(ViaToProtoBufTypeRef rref r) - rpc (\_ _ -> ViaToProtoBufTypeRef <$> raiseErrors (f h)) +----- -instance (ToProtoBufTypeRef rref r, MonadIO m) - => GRpcMethodHandler m '[ ] ('RetStream rref) +instance (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)) + +----- + +instance (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 @_ @() @(ViaToProtoBufTypeRef rref r) rpc sstream + gRpcMethodHandler f _ _ _ rpc h + = serverStream @_ @() @(GRpcOWTy p rref r) rpc sstream where sstream :: req -> () - -> IO ((), ServerStream (ViaToProtoBufTypeRef rref r) ()) + -> IO ((), ServerStream (GRpcOWTy p rref r) ()) sstream _ _ = do -- Variable to connect input and output var <- newEmptyTMVarIO :: IO (TMVar (Maybe r)) -- Start executing the handler - promise <- async (raiseErrors $ ViaToProtoBufTypeRef <$> f (h (toTMVarConduit var))) + promise <- async (raiseErrors $ f (h (toTMVarConduit var))) -- Return the information let readNext _ = do nextOutput <- atomically $ takeTMVar var case nextOutput of - Just o -> return $ Just ((), ViaToProtoBufTypeRef o) + Just o -> return $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) Nothing -> do cancel promise return Nothing return ((), ServerStream readNext) -instance (FromProtoBufTypeRef vref v) - => GRpcMethodHandler m '[ 'ArgSingle vref ] 'RetNothing (v -> m ()) where - gRpcMethodHandler f _ _ rpc h - = unary @_ @(ViaFromProtoBufTypeRef vref v) @() - rpc (\_ -> raiseErrors . f . h . unViaFromProtoBufTypeRef) +----- -instance (FromProtoBufTypeRef vref v, ToProtoBufTypeRef rref r) - => GRpcMethodHandler m '[ 'ArgSingle vref ] ('RetSingle rref) (v -> m r) where - gRpcMethodHandler f _ _ rpc h - = unary @_ @(ViaFromProtoBufTypeRef vref v) @(ViaToProtoBufTypeRef rref r) - rpc (\_ -> (ViaToProtoBufTypeRef <$>) . raiseErrors . f . h . unViaFromProtoBufTypeRef) +instance (GRpcInputWrapper p vref v, GRPCOutput (RPCTy p) ()) + => GRpcMethodHandler p m '[ 'ArgSingle vref ] 'RetNothing (v -> m ()) where + gRpcMethodHandler f _ _ _ rpc h + = unary @_ @(GRpcIWTy p vref v) @() + rpc (\_ -> raiseErrors . f . h . unGRpcIWTy (Proxy @p) (Proxy @vref)) -instance (FromProtoBufTypeRef vref v, ToProtoBufTypeRef rref r, MonadIO m) - => GRpcMethodHandler m '[ 'ArgStream vref ] ('RetSingle rref) +----- + +instance (GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r) + => GRpcMethodHandler p m '[ 'ArgSingle 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)) + +----- + +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 f _ _ rpc h - = clientStream @_ @(ViaFromProtoBufTypeRef vref v) @(ViaToProtoBufTypeRef rref r) + gRpcMethodHandler f _ _ _ rpc h + = clientStream @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) rpc cstream where cstream :: req - -> IO ((), ClientStream (ViaFromProtoBufTypeRef vref v) - (ViaToProtoBufTypeRef rref r) ()) + -> IO ((), ClientStream (GRpcIWTy p vref v) + (GRpcOWTy p rref r) ()) cstream _ = do -- Create a new TMChan chan <- newTMChanIO :: IO (TMChan v) let producer = sourceTMChan @m chan -- Start executing the handler in another thread - promise <- async (raiseErrors $ ViaToProtoBufTypeRef <$> f (h producer)) + promise <- async (raiseErrors $ buildGRpcOWTy (Proxy @p) (Proxy @rref) <$> f (h producer)) -- Build the actual handler - let cstreamHandler _ (ViaFromProtoBufTypeRef newInput) - = atomically (writeTMChan chan newInput) + let cstreamHandler _ newInput + = atomically (writeTMChan chan (unGRpcIWTy (Proxy @p) (Proxy @vref) newInput)) cstreamFinalizer _ = atomically (closeTMChan chan) >> wait promise -- Return the information return ((), ClientStream cstreamHandler cstreamFinalizer) -instance (FromProtoBufTypeRef vref v, ToProtoBufTypeRef rref r, MonadIO m) - => GRpcMethodHandler m '[ 'ArgSingle vref ] ('RetStream rref) +----- + +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 - gRpcMethodHandler f _ _ rpc h - = serverStream @_ @(ViaFromProtoBufTypeRef vref v) @(ViaToProtoBufTypeRef rref r) + gRpcMethodHandler f _ _ _ rpc h + = serverStream @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) rpc sstream - where sstream :: req -> ViaFromProtoBufTypeRef vref v - -> IO ((), ServerStream (ViaToProtoBufTypeRef rref r) ()) - sstream _ (ViaFromProtoBufTypeRef v) = do + 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 - promise <- async (raiseErrors $ ViaToProtoBufTypeRef <$> f (h v (toTMVarConduit var))) + 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 ((), ViaToProtoBufTypeRef o) + Just o -> return $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) Nothing -> do cancel promise return Nothing return ((), ServerStream readNext) -instance (FromProtoBufTypeRef vref v, ToProtoBufTypeRef rref r, MonadIO m) - => GRpcMethodHandler m '[ 'ArgStream vref ] ('RetStream rref) +----- + +instance (GRpcInputWrapper p vref v, GRpcOutputWrapper p rref r, MonadIO m) + => GRpcMethodHandler p m '[ 'ArgStream vref ] ('RetStream rref) (ConduitT () v m () -> ConduitT r Void m () -> m ()) where - gRpcMethodHandler f _ _ rpc h - = generalStream @_ @(ViaFromProtoBufTypeRef vref v) @(ViaToProtoBufTypeRef rref r) + gRpcMethodHandler f _ _ _ rpc h + = generalStream @_ @(GRpcIWTy p vref v) @(GRpcOWTy p rref r) rpc bdstream - where bdstream :: req -> IO ( (), IncomingStream (ViaFromProtoBufTypeRef vref v) () - , (), OutgoingStream (ViaToProtoBufTypeRef rref r) () ) + where bdstream :: req -> IO ( (), IncomingStream (GRpcIWTy p vref v) () + , (), OutgoingStream (GRpcOWTy p rref r) () ) bdstream _ = do -- Create a new TMChan and a new variable chan <- newTMChanIO :: IO (TMChan v) @@ -299,21 +375,23 @@ -- Start executing the handler promise <- async (raiseErrors $ f $ h producer (toTMVarConduit var)) -- Build the actual handler - let cstreamHandler _ (ViaFromProtoBufTypeRef newInput) - = atomically (writeTMChan chan newInput) + let cstreamHandler _ newInput + = atomically (writeTMChan chan (unGRpcIWTy (Proxy @p) (Proxy @vref) newInput)) cstreamFinalizer _ = atomically (closeTMChan chan) >> wait promise readNext _ = do nextOutput <- atomically $ tryTakeTMVar var case nextOutput of Just (Just o) -> - return $ Just ((), ViaToProtoBufTypeRef o) + return $ Just ((), buildGRpcOWTy (Proxy @p) (Proxy @rref) o) Just Nothing -> do cancel promise return Nothing Nothing -> -- no new elements to output readNext () return ((), IncomingStream cstreamHandler cstreamFinalizer, (), OutgoingStream readNext) + +----- toTMVarConduit :: MonadIO m => TMVar (Maybe r) -> ConduitT r Void m () toTMVarConduit var = do