mu-grpc-server 0.2.0.0 → 0.3.0.0

raw patch · 3 files changed

+195/−146 lines, 3 filesdep +avrodep ~mu-grpc-commondep ~mu-protobufdep ~mu-rpcPVP ok

version bump matches the API change (PVP)

Dependencies added: avro

Dependency ranges changed: mu-grpc-common, mu-protobuf, mu-rpc, mu-schema, warp-grpc

API changes (from Hackage documentation)

Files

• mu-grpc-server.cabal +15/−13
• src/ExampleServer.hs +4/−5
• src/Mu/GRpc/Server.hs +176/−128

@@ -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

@@ -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

@@ -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)  -----