diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Revision history for mu-haskell + +## 0.1.0.0 -- YYYY-mm-dd + +* First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright © 2019-2020 47 Degrees. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/mu-grpc-client.cabal b/mu-grpc-client.cabal new file mode 100644 --- /dev/null +++ b/mu-grpc-client.cabal @@ -0,0 +1,47 @@ +cabal-version: >=1.10 +name: mu-grpc-client +version: 0.1.0.0 +synopsis: gRPC clients from Mu definitions +description: With @mu-grpc-client@ you can easily build gRPC clients 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 + +source-repository head + type: git + location: https://github.com/higherkindness/mu-haskell + +library + exposed-modules: Mu.GRpc.Client.TyApps, + Mu.GRpc.Client.Record, + Mu.GRpc.Client.Examples + other-modules: Mu.GRpc.Client.Internal + build-depends: base >=4.12 && <5 + , async + , bytestring + , conduit + , http2 + , http2-client + , http2-client-grpc + , http2-grpc-proto3-wire + , mu-protobuf + , mu-rpc + , mu-schema + , sop-core + , stm + , stm-chans + , stm-conduit + , template-haskell >= 2.12 + , text + , th-abstraction + hs-source-dirs: src + default-language: Haskell2010 + ghc-options: -Wall + -fprint-potential-instances diff --git a/src/Mu/GRpc/Client/Examples.hs b/src/Mu/GRpc/Client/Examples.hs new file mode 100644 --- /dev/null +++ b/src/Mu/GRpc/Client/Examples.hs @@ -0,0 +1,44 @@ +{-# language DataKinds #-} +{-# language TypeApplications #-} +{-# language TypeFamilies #-} +{-| +Description : Examples for gRPC clients + +Look at the source code of this module. +-} +module Mu.GRpc.Client.Examples where + +import Data.Conduit +import Data.Conduit.Combinators as C +import Data.Conduit.List (consume) +import qualified Data.Text as T +import Network.HTTP2.Client (HostName, PortNumber) + +import Mu.Adapter.ProtoBuf +import Mu.GRpc.Client.TyApps +import Mu.Rpc.Examples +import Mu.Schema + +type instance AnnotatedSchema ProtoBufAnnotation QuickstartSchema + = '[ 'AnnField "HelloRequest" "name" ('ProtoBufId 1) + , 'AnnField "HelloResponse" "message" ('ProtoBufId 1) + , 'AnnField "HiRequest" "number" ('ProtoBufId 1) ] + +type M a = a Maybe + +sayHello' :: HostName -> PortNumber -> T.Text -> IO (GRpcReply (Maybe T.Text)) +sayHello' host port req + = do Right c <- setupGrpcClient' (grpcClientConfigSimple host port False) + fmap (\(HelloResponse r) -> r) <$> sayHello c (HelloRequest (Just req)) + +sayHello :: GrpcClient -> M HelloRequest -> IO (GRpcReply (M HelloResponse)) +sayHello = gRpcCall @QuickStartService @"SayHello" + +sayHi' :: HostName -> PortNumber -> Int -> IO [GRpcReply (Maybe T.Text)] +sayHi' host port n + = do Right c <- setupGrpcClient' (grpcClientConfigSimple host port False) + cndt <- sayHi c (HiRequest (Just n)) + runConduit $ cndt .| C.map (fmap (\(HelloResponse r) -> r)) .| consume + +sayHi :: GrpcClient -> M HiRequest -> IO (ConduitT () (GRpcReply (M HelloResponse)) IO ()) +sayHi = gRpcCall @QuickStartService @"SayHi" diff --git a/src/Mu/GRpc/Client/Internal.hs b/src/Mu/GRpc/Client/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Mu/GRpc/Client/Internal.hs @@ -0,0 +1,255 @@ +{-# language AllowAmbiguousTypes #-} +{-# language DataKinds #-} +{-# language DeriveFunctor #-} +{-# language FlexibleContexts #-} +{-# language FlexibleInstances #-} +{-# language GADTs #-} +{-# language MultiParamTypeClasses #-} +{-# language PolyKinds #-} +{-# language ScopedTypeVariables #-} +{-# language TypeApplications #-} +{-# language TypeOperators #-} +{-# language UndecidableInstances #-} +-- | Client for gRPC services defined using Mu 'Service' +module Mu.GRpc.Client.Internal where + +import Control.Concurrent.Async +import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM.TMChan +import Control.Concurrent.STM.TMVar +import Control.Monad.IO.Class +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as BS +import Data.Conduit +import qualified Data.Conduit.Combinators as C +import Data.Conduit.TMChan +import Network.GRPC.Client (CompressMode (..), IncomingEvent (..), + OutgoingEvent (..), RawReply, StreamDone (..)) +import Network.GRPC.Client.Helpers +import Network.GRPC.HTTP2.Proto3Wire +import Network.HTTP2 (ErrorCode) +import Network.HTTP2.Client (ClientError, ClientIO, TooMuchConcurrency, + runExceptT) + +import Mu.Adapter.ProtoBuf.Via +import Mu.Rpc +import Mu.Schema + +setupGrpcClient' :: GrpcClientConfig -> IO (Either ClientError GrpcClient) +setupGrpcClient' = runExceptT . setupGrpcClient + +class GRpcServiceMethodCall (s :: Service snm mnm) (m :: Method mnm) h where + gRpcServiceMethodCall :: Proxy s -> Proxy m -> GrpcClient -> h +instance (KnownName serviceName, KnownName (FindPackageName anns), GRpcMethodCall m h) + => GRpcServiceMethodCall ('Service serviceName anns methods) m h where + gRpcServiceMethodCall _ = gRpcMethodCall pkgName svrName + where pkgName = BS.pack (nameVal (Proxy @(FindPackageName anns))) + svrName = BS.pack (nameVal (Proxy @serviceName)) + +data GRpcReply a + = GRpcTooMuchConcurrency TooMuchConcurrency + | GRpcErrorCode ErrorCode + | GRpcErrorString String + | GRpcClientError ClientError + | GRpcOk a + deriving (Show, Functor) + +buildGRpcReply1 :: Either TooMuchConcurrency (RawReply a) -> GRpcReply a +buildGRpcReply1 (Left tmc) = GRpcTooMuchConcurrency tmc +buildGRpcReply1 (Right (Left ec)) = GRpcErrorCode ec +buildGRpcReply1 (Right (Right (_, _, Left es))) = GRpcErrorString es +buildGRpcReply1 (Right (Right (_, _, Right r))) = GRpcOk r + +buildGRpcReply2 :: Either TooMuchConcurrency (r, RawReply a) -> GRpcReply a +buildGRpcReply2 (Left tmc) = GRpcTooMuchConcurrency tmc +buildGRpcReply2 (Right (_, Left ec)) = GRpcErrorCode ec +buildGRpcReply2 (Right (_, Right (_, _, Left es))) = GRpcErrorString es +buildGRpcReply2 (Right (_, Right (_, _, Right r))) = GRpcOk r + +buildGRpcReply3 :: Either TooMuchConcurrency v -> GRpcReply () +buildGRpcReply3 (Left tmc) = GRpcTooMuchConcurrency tmc +buildGRpcReply3 (Right _) = GRpcOk () + +simplifyResponse :: ClientIO (GRpcReply a) -> IO (GRpcReply a) +simplifyResponse reply = do + r <- runExceptT reply + case r of + Left e -> return $ GRpcClientError e + Right v -> return v + +class GRpcMethodCall method h where + gRpcMethodCall :: ByteString -> ByteString -> Proxy method -> GrpcClient -> h + +instance (KnownName name, handler ~ IO (GRpcReply ())) + => GRpcMethodCall ('Method name anns '[ ] 'RetNothing) handler where + gRpcMethodCall pkgName srvName _ client + = simplifyResponse $ + buildGRpcReply1 <$> + rawUnary rpc client () + where methodName = BS.pack (nameVal (Proxy @name)) + rpc = RPC pkgName srvName methodName + +instance ( KnownName name, FromProtoBufTypeRef rref r + , handler ~ IO (GRpcReply r) ) + => GRpcMethodCall ('Method name anns '[ ] ('RetSingle rref)) handler where + gRpcMethodCall pkgName srvName _ client + = fmap (fmap unViaFromProtoBufTypeRef) $ + simplifyResponse $ + buildGRpcReply1 <$> + rawUnary @_ @_ @(ViaFromProtoBufTypeRef rref _) rpc client () + where methodName = BS.pack (nameVal (Proxy @name)) + rpc = RPC pkgName srvName methodName + +instance ( KnownName name, FromProtoBufTypeRef rref r + , handler ~ (IO (ConduitT () (GRpcReply r) IO ())) ) + => GRpcMethodCall ('Method name anns '[ ] ('RetStream rref)) handler where + gRpcMethodCall pkgName srvName _ client + = do -- Create a new TMChan + chan <- newTMChanIO :: IO (TMChan r) + var <- newEmptyTMVarIO -- if full, this means an error + -- Start executing the client in another thread + _ <- async $ do + v <- simplifyResponse $ + buildGRpcReply3 <$> + rawStreamServer @_ @() @(ViaFromProtoBufTypeRef rref r) + rpc client () () + (\_ _ (ViaFromProtoBufTypeRef newVal) -> liftIO $ atomically $ + -- on the first iteration, say that everything is OK + tryPutTMVar var (GRpcOk ()) >> writeTMChan chan newVal) + case v of + GRpcOk () -> liftIO $ atomically $ closeTMChan chan + _ -> liftIO $ atomically $ putTMVar var v + -- This conduit feeds information to the other thread + let go = do firstResult <- liftIO $ atomically $ takeTMVar var + case firstResult of + GRpcOk _ -> -- no error, everything is fine + sourceTMChan chan .| C.map GRpcOk + e -> yield $ (\_ -> error "this should never happen") <$> e + return go + where methodName = BS.pack (nameVal (Proxy @name)) + rpc = RPC pkgName srvName methodName + +instance ( KnownName name, ToProtoBufTypeRef vref v + , handler ~ (v -> IO (GRpcReply ())) ) + => GRpcMethodCall ('Method name anns '[ 'ArgSingle vref ] 'RetNothing) handler where + gRpcMethodCall pkgName srvName _ client x + = simplifyResponse $ + buildGRpcReply1 <$> + rawUnary @_ @(ViaToProtoBufTypeRef vref _) rpc client (ViaToProtoBufTypeRef x) + where methodName = BS.pack (nameVal (Proxy @name)) + rpc = RPC pkgName srvName methodName + +instance ( KnownName name, ToProtoBufTypeRef vref v, FromProtoBufTypeRef rref r + , handler ~ (v -> IO (GRpcReply r)) ) + => GRpcMethodCall ('Method name anns '[ 'ArgSingle vref ] ('RetSingle rref)) handler where + gRpcMethodCall pkgName srvName _ client x + = fmap (fmap unViaFromProtoBufTypeRef) $ + simplifyResponse $ + buildGRpcReply1 <$> + rawUnary @_ @(ViaToProtoBufTypeRef vref _) @(ViaFromProtoBufTypeRef rref _) + rpc client (ViaToProtoBufTypeRef x) + where methodName = BS.pack (nameVal (Proxy @name)) + rpc = RPC pkgName srvName methodName + +instance ( KnownName name, ToProtoBufTypeRef vref v, FromProtoBufTypeRef rref r + , handler ~ (CompressMode -> IO (ConduitT v Void IO (GRpcReply r))) ) + => GRpcMethodCall ('Method name anns '[ 'ArgStream vref ] ('RetSingle rref)) handler where + gRpcMethodCall pkgName srvName _ client compress + = do -- Create a new TMChan + chan <- newTMChanIO :: IO (TMChan v) + -- Start executing the client in another thread + promise <- async $ + fmap (fmap unViaFromProtoBufTypeRef) $ + simplifyResponse $ + buildGRpcReply2 <$> + rawStreamClient @_ @(ViaToProtoBufTypeRef vref v) @(ViaFromProtoBufTypeRef rref r) rpc client () + (\_ -> do nextVal <- liftIO $ atomically $ readTMChan chan + case nextVal of + Nothing -> return ((), Left StreamDone) + Just v -> return ((), Right (compress, ViaToProtoBufTypeRef v))) + -- This conduit feeds information to the other thread + let go = do x <- await + case x of + Just v -> do liftIO $ atomically $ writeTMChan chan v + go + Nothing -> do liftIO $ atomically $ closeTMChan chan + liftIO $ wait promise + return go + where methodName = BS.pack (nameVal (Proxy @name)) + rpc = RPC pkgName srvName methodName + +instance ( KnownName name, ToProtoBufTypeRef vref v, FromProtoBufTypeRef rref r + , handler ~ (v -> IO (ConduitT () (GRpcReply r) IO ())) ) + => GRpcMethodCall ('Method name anns '[ 'ArgSingle vref ] ('RetStream rref)) handler where + gRpcMethodCall pkgName srvName _ client x + = do -- Create a new TMChan + chan <- newTMChanIO :: IO (TMChan r) + var <- newEmptyTMVarIO -- if full, this means an error + -- Start executing the client in another thread + _ <- async $ do + v <- simplifyResponse $ + buildGRpcReply3 <$> + rawStreamServer @_ @(ViaToProtoBufTypeRef vref v) @(ViaFromProtoBufTypeRef rref r) + rpc client () (ViaToProtoBufTypeRef x) + (\_ _ (ViaFromProtoBufTypeRef newVal) -> liftIO $ atomically $ + -- on the first iteration, say that everything is OK + tryPutTMVar var (GRpcOk ()) >> writeTMChan chan newVal) + case v of + GRpcOk () -> liftIO $ atomically $ closeTMChan chan + _ -> liftIO $ atomically $ putTMVar var v + -- This conduit feeds information to the other thread + let go = do firstResult <- liftIO $ atomically $ takeTMVar var + case firstResult of + GRpcOk _ -> -- no error, everything is fine + sourceTMChan chan .| C.map GRpcOk + e -> yield $ (\_ -> error "this should never happen") <$> e + return go + where methodName = BS.pack (nameVal (Proxy @name)) + rpc = RPC pkgName srvName methodName + +instance ( KnownName name, ToProtoBufTypeRef vref v, FromProtoBufTypeRef rref r + , handler ~ (CompressMode -> IO (ConduitT v (GRpcReply r) IO ())) ) + => GRpcMethodCall ('Method name anns '[ 'ArgStream vref ] ('RetStream rref)) handler where + gRpcMethodCall pkgName srvName _ 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 + -- Start executing the client in another thread + _ <- async $ do + v <- simplifyResponse $ + buildGRpcReply3 <$> + rawGeneralStream + @_ @(ViaToProtoBufTypeRef vref v) @(ViaFromProtoBufTypeRef 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 $ unViaFromProtoBufTypeRef o) + Invalid e -> liftIO $ atomically $ writeTMChan inchan (GRpcErrorString (show e)) + _ -> return () ) + () (\_ -> do + nextVal <- liftIO $ atomically $ readTMChan outchan + case nextVal of + Nothing -> return ((), Finalize) + Just v -> return ((), SendMessage compress (ViaToProtoBufTypeRef 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 -> return () -- both are empty, end + Just Nothing -> go2 + Just (Just nextIn) -> yield nextIn >> go2 + return go + where methodName = BS.pack (nameVal (Proxy @name)) + rpc = RPC pkgName srvName methodName diff --git a/src/Mu/GRpc/Client/Record.hs b/src/Mu/GRpc/Client/Record.hs new file mode 100644 --- /dev/null +++ b/src/Mu/GRpc/Client/Record.hs @@ -0,0 +1,246 @@ +{-# language AllowAmbiguousTypes #-} +{-# language DataKinds #-} +{-# language FlexibleContexts #-} +{-# language FlexibleInstances #-} +{-# language MultiParamTypeClasses #-} +{-# language PolyKinds #-} +{-# language ScopedTypeVariables #-} +{-# language TemplateHaskell #-} +{-# language TypeApplications #-} +{-# language TypeFamilies #-} +{-# language TypeOperators #-} +{-# language UndecidableInstances #-} +{-| +Description : Client for gRPC services using plain Haskell records + +For further information over initialization of the connection, +consult the . +-} +module Mu.GRpc.Client.Record ( + -- * Initialization of the gRPC client + GrpcClient +, GrpcClientConfig +, grpcClientConfigSimple +, setupGrpcClient' + -- * Fill and generate the Haskell record of functions +, buildService +, CompressMode(..) +, GRpcReply(..) +, generateRecordFromService +) where + +import Control.Applicative +import Data.Char +import Data.Conduit (ConduitT) +import Data.Proxy +import Data.Void +import GHC.Generics hiding (NoSourceStrictness, NoSourceUnpackedness) +import GHC.TypeLits +import Language.Haskell.TH hiding (ppr) +import Language.Haskell.TH.Datatype + +import Network.GRPC.Client (CompressMode (..)) +import Network.GRPC.Client.Helpers + +import Mu.GRpc.Client.Internal +import Mu.Rpc + +-- | Fills in a Haskell record of functions with the corresponding +-- calls to gRPC services from a Mu 'Service' declaration. +buildService :: forall (s :: Service') (p :: Symbol) t + (nm :: Symbol) (anns :: [ServiceAnnotation]) (ms :: [Method Symbol]). + (s ~ 'Service nm anns ms, Generic t, BuildService s p ms (Rep t)) + => GrpcClient -> t +buildService client = to (buildService' (Proxy @s) (Proxy @p) (Proxy @ms) client) + +class BuildService (s :: Service') (p :: Symbol) (ms :: [Method Symbol]) (f :: * -> *) where + buildService' :: Proxy s -> Proxy p -> Proxy ms -> GrpcClient -> f a + +instance BuildService s p ms U1 where + buildService' _ _ _ _ = U1 +instance BuildService s p ms f => BuildService s p ms (D1 meta f) where + buildService' ps ppr pms client + = M1 (buildService' ps ppr pms client) +instance BuildService s p ms f => BuildService s p ms (C1 meta f) where + buildService' ps ppr pms client + = M1 (buildService' ps ppr pms client) +instance TypeError ('Text "building a service from sums is not supported") + => BuildService s p ms (f :+: g) where + buildService' = error "this should never happen" +instance (BuildService s p ms f, BuildService s p ms g) + => BuildService s p ms (f :*: g) where + buildService' ps ppr pms client + = buildService' ps ppr pms client :*: buildService' ps ppr pms client +instance (m ~ AppendSymbol p x, GRpcServiceMethodCall s (s :-->: x) h) + => BuildService s p ms (S1 ('MetaSel ('Just m) u ss ds) (K1 i h)) where + buildService' ps _ _ client + = M1 $ K1 $ gRpcServiceMethodCall ps (Proxy @(s :-->: x)) client + +-- TEMPLATE HASKELL +-- ================ + +-- | Generate the plain Haskell record corresponding to +-- a Mu 'Service' definition, and a concrete implementation +-- of 'buildService' for that record. +generateRecordFromService :: String -> String -> Namer -> Name -> Q [Dec] +generateRecordFromService newRecordName fieldsPrefix tNamer serviceTyName + = do let serviceTy = ConT serviceTyName + srvDef <- typeToServiceDef serviceTy + case srvDef of + Nothing -> fail "service definition cannot be parsed" + Just sd -> serviceDefToDecl serviceTyName newRecordName fieldsPrefix tNamer sd + +type Namer = String -> String + +serviceDefToDecl :: Name -> String -> String -> Namer -> Service String String -> Q [Dec] +serviceDefToDecl serviceTyName complete fieldsPrefix tNamer (Service _ _ methods) + = do d <- dataD (pure []) + (mkName complete) + [] + Nothing + [RecC (mkName complete) <$> mapM (methodToDecl fieldsPrefix tNamer) methods] + [pure (DerivClause Nothing [ConT ''Generic])] + let buildName = mkName ("build" ++ complete) + s <- SigD buildName <$> [t|GrpcClient -> $(return (ConT (mkName complete)))|] + c <- Clause <$> pure [] + <*> (NormalB <$> [e|buildService @($(return $ ConT serviceTyName)) + @($(return $ LitT (StrTyLit fieldsPrefix)))|]) + <*> pure [] + return [d, s, FunD buildName [c]] + +methodToDecl :: String -> Namer -> Method String -> Q (Name, Bang, Type) +methodToDecl fieldsPrefix tNamer (Method mName _ args ret) + = do let nm = firstLower (fieldsPrefix ++ mName) + ty <- computeMethodType tNamer args ret + return ( mkName nm, Bang NoSourceUnpackedness NoSourceStrictness, ty ) + +computeMethodType :: Namer -> [Argument] -> Return -> Q Type +computeMethodType _ [] RetNothing + = [t|IO (GRpcReply ())|] +computeMethodType n [] (RetSingle r) + = [t|IO (GRpcReply $(typeRefToType n r))|] +computeMethodType n [ArgSingle v] RetNothing + = [t|$(typeRefToType n v) -> IO (GRpcReply ())|] +computeMethodType n [ArgSingle v] (RetSingle r) + = [t|$(typeRefToType n v) -> IO (GRpcReply $(typeRefToType n r))|] +computeMethodType n [ArgStream v] (RetSingle r) + = [t|CompressMode -> IO (ConduitT $(typeRefToType n v) Void IO (GRpcReply $(typeRefToType n r)))|] +computeMethodType n [ArgSingle v] (RetStream r) + = [t|$(typeRefToType n v) -> IO (ConduitT () (GRpcReply $(typeRefToType n r)) IO ())|] +computeMethodType n [ArgStream v] (RetStream r) + = [t|CompressMode -> IO (ConduitT $(typeRefToType n v) (GRpcReply $(typeRefToType n r)) IO ())|] +computeMethodType _ _ _ = fail "method signature not supported" + +typeRefToType :: Namer -> TypeRef -> Q Type +typeRefToType tNamer (ViaTH (LitT (StrTyLit s))) + = return $ ConT (mkName $ completeName tNamer s) +typeRefToType _tNamer (ViaTH ty) + = return ty +typeRefToType _ _ = error "this should never happen" + +completeName :: Namer -> String -> String +completeName namer name = firstUpper (namer (firstUpper name)) + +firstUpper :: String -> String +firstUpper [] = error "Empty names are not allowed" +firstUpper (x:rest) = toUpper x : rest + +firstLower :: String -> String +firstLower [] = error "Empty names are not allowed" +firstLower (x:rest) = toLower x : rest + +-- Parsing +-- ======= + +typeToServiceDef :: Type -> Q (Maybe (Service String String)) +typeToServiceDef toplevelty + = typeToServiceDef' <$> resolveTypeSynonyms toplevelty + where + typeToServiceDef' :: Type -> Maybe (Service String String) + typeToServiceDef' expanded + = do (sn, _, methods) <- tyD3 'Service expanded + methods' <- tyList methods + Service <$> tyString sn + <*> pure [] + <*> mapM typeToMethodDef methods' + + typeToMethodDef :: Type -> Maybe (Method String) + typeToMethodDef ty + = do (mn, _, args, ret) <- tyD4 'Method ty + args' <- tyList args + Method <$> tyString mn + <*> pure [] + <*> mapM typeToArgDef args' + <*> typeToRetDef ret + + typeToArgDef :: Type -> Maybe Argument + typeToArgDef ty + = ArgSingle <$> (tyD1 'ArgSingle ty >>= typeToTypeRef) + <|> ArgStream <$> (tyD1 'ArgStream ty >>= typeToTypeRef) + + typeToRetDef :: Type -> Maybe Return + typeToRetDef ty + = RetNothing <$ tyD0 'RetNothing ty + <|> RetSingle <$> (tyD1 'RetSingle ty >>= typeToTypeRef) + <|> (do (e, v) <- tyD2 'RetThrows ty + RetThrows <$> typeToTypeRef e <*> typeToTypeRef v) + <|> RetStream <$> (tyD1 'RetStream ty >>= typeToTypeRef) + + typeToTypeRef :: Type -> Maybe TypeRef + typeToTypeRef ty + = (do (_,innerTy) <- tyD2 'ViaSchema ty + return (ViaTH innerTy)) + <|> (do (_,innerTy,_) <- tyD3 'ViaRegistry ty + return (ViaTH innerTy)) + +tyString :: Type -> Maybe String +tyString (SigT t _) + = tyString t +tyString (LitT (StrTyLit s)) + = Just s +tyString _ + = Nothing + +tyList :: Type -> Maybe [Type] +tyList (SigT t _) + = tyList t +tyList PromotedNilT + = Just [] +tyList (AppT (AppT PromotedConsT ty) rest) + = (ty :) <$> tyList rest +tyList _ = Nothing + +tyD0 :: Name -> Type -> Maybe () +tyD0 name (SigT t _) = tyD0 name t +tyD0 name (PromotedT c) + | c == name = Just () + | otherwise = Nothing +tyD0 _ _ = Nothing + +tyD1 :: Name -> Type -> Maybe Type +tyD1 name (SigT t _) = tyD1 name t +tyD1 name (AppT (PromotedT c) x) + | c == name = Just x + | otherwise = Nothing +tyD1 _ _ = Nothing + +tyD2 :: Name -> Type -> Maybe (Type, Type) +tyD2 name (SigT t _) = tyD2 name t +tyD2 name (AppT (AppT (PromotedT c) x) y) + | c == name = Just (x, y) + | otherwise = Nothing +tyD2 _ _ = Nothing + +tyD3 :: Name -> Type -> Maybe (Type, Type, Type) +tyD3 name (SigT t _) = tyD3 name t +tyD3 name (AppT (AppT (AppT (PromotedT c) x) y) z) + | c == name = Just (x, y, z) + | otherwise = Nothing +tyD3 _ _ = Nothing + +tyD4 :: Name -> Type -> Maybe (Type, Type, Type, Type) +tyD4 name (SigT t _) = tyD4 name t +tyD4 name (AppT (AppT (AppT (AppT (PromotedT c) x) y) z) u) + | c == name = Just (x, y, z, u) + | otherwise = Nothing +tyD4 _ _ = Nothing diff --git a/src/Mu/GRpc/Client/TyApps.hs b/src/Mu/GRpc/Client/TyApps.hs new file mode 100644 --- /dev/null +++ b/src/Mu/GRpc/Client/TyApps.hs @@ -0,0 +1,49 @@ +{-# language AllowAmbiguousTypes #-} +{-# language DataKinds #-} +{-# language FlexibleContexts #-} +{-# language GADTs #-} +{-# language MultiParamTypeClasses #-} +{-# language PolyKinds #-} +{-# language ScopedTypeVariables #-} +{-# language TypeApplications #-} +{-# language TypeOperators #-} +{-| +Description : Client for gRPC services using @TypeApplications@ + +For further information over initialization of the connection, +consult the . +-} +module Mu.GRpc.Client.TyApps ( + -- * Initialization of the gRPC client + GrpcClient +, GrpcClientConfig +, grpcClientConfigSimple +, setupGrpcClient' + -- * Call methods from the gRPC service +, gRpcCall +, CompressMode(..) +, GRpcReply(..) +) where + +import Network.GRPC.Client (CompressMode (..)) +import Network.GRPC.Client.Helpers + +import Mu.Rpc +import Mu.Schema + +import Mu.GRpc.Client.Internal + +-- | Call a method from a Mu definition. +-- This method is thought to be used with @TypeApplications@: +-- +-- > gRpcCall @"packageName" @ServiceDeclaration @"method" +-- +-- The additional arguments you must provide to 'gRpcCall' +-- depend on the signature of the method itself: +-- * The resulting value is always wrapped in 'GRpcReply'. +-- * A single input or output turns into a single value. +-- * A streaming input or output turns into a Conduit. +gRpcCall :: forall s methodName h. + (GRpcServiceMethodCall s (s :-->: methodName) h) + => GrpcClient -> h +gRpcCall = gRpcServiceMethodCall (Proxy @s) (Proxy @(s :-->: methodName))