eeva/gRPC-haskell
1
0
Fork 0
mirror of https://github.com/unclechu/gRPC-haskell.git synced 2024-11-26 21:19:43 +01:00
Code Issues Projects Releases Wiki Activity

Preliminary streaming mode support (client streaming, server streaming, bidirectional) (#37)

Browse source 
* Tweak runOps param order, inline common op sequences, clean up serverHandleNormalCall

* More ops sequence inlining for clarity, experimenting with Managed

* Checkpoint: preliminary support for all streaming modes; much cleanup/refactoring and api design still needed

* Use mempty for default StatusDetails; tweak bad status matching mechanism

* Preliminary user-facing, server-streaming, low-level api and test

* renaming wibbles

* Preliminary user-facing, client-streaming, low-level api and test

* Move sendMsgs comb to Network.GRPC.LowLevel.Op; misc cleanup/DCR

* Modify bidi streaming to omit request payload

* Add transformers dep

* Preliminary user-facing low-level bidirectional streaming api and test

* Fix missing peek import

* Remove TimeoutSeconds params on streaming mode functions

* Fix serverHandleNormalCall rebase wart

* Fix rebase warts; minor hlint fixes and wibbles

* Post-rebase tweaks to optional payload use in serverRequestCall (i.e., now respects payloadHandling again)

* Cleanup/refactor serverRequestCall

* Fix comment

* Change ServerRWHandler type so that handler does not have to invoke a finalizer

* Change ServerReaderHandler type so that handler does not have to invoke a finalizer

* Simplify serverWriter interface and ServerWriterHandler structure

* Simplify serverRW (get rid of exec param), improve bidi streaming tests

* Use ExceptT in serverRW impl

* Change ServerRWHandler type to pass recv/send operations.

* Renaming

* Define ClientRWHandler, pass recv/send ops

* wibbles

* Use ExceptT in clientRW impl

* Add DataKinded phantom typing to RegisteredMethod; misc cleanup

* Simplify sendMsgs interface; add SingleSend type and related helpers

* Rename SingleSend to SendSingle, use ExceptT to clean up {client,server}Writer and sendMsgs

* More ExceptT cleanup in clientWriter

* Factor out reusable bits of clientWriter

* Shrink ServerReaderHandler

* Delete stale comments

* Use common machinery for all streaming modes; make handler types more consistent

* wibbles
This commit is contained in:
Joel Stanley 2016-07-06 08:59:38 -05:00
parent e9f6340e40
commit 96d12c1e6c
18 changed files with 785 additions and 344 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
cbits/grpc_haskell.c
View file

@ -401,13 +401,10 @@ gpr_timespec* call_details_get_deadline(grpc_call_details* details){
return &(details->deadline); return &(details->deadline);
} }
void* grpc_server_register_method_(grpc_server* server, const char* method, void* grpc_server_register_method_(
const char* host){ grpc_server* server, const char* method,
//NOTE: grpc 0.14.0 added more params to this function. None of our code takes const char* host, grpc_server_register_method_payload_handling payload_handling ){
//advantage of them, so we hardcode to the equivalent of 0.13.0's behavior. return grpc_server_register_method(server, method, host, payload_handling, 0);
return grpc_server_register_method(server, method, host,
GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER,
0);
} }
grpc_arg* create_arg_array(size_t n){ grpc_arg* create_arg_array(size_t n){

7
examples/echo/echo-server/Main.hs
View file

@ -1,3 +1,4 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE OverloadedLists #-} {-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE RecordWildCards #-}
@ -35,7 +36,7 @@ regMain = withGRPC $ \grpc -> do
let methods = [(MethodName "/echo.Echo/DoEcho", Normal)] let methods = [(MethodName "/echo.Echo/DoEcho", Normal)]
withServer grpc (ServerConfig "localhost" 50051 methods []) $ \server -> withServer grpc (ServerConfig "localhost" 50051 methods []) $ \server ->
forever $ do forever $ do
let method = head (registeredMethods server) let method = head (normalMethods server)
result <- serverHandleNormalCall server method serverMeta $ result <- serverHandleNormalCall server method serverMeta $
\_call reqBody _reqMeta -> return (reqBody, serverMeta, StatusOk, \_call reqBody _reqMeta -> return (reqBody, serverMeta, StatusOk,
StatusDetails "") StatusDetails "")
@ -44,7 +45,7 @@ regMain = withGRPC $ \grpc -> do
Right _ -> return () Right _ -> return ()
-- | loop to fork n times -- | loop to fork n times
regLoop :: Server -> RegisteredMethod -> IO () regLoop :: Server -> RegisteredMethod 'Normal -> IO ()
regLoop server method = forever $ do regLoop server method = forever $ do
result <- serverHandleNormalCall server method serverMeta $ result <- serverHandleNormalCall server method serverMeta $
\_call reqBody _reqMeta -> return (reqBody, serverMeta, StatusOk, \_call reqBody _reqMeta -> return (reqBody, serverMeta, StatusOk,
@ -58,7 +59,7 @@ regMainThreaded = do
withGRPC $ \grpc -> do withGRPC $ \grpc -> do
let methods = [(MethodName "/echo.Echo/DoEcho", Normal)] let methods = [(MethodName "/echo.Echo/DoEcho", Normal)]
withServer grpc (ServerConfig "localhost" 50051 methods []) $ \server -> do withServer grpc (ServerConfig "localhost" 50051 methods []) $ \server -> do
let method = head (registeredMethods server) let method = head (normalMethods server)
tid1 <- async $ regLoop server method tid1 <- async $ regLoop server method
tid2 <- async $ regLoop server method tid2 <- async $ regLoop server method
wait tid1 wait tid1

15
grpc-haskell.cabal
View file

@ -30,6 +30,15 @@ library
, bytestring ==0.10.* , bytestring ==0.10.*
, stm == 2.4.* , stm == 2.4.*
, containers ==0.5.* , containers ==0.5.*
, managed >= 1.0.5 && < 1.1
, pipes ==4.1.*
, transformers
, async
, tasty >= 0.11 && <0.12
, tasty-hunit >= 0.9 && <0.10
, safe
c-sources: c-sources:
cbits/grpc_haskell.c cbits/grpc_haskell.c
exposed-modules: exposed-modules:
@ -70,7 +79,6 @@ library
include-dirs: include include-dirs: include
hs-source-dirs: src hs-source-dirs: src
default-extensions: CPP default-extensions: CPP
if flag(debug) if flag(debug)
CPP-Options: -DDEBUG CPP-Options: -DDEBUG
CC-Options: -DGRPC_HASKELL_DEBUG CC-Options: -DGRPC_HASKELL_DEBUG
@ -115,6 +123,10 @@ test-suite test
, tasty >= 0.11 && <0.12 , tasty >= 0.11 && <0.12
, tasty-hunit >= 0.9 && <0.10 , tasty-hunit >= 0.9 && <0.10
, containers ==0.5.* , containers ==0.5.*
, managed >= 1.0.5 && < 1.1
, pipes ==4.1.*
, transformers
, safe
other-modules: other-modules:
LowLevelTests, LowLevelTests,
LowLevelTests.Op, LowLevelTests.Op,
@ -125,7 +137,6 @@ test-suite test
main-is: Properties.hs main-is: Properties.hs
type: exitcode-stdio-1.0 type: exitcode-stdio-1.0
default-extensions: CPP default-extensions: CPP
if flag(debug) if flag(debug)
CPP-Options: -DDEBUG CPP-Options: -DDEBUG
CC-Options: -DGRPC_HASKELL_DEBUG CC-Options: -DGRPC_HASKELL_DEBUG

5
include/grpc_haskell.h
View file

@ -137,8 +137,9 @@ char* call_details_get_host(grpc_call_details* details);
gpr_timespec* call_details_get_deadline(grpc_call_details* details); gpr_timespec* call_details_get_deadline(grpc_call_details* details);
void* grpc_server_register_method_(grpc_server* server, const char* method, void* grpc_server_register_method_(
const char* host); grpc_server* server, const char* method, const char* host,
grpc_server_register_method_payload_handling payload_handling);
//c2hs doesn't support #const pragmas referring to #define'd strings, so we use //c2hs doesn't support #const pragmas referring to #define'd strings, so we use
//this enum as a workaround. These are converted into actual GRPC #defines in //this enum as a workaround. These are converted into actual GRPC #defines in

12
src/Network/GRPC/LowLevel.hs
View file

@ -32,14 +32,17 @@ GRPC
-- * Server -- * Server
, ServerConfig(..) , ServerConfig(..)
, Server , Server(normalMethods, sstreamingMethods, cstreamingMethods,
, ServerCall bidiStreamingMethods)
, registeredMethods , ServerCall(optionalPayload, requestMetadataRecv)
, withServer , withServer
, serverHandleNormalCall , serverHandleNormalCall
, withServerCall , withServerCall
, serverCallCancel , serverCallCancel
, serverCallIsExpired , serverCallIsExpired
, serverReader -- for client streaming
, serverWriter -- for server streaming
, serverRW -- for bidirectional streaming
-- * Client -- * Client
, ClientConfig(..) , ClientConfig(..)
@ -50,6 +53,9 @@ GRPC
, withClient , withClient
, clientRegisterMethod , clientRegisterMethod
, clientRequest , clientRequest
, clientReader -- for server streaming
, clientWriter -- for client streaming
, clientRW -- for bidirectional streaming
, withClientCall , withClientCall
, withClientCallParent , withClientCallParent
, clientCallCancel , clientCallCancel

48
src/Network/GRPC/LowLevel/Call.hs
View file

@ -1,16 +1,19 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE FlexibleInstances #-}
-- | This module defines data structures and operations pertaining to registered -- | This module defines data structures and operations pertaining to registered
-- calls; for unregistered call support, see -- calls; for unregistered call support, see
-- `Network.GRPC.LowLevel.Call.Unregistered`. -- `Network.GRPC.LowLevel.Call.Unregistered`.
module Network.GRPC.LowLevel.Call where module Network.GRPC.LowLevel.Call where
import Control.Monad
import Data.ByteString (ByteString) import Data.ByteString (ByteString)
import Data.String (IsString) import Data.String (IsString)
import Foreign.Marshal.Alloc (free) #ifdef DEBUG
import Foreign.Ptr (Ptr) import Foreign.Storable (peek)
#endif
import System.Clock import System.Clock
import qualified Network.GRPC.Unsafe as C import qualified Network.GRPC.Unsafe as C
@ -18,9 +21,13 @@ import qualified Network.GRPC.Unsafe.Op as C
import Network.GRPC.LowLevel.GRPC (MetadataMap, grpcDebug) import Network.GRPC.LowLevel.GRPC (MetadataMap, grpcDebug)
-- | Models the four types of RPC call supported by gRPC. We currently only -- | Models the four types of RPC call supported by gRPC (and correspond to
-- support the first alternative, and only in a preliminary fashion. -- DataKinds phantom types on RegisteredMethods).
data GRPCMethodType = Normal | ClientStreaming | ServerStreaming | BiDiStreaming data GRPCMethodType
= Normal
| ClientStreaming
| ServerStreaming
| BiDiStreaming
deriving (Show, Eq, Ord, Enum) deriving (Show, Eq, Ord, Enum)
newtype MethodName = MethodName {unMethodName :: String} newtype MethodName = MethodName {unMethodName :: String}
@ -40,15 +47,18 @@ endpoint :: Host -> Port -> Endpoint
endpoint (Host h) (Port p) = Endpoint (h ++ ":" ++ show p) endpoint (Host h) (Port p) = Endpoint (h ++ ":" ++ show p)
-- | Represents a registered method. Methods can optionally be registered in -- | Represents a registered method. Methods can optionally be registered in
-- order to make the C-level request/response code simpler. -- order to make the C-level request/response code simpler. Before making or
-- Before making or awaiting a registered call, the -- awaiting a registered call, the method must be registered with the client
-- method must be registered with the client (see 'clientRegisterMethod') and -- (see 'clientRegisterMethod') and the server (see 'serverRegisterMethod').
-- the server (see 'serverRegisterMethod'). -- Contains state for identifying that method in the underlying gRPC
-- Contains state for identifying that method in the underlying gRPC library. -- library. Note that we use a DataKind-ed phantom type to help constrain use of
data RegisteredMethod = RegisteredMethod {methodType :: GRPCMethodType, -- different kinds of registered methods.
methodName :: MethodName, data RegisteredMethod (mt :: GRPCMethodType) = RegisteredMethod
methodEndpoint :: Endpoint, { methodType :: GRPCMethodType
methodHandle :: C.CallHandle} , methodName :: MethodName
, methodEndpoint :: Endpoint
, methodHandle :: C.CallHandle
}
-- | Represents one GRPC call (i.e. request) on the client. -- | Represents one GRPC call (i.e. request) on the client.
-- This is used to associate send/receive 'Op's with a request. -- This is used to associate send/receive 'Op's with a request.
@ -70,6 +80,14 @@ serverCallCancel :: ServerCall -> C.StatusCode -> String -> IO ()
serverCallCancel sc code reason = serverCallCancel sc code reason =
C.grpcCallCancelWithStatus (unServerCall sc) code reason C.reserved C.grpcCallCancelWithStatus (unServerCall sc) code reason C.reserved
-- | NB: For now, we've assumed that the method type is all the info we need to
-- decide the server payload handling method.
payloadHandling :: GRPCMethodType -> C.ServerRegisterMethodPayloadHandling
payloadHandling Normal = C.SrmPayloadReadInitialByteBuffer
payloadHandling ClientStreaming = C.SrmPayloadNone
payloadHandling ServerStreaming = C.SrmPayloadReadInitialByteBuffer
payloadHandling BiDiStreaming = C.SrmPayloadNone
serverCallIsExpired :: ServerCall -> IO Bool serverCallIsExpired :: ServerCall -> IO Bool
serverCallIsExpired sc = do serverCallIsExpired sc = do
currTime <- getTime Monotonic currTime <- getTime Monotonic

11
src/Network/GRPC/LowLevel/Call/Unregistered.hs
View file

@ -5,6 +5,9 @@ module Network.GRPC.LowLevel.Call.Unregistered where
import Control.Monad import Control.Monad
import Foreign.Marshal.Alloc (free) import Foreign.Marshal.Alloc (free)
import Foreign.Ptr (Ptr) import Foreign.Ptr (Ptr)
#ifdef DEBUG
import Foreign.Storable (peek)
#endif
import System.Clock (TimeSpec) import System.Clock (TimeSpec)
import Network.GRPC.LowLevel.Call (Host (..), MethodName (..)) import Network.GRPC.LowLevel.Call (Host (..), MethodName (..))
@ -30,14 +33,14 @@ serverCallCancel sc code reason =
debugServerCall :: ServerCall -> IO () debugServerCall :: ServerCall -> IO ()
#ifdef DEBUG #ifdef DEBUG
debugServerCall call@ServerCall{..} = do debugServerCall ServerCall{..} = do
let (C.Call ptr) = unServerCall let C.Call ptr = unServerCall
grpcDebug $ "debugServerCall(U): server call: " ++ (show ptr) grpcDebug $ "debugServerCall(U): server call: " ++ show ptr
grpcDebug $ "debugServerCall(U): metadata: " grpcDebug $ "debugServerCall(U): metadata: "
++ show requestMetadataRecv ++ show requestMetadataRecv
forM_ parentPtr $ \parentPtr' -> do forM_ parentPtr $ \parentPtr' -> do
grpcDebug $ "debugServerCall(U): parent ptr: " ++ show parentPtr' grpcDebug $ "debugServerCall(U): parent ptr: " ++ show parentPtr'
(C.Call parent) <- peek parentPtr' C.Call parent <- peek parentPtr'
grpcDebug $ "debugServerCall(U): parent: " ++ show parent grpcDebug $ "debugServerCall(U): parent: " ++ show parent
grpcDebug $ "debugServerCall(U): deadline: " ++ show callDeadline grpcDebug $ "debugServerCall(U): deadline: " ++ show callDeadline
grpcDebug $ "debugServerCall(U): method: " ++ show callMethod grpcDebug $ "debugServerCall(U): method: " ++ show callMethod

208
src/Network/GRPC/LowLevel/Client.hs
View file

@ -1,25 +1,33 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ViewPatterns #-}
-- | This module defines data structures and operations pertaining to registered -- | This module defines data structures and operations pertaining to registered
-- clients using registered calls; for unregistered support, see -- clients using registered calls; for unregistered support, see
-- `Network.GRPC.LowLevel.Client.Unregistered`. -- `Network.GRPC.LowLevel.Client.Unregistered`.
module Network.GRPC.LowLevel.Client where module Network.GRPC.LowLevel.Client where
import Control.Arrow
import Control.Exception (bracket, finally) import Control.Exception (bracket, finally)
import Control.Monad (join) import Control.Monad
import Control.Monad.Trans.Class (MonadTrans(lift))
import Control.Monad.Trans.Except
import Data.ByteString (ByteString) import Data.ByteString (ByteString)
import Foreign.Ptr (nullPtr) import Network.GRPC.LowLevel.Call
import Network.GRPC.LowLevel.CompletionQueue
import Network.GRPC.LowLevel.GRPC
import Network.GRPC.LowLevel.Op
import qualified Network.GRPC.Unsafe as C import qualified Network.GRPC.Unsafe as C
import qualified Network.GRPC.Unsafe.ChannelArgs as C import qualified Network.GRPC.Unsafe.ChannelArgs as C
import qualified Network.GRPC.Unsafe.Constants as C import qualified Network.GRPC.Unsafe.Constants as C
import qualified Network.GRPC.Unsafe.Op as C import qualified Network.GRPC.Unsafe.Op as C
import qualified Network.GRPC.Unsafe.Time as C import qualified Network.GRPC.Unsafe.Time as C
import qualified Pipes as P
import Network.GRPC.LowLevel.Call import qualified Pipes.Core as P
import Network.GRPC.LowLevel.CompletionQueue
import Network.GRPC.LowLevel.GRPC
import Network.GRPC.LowLevel.Op
-- | Represents the context needed to perform client-side gRPC operations. -- | Represents the context needed to perform client-side gRPC operations.
data Client = Client {clientChannel :: C.Channel, data Client = Client {clientChannel :: C.Channel,
@ -72,21 +80,19 @@ clientConnectivity Client{..} =
-- 'clientRequest'. -- 'clientRequest'.
clientRegisterMethod :: Client clientRegisterMethod :: Client
-> MethodName -> MethodName
-- ^ method name, e.g. "/foo"
-> GRPCMethodType -> GRPCMethodType
-> IO RegisteredMethod -> IO (RegisteredMethod mt)
clientRegisterMethod Client{..} meth Normal = do clientRegisterMethod Client{..} meth mty = do
let e = clientEndpoint clientConfig let e = clientEndpoint clientConfig
handle <- C.grpcChannelRegisterCall clientChannel RegisteredMethod mty meth e <$>
C.grpcChannelRegisterCall clientChannel
(unMethodName meth) (unEndpoint e) C.reserved (unMethodName meth) (unEndpoint e) C.reserved
return $ RegisteredMethod Normal meth e handle
clientRegisterMethod _ _ _ = error "Streaming methods not yet implemented."
-- | Create a new call on the client for a registered method. -- | Create a new call on the client for a registered method.
-- Returns 'Left' if the CQ is shutting down or if the job to create a call -- Returns 'Left' if the CQ is shutting down or if the job to create a call
-- timed out. -- timed out.
clientCreateCall :: Client clientCreateCall :: Client
-> RegisteredMethod -> RegisteredMethod mt
-> TimeoutSeconds -> TimeoutSeconds
-> IO (Either GRPCIOError ClientCall) -> IO (Either GRPCIOError ClientCall)
clientCreateCall c rm ts = clientCreateCallParent c rm ts Nothing clientCreateCall c rm ts = clientCreateCallParent c rm ts Nothing
@ -95,9 +101,9 @@ clientCreateCall c rm ts = clientCreateCallParent c rm ts Nothing
-- a client call with an optional parent server call. This allows for cascading -- a client call with an optional parent server call. This allows for cascading
-- call cancellation from the `ServerCall` to the `ClientCall`. -- call cancellation from the `ServerCall` to the `ClientCall`.
clientCreateCallParent :: Client clientCreateCallParent :: Client
-> RegisteredMethod -> RegisteredMethod mt
-> TimeoutSeconds -> TimeoutSeconds
-> (Maybe ServerCall) -> Maybe ServerCall
-- ^ Optional parent call for cascading cancellation. -- ^ Optional parent call for cascading cancellation.
-> IO (Either GRPCIOError ClientCall) -> IO (Either GRPCIOError ClientCall)
clientCreateCallParent Client{..} RegisteredMethod{..} timeout parent = do clientCreateCallParent Client{..} RegisteredMethod{..} timeout parent = do
@ -107,7 +113,7 @@ clientCreateCallParent Client{..} RegisteredMethod{..} timeout parent = do
-- | Handles safe creation and cleanup of a client call -- | Handles safe creation and cleanup of a client call
withClientCall :: Client withClientCall :: Client
-> RegisteredMethod -> RegisteredMethod mt
-> TimeoutSeconds -> TimeoutSeconds
-> (ClientCall -> IO (Either GRPCIOError a)) -> (ClientCall -> IO (Either GRPCIOError a))
-> IO (Either GRPCIOError a) -> IO (Either GRPCIOError a)
@ -119,7 +125,7 @@ withClientCall client regmethod timeout f =
-- `ServerCall` to the created `ClientCall`. Obviously, this is only useful if -- `ServerCall` to the created `ClientCall`. Obviously, this is only useful if
-- the given gRPC client is also a server. -- the given gRPC client is also a server.
withClientCallParent :: Client withClientCallParent :: Client
-> RegisteredMethod -> RegisteredMethod mt
-> TimeoutSeconds -> TimeoutSeconds
-> (Maybe ServerCall) -> (Maybe ServerCall)
-- ^ Optional parent call for cascading cancellation. -- ^ Optional parent call for cascading cancellation.
@ -135,8 +141,8 @@ withClientCallParent client regmethod timeout parent f = do
data NormalRequestResult = NormalRequestResult data NormalRequestResult = NormalRequestResult
{ rspBody :: ByteString { rspBody :: ByteString
, initMD :: MetadataMap -- initial metadata , initMD :: MetadataMap -- ^ initial metadata
, trailMD :: MetadataMap -- trailing metadata , trailMD :: MetadataMap -- ^ trailing metadata
, rspCode :: C.StatusCode , rspCode :: C.StatusCode
, details :: StatusDetails , details :: StatusDetails
} }
@ -156,58 +162,136 @@ compileNormalRequestResults x =
Just (_meta, status, details) -> Just (_meta, status, details) ->
Left (GRPCIOBadStatusCode status (StatusDetails details)) Left (GRPCIOBadStatusCode status (StatusDetails details))
-- | Make a request of the given method with the given body. Returns the --------------------------------------------------------------------------------
-- server's response. TODO: This is preliminary until we figure out how many -- clientReader (client side of server streaming mode)
-- different variations on sending request ops will be needed for full gRPC
-- functionality. -- | First parameter is initial server metadata.
clientRequest :: Client type ClientReaderHandler = MetadataMap -> StreamRecv -> Streaming ()
-> RegisteredMethod
clientReader :: Client
-> RegisteredMethod 'ServerStreaming
-> TimeoutSeconds
-> ByteString -- ^ The body of the request
-> MetadataMap -- ^ Metadata to send with the request
-> ClientReaderHandler
-> IO (Either GRPCIOError (MetadataMap, C.StatusCode, StatusDetails))
clientReader cl@Client{ clientCQ = cq } rm tm body initMeta f =
withClientCall cl rm tm go
where
go cc@(unClientCall -> c) = runExceptT $ do
lift (debugClientCall cc)
runOps' c cq [ OpSendInitialMetadata initMeta
, OpSendMessage body
, OpSendCloseFromClient
]
srvMD <- recvInitialMetadata c cq
runStreamingProxy "clientReader'" c cq (f srvMD streamRecv)
recvStatusOnClient c cq
--------------------------------------------------------------------------------
-- clientWriter (client side of client streaming mode)
type ClientWriterHandler = StreamSend -> Streaming ()
type ClientWriterResult = (Maybe ByteString, MetadataMap, MetadataMap,
C.StatusCode, StatusDetails)
clientWriter :: Client
-> RegisteredMethod 'ClientStreaming
-> TimeoutSeconds
-> MetadataMap -- ^ Initial client metadata
-> ClientWriterHandler
-> IO (Either GRPCIOError ClientWriterResult)
clientWriter cl rm tm initMeta =
withClientCall cl rm tm . clientWriterCmn cl initMeta
clientWriterCmn :: Client -- ^ The active client
-> MetadataMap -- ^ Initial client metadata
-> ClientWriterHandler
-> ClientCall -- ^ The active client call
-> IO (Either GRPCIOError ClientWriterResult)
clientWriterCmn (clientCQ -> cq) initMeta f cc@(unClientCall -> c) =
runExceptT $ do
lift (debugClientCall cc)
sendInitialMetadata c cq initMeta
runStreamingProxy "clientWriterCmn" c cq (f streamSend)
sendSingle c cq OpSendCloseFromClient
let ops = [OpRecvInitialMetadata, OpRecvMessage, OpRecvStatusOnClient]
runOps' c cq ops >>= \case
CWRFinal mmsg initMD trailMD st ds
-> return (mmsg, initMD, trailMD, st, ds)
_ -> throwE (GRPCIOInternalUnexpectedRecv "clientWriter")
pattern CWRFinal mmsg initMD trailMD st ds
<- [ OpRecvInitialMetadataResult initMD
, OpRecvMessageResult mmsg
, OpRecvStatusOnClientResult trailMD st (StatusDetails -> ds)
]
--------------------------------------------------------------------------------
-- clientRW (client side of bidirectional streaming mode)
-- | First parameter is initial server metadata.
type ClientRWHandler = MetadataMap -> StreamRecv -> StreamSend -> Streaming ()
-- | For bidirectional-streaming registered requests
clientRW :: Client
-> RegisteredMethod 'BiDiStreaming
-> TimeoutSeconds
-> MetadataMap
-- ^ request metadata
-> ClientRWHandler
-> IO (Either GRPCIOError (MetadataMap, C.StatusCode, StatusDetails))
clientRW c@Client{ clientCQ = cq } rm tm initMeta f =
withClientCall c rm tm go
where
go cc@(unClientCall -> call) = runExceptT $ do
lift (debugClientCall cc)
sendInitialMetadata call cq initMeta
srvMeta <- recvInitialMetadata call cq
runStreamingProxy "clientRW" call cq (f srvMeta streamRecv streamSend)
runOps' call cq [OpSendCloseFromClient] -- WritesDone()
recvStatusOnClient call cq -- Finish()
--------------------------------------------------------------------------------
-- clientRequest (client side of normal request/response)
-- | Make a request of the given method with the given body. Returns the
-- server's response.
clientRequest :: Client
-> RegisteredMethod 'Normal
-> TimeoutSeconds -> TimeoutSeconds
-- ^ Timeout of both the grpc_call and the max time to wait for
-- the completion of the batch. TODO: I think we will need to
-- decouple the lifetime of the call from the queue deadline once
-- we expose functionality for streaming calls, where one call
-- object persists across many batches.
-> ByteString -> ByteString
-- ^ The body of the request -- ^ The body of the request
-> MetadataMap -> MetadataMap
-- ^ Metadata to send with the request -- ^ Metadata to send with the request
-> IO (Either GRPCIOError NormalRequestResult) -> IO (Either GRPCIOError NormalRequestResult)
clientRequest client@(Client{..}) rm@(RegisteredMethod{..}) clientRequest c@Client{ clientCQ = cq } rm tm body initMeta =
timeLimit body meta = withClientCall c rm tm (fmap join . go)
fmap join $ case methodType of where
Normal -> withClientCall client rm timeLimit $ \call -> do go cc@(unClientCall -> call) = do
grpcDebug "clientRequest(R): created call." grpcDebug "clientRequest(R): created call."
debugClientCall call debugClientCall cc
let call' = unClientCall call -- NB: the send and receive operations below *must* be in separate
-- NOTE: sendOps and recvOps *must* be in separate batches or -- batches, or the client hangs when the server can't be reached.
-- the client hangs when the server can't be reached. runOps call cq
let sendOps = [OpSendInitialMetadata meta [ OpSendInitialMetadata initMeta
, OpSendMessage body , OpSendMessage body
, OpSendCloseFromClient] , OpSendCloseFromClient
sendRes <- runOps call' clientCQ sendOps ]
case sendRes of >>= \case
Left x -> do grpcDebug "clientRequest(R) : batch error sending."
return $ Left x
Right rs -> do
let recvOps = [OpRecvInitialMetadata,
OpRecvMessage,
OpRecvStatusOnClient]
recvRes <- runOps call' clientCQ recvOps
case recvRes of
Left x -> do Left x -> do
grpcDebug "clientRequest(R): batch error receiving." grpcDebug "clientRequest(R) : batch error sending."
return $ Left x
Right rs ->
runOps call cq
[ OpRecvInitialMetadata
, OpRecvMessage
, OpRecvStatusOnClient
]
>>= \case
Left x -> do
grpcDebug "clientRequest(R): batch error receiving.."
return $ Left x return $ Left x
Right rs' -> do Right rs' -> do
grpcDebug $ "clientRequest(R): got " ++ show rs' grpcDebug $ "clientRequest(R): got " ++ show rs'
return $ Right $ compileNormalRequestResults (rs ++ rs') return $ Right $ compileNormalRequestResults (rs ++ rs')
_ -> error "Streaming methods not yet implemented."
clientNormalRequestOps :: ByteString -> MetadataMap -> [Op]
clientNormalRequestOps body metadata =
[OpSendInitialMetadata metadata,
OpSendMessage body,
OpSendCloseFromClient,
OpRecvInitialMetadata,
OpRecvMessage,
OpRecvStatusOnClient]

19
src/Network/GRPC/LowLevel/Client/Unregistered.hs
View file

@ -2,6 +2,7 @@
module Network.GRPC.LowLevel.Client.Unregistered where module Network.GRPC.LowLevel.Client.Unregistered where
import Control.Arrow
import Control.Exception (finally) import Control.Exception (finally)
import Control.Monad (join) import Control.Monad (join)
import Data.ByteString (ByteString) import Data.ByteString (ByteString)
@ -14,7 +15,6 @@ import Network.GRPC.LowLevel.Call
import Network.GRPC.LowLevel.Client (Client (..), import Network.GRPC.LowLevel.Client (Client (..),
NormalRequestResult (..), NormalRequestResult (..),
clientEndpoint, clientEndpoint,
clientNormalRequestOps,
compileNormalRequestResults) compileNormalRequestResults)
import Network.GRPC.LowLevel.CompletionQueue (TimeoutSeconds) import Network.GRPC.LowLevel.CompletionQueue (TimeoutSeconds)
import qualified Network.GRPC.LowLevel.CompletionQueue.Unregistered as U import qualified Network.GRPC.LowLevel.CompletionQueue.Unregistered as U
@ -60,11 +60,14 @@ clientRequest :: Client
-- ^ Request metadata. -- ^ Request metadata.
-> IO (Either GRPCIOError NormalRequestResult) -> IO (Either GRPCIOError NormalRequestResult)
clientRequest client@Client{..} meth timeLimit body meta = clientRequest client@Client{..} meth timeLimit body meta =
fmap join $ do fmap join $ withClientCall client meth timeLimit $ \call -> do
withClientCall client meth timeLimit $ \call -> do results <- runOps (unClientCall call) clientCQ
let ops = clientNormalRequestOps body meta [ OpSendInitialMetadata meta
results <- runOps (unClientCall call) clientCQ ops , OpSendMessage body
, OpSendCloseFromClient
, OpRecvInitialMetadata
, OpRecvMessage
, OpRecvStatusOnClient
]
grpcDebug "clientRequest(U): ops ran." grpcDebug "clientRequest(U): ops ran."
case results of return $ right compileNormalRequestResults results
Left x -> return $ Left x
Right rs -> return $ Right $ compileNormalRequestResults rs

114
src/Network/GRPC/LowLevel/CompletionQueue.hs
View file

@ -10,7 +10,11 @@
-- implementation details to both are kept in -- implementation details to both are kept in
-- `Network.GRPC.LowLevel.CompletionQueue.Internal`. -- `Network.GRPC.LowLevel.CompletionQueue.Internal`.
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE MultiWayIf #-}
{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
module Network.GRPC.LowLevel.CompletionQueue module Network.GRPC.LowLevel.CompletionQueue
( CompletionQueue ( CompletionQueue
@ -33,12 +37,15 @@ import Control.Concurrent.STM (atomically, check)
import Control.Concurrent.STM.TVar (newTVarIO, readTVar, import Control.Concurrent.STM.TVar (newTVarIO, readTVar,
writeTVar) writeTVar)
import Control.Exception (bracket) import Control.Exception (bracket)
import Control.Monad.Trans.Class (MonadTrans(lift))
import Control.Monad.Trans.Except
import Control.Monad (liftM2) import Control.Monad (liftM2)
import Control.Monad.Managed
import Data.IORef (newIORef) import Data.IORef (newIORef)
import Data.List (intersperse) import Data.List (intersperse)
import Foreign.Marshal.Alloc (free, malloc) import Foreign.Marshal.Alloc (free, malloc)
import Foreign.Ptr (nullPtr) import Foreign.Ptr (Ptr, nullPtr)
import Foreign.Storable (peek) import Foreign.Storable (Storable, peek)
import qualified Network.GRPC.Unsafe as C import qualified Network.GRPC.Unsafe as C
import qualified Network.GRPC.Unsafe.Constants as C import qualified Network.GRPC.Unsafe.Constants as C
import qualified Network.GRPC.Unsafe.Metadata as C import qualified Network.GRPC.Unsafe.Metadata as C
@ -63,11 +70,8 @@ createCompletionQueue _ = do
currentPushers <- newTVarIO 0 currentPushers <- newTVarIO 0
shuttingDown <- newTVarIO False shuttingDown <- newTVarIO False
nextTag <- newIORef minBound nextTag <- newIORef minBound
return $ CompletionQueue{..} return CompletionQueue{..}
-- TODO: I'm thinking it might be easier to use 'Either' uniformly everywhere
-- even when it's isomorphic to 'Maybe'. If that doesn't turn out to be the
-- case, switch these to 'Maybe'.
-- | Very simple wrapper around 'grpcCallStartBatch'. Throws 'GRPCIOShutdown' -- | Very simple wrapper around 'grpcCallStartBatch'. Throws 'GRPCIOShutdown'
-- without calling 'grpcCallStartBatch' if the queue is shutting down. -- without calling 'grpcCallStartBatch' if the queue is shutting down.
-- Throws 'CallError' if 'grpcCallStartBatch' returns a non-OK code. -- Throws 'CallError' if 'grpcCallStartBatch' returns a non-OK code.
@ -87,7 +91,7 @@ startBatch cq@CompletionQueue{..} call opArray opArraySize tag =
-- queue after we begin the shutdown process. Errors with -- queue after we begin the shutdown process. Errors with
-- 'GRPCIOShutdownFailure' if the queue can't be shut down within 5 seconds. -- 'GRPCIOShutdownFailure' if the queue can't be shut down within 5 seconds.
shutdownCompletionQueue :: CompletionQueue -> IO (Either GRPCIOError ()) shutdownCompletionQueue :: CompletionQueue -> IO (Either GRPCIOError ())
shutdownCompletionQueue (CompletionQueue{..}) = do shutdownCompletionQueue CompletionQueue{..} = do
atomically $ writeTVar shuttingDown True atomically $ writeTVar shuttingDown True
atomically $ readTVar currentPushers >>= \x -> check (x == 0) atomically $ readTVar currentPushers >>= \x -> check (x == 0)
atomically $ readTVar currentPluckers >>= \x -> check (x == 0) atomically $ readTVar currentPluckers >>= \x -> check (x == 0)
@ -105,7 +109,7 @@ shutdownCompletionQueue (CompletionQueue{..}) = do
ev <- C.withDeadlineSeconds 1 $ \deadline -> ev <- C.withDeadlineSeconds 1 $ \deadline ->
C.grpcCompletionQueueNext unsafeCQ deadline C.reserved C.grpcCompletionQueueNext unsafeCQ deadline C.reserved
grpcDebug $ "drainLoop: next() call got " ++ show ev grpcDebug $ "drainLoop: next() call got " ++ show ev
case (C.eventCompletionType ev) of case C.eventCompletionType ev of
C.QueueShutdown -> return () C.QueueShutdown -> return ()
C.QueueTimeout -> drainLoop C.QueueTimeout -> drainLoop
C.OpComplete -> drainLoop C.OpComplete -> drainLoop
@ -133,68 +137,46 @@ channelCreateCall
-- | Create the call object to handle a registered call. -- | Create the call object to handle a registered call.
serverRequestCall :: C.Server serverRequestCall :: C.Server
-> CompletionQueue -> CompletionQueue
-> RegisteredMethod -> RegisteredMethod mt
-> IO (Either GRPCIOError ServerCall) -> IO (Either GRPCIOError ServerCall)
serverRequestCall serverRequestCall s cq@CompletionQueue{.. } RegisteredMethod{..} =
server cq@CompletionQueue{..} RegisteredMethod{..} = -- NB: The method type dictates whether or not a payload is present, according
withPermission Push cq $ -- to the payloadHandling function. We do not allocate a buffer for the
bracket (liftM2 (,) malloc malloc) -- payload when it is not present.
(\(p1,p2) -> free p1 >> free p2) withPermission Push cq . with allocs $ \(dead, call, pay, meta) -> do
$ \(deadlinePtr, callPtr) -> md <- peek meta
C.withByteBufferPtr $ \bbPtr ->
C.withMetadataArrayPtr $ \metadataArrayPtr -> do
metadataArray <- peek metadataArrayPtr
tag <- newTag cq tag <- newTag cq
grpcDebug $ "serverRequestCall(R): tag is " ++ show tag dbug $ "tag is " ++ show tag
callError <- C.grpcServerRequestRegisteredCall ce <- C.grpcServerRequestRegisteredCall s methodHandle call
server methodHandle callPtr deadlinePtr dead md pay unsafeCQ unsafeCQ tag
metadataArray bbPtr unsafeCQ unsafeCQ tag dbug $ "callError: " ++ show ce
grpcDebug $ "serverRequestCall(R): callError: " runExceptT $ case ce of
++ show callError C.CallOk -> do
if callError /= C.CallOk ExceptT $ do
then do grpcDebug "serverRequestCall(R): callError. cleaning up" r <- pluck cq tag Nothing
return $ Left $ GRPCIOCallError callError dbug $ "pluck finished:" ++ show r
else do pluckResult <- pluck cq tag Nothing return r
grpcDebug $ "serverRequestCall(R): finished pluck:" lift $
++ show pluckResult ServerCall
case pluckResult of <$> peek call
Left x -> do <*> C.getAllMetadataArray md
grpcDebug "serverRequestCall(R): cleanup pluck err" <*> (if havePay then toBS pay else return Nothing)
return $ Left x <*> liftM2 (+) (getTime Monotonic) (C.timeSpec <$> peek dead)
Right () -> do
rawCall <- peek callPtr
deadline <- convertDeadline deadlinePtr
payload <- convertPayload bbPtr
meta <- convertMeta metadataArrayPtr
let assembledCall = ServerCall rawCall
meta
payload
deadline
grpcDebug "serverRequestCall(R): About to return"
return $ Right assembledCall
where convertDeadline deadline = do
-- gRPC gives us a deadline that is just a delta, so we convert it -- gRPC gives us a deadline that is just a delta, so we convert it
-- to a proper deadline. -- to a proper deadline.
deadline' <- C.timeSpec <$> peek deadline _ -> throwE (GRPCIOCallError ce)
now <- getTime Monotonic where
return $ now + deadline' allocs = (,,,) <$> ptr <*> ptr <*> pay <*> md
convertPayload bbPtr = do where
-- TODO: the reason this returns @Maybe ByteString@ is because the md = managed C.withMetadataArrayPtr
-- gRPC library calls the underlying out parameter pay = if havePay then managed C.withByteBufferPtr else return nullPtr
-- "optional_payload". I am not sure exactly in what cases it ptr :: forall a. Storable a => Managed (Ptr a)
-- won't be present. The C++ library checks a ptr = managed (bracket malloc free)
-- has_request_payload_ bool and passes in nullptr to dbug = grpcDebug . ("serverRequestCall(R): " ++)
-- request_registered_call if the bool is false, so we havePay = payloadHandling methodType /= C.SrmPayloadNone
-- may be able to do the payload present/absent check earlier. toBS p = peek p >>= \bb@(C.ByteBuffer rawPtr) ->
bb@(C.ByteBuffer rawPtr) <- peek bbPtr if | rawPtr == nullPtr -> return Nothing
if rawPtr == nullPtr | otherwise -> Just <$> C.copyByteBufferToByteString bb
then return Nothing
else do bs <- C.copyByteBufferToByteString bb
return $ Just bs
convertMeta requestMetadataRecv = do
mArray <- peek requestMetadataRecv
metamap <- C.getAllMetadataArray mArray
return metamap
-- | Register the server's completion queue. Must be done before the server is -- | Register the server's completion queue. Must be done before the server is
-- started. -- started.

12
src/Network/GRPC/LowLevel/GRPC.hs
View file

@ -1,23 +1,26 @@
{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE StandaloneDeriving #-}
{-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE StandaloneDeriving #-}
module Network.GRPC.LowLevel.GRPC where module Network.GRPC.LowLevel.GRPC where
import Control.Concurrent (threadDelay) import Control.Concurrent (threadDelay)
import Control.Exception import Control.Exception
import Data.String (IsString)
import qualified Data.ByteString as B import qualified Data.ByteString as B
import qualified Data.Map as M import qualified Data.Map as M
import Data.String (IsString)
import qualified Network.GRPC.Unsafe as C import qualified Network.GRPC.Unsafe as C
import qualified Network.GRPC.Unsafe.Op as C import qualified Network.GRPC.Unsafe.Op as C
#ifdef DEBUG #ifdef DEBUG
import GHC.Conc (myThreadId) import GHC.Conc (myThreadId)
#endif #endif
type MetadataMap = M.Map B.ByteString B.ByteString type MetadataMap = M.Map B.ByteString B.ByteString
newtype StatusDetails = StatusDetails B.ByteString deriving (Show, Eq, IsString) newtype StatusDetails = StatusDetails B.ByteString
deriving (Eq, IsString, Monoid, Show)
-- | Functions as a proof that the gRPC core has been started. The gRPC core -- | Functions as a proof that the gRPC core has been started. The gRPC core
-- must be initialized to create any gRPC state, so this is a requirement for -- must be initialized to create any gRPC state, so this is a requirement for
@ -44,6 +47,9 @@ data GRPCIOError = GRPCIOCallError C.CallError
-- reasonable amount of time. -- reasonable amount of time.
| GRPCIOUnknownError | GRPCIOUnknownError
| GRPCIOBadStatusCode C.StatusCode StatusDetails | GRPCIOBadStatusCode C.StatusCode StatusDetails
| GRPCIOInternalMissingExpectedPayload
| GRPCIOInternalUnexpectedRecv String -- debugging description
deriving (Show, Eq) deriving (Show, Eq)
throwIfCallError :: C.CallError -> Either GRPCIOError () throwIfCallError :: C.CallError -> Either GRPCIOError ()

118
src/Network/GRPC/LowLevel/Op.hs
View file

@ -1,9 +1,17 @@
{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ViewPatterns #-}
module Network.GRPC.LowLevel.Op where module Network.GRPC.LowLevel.Op where
import Control.Arrow
import Control.Exception import Control.Exception
import Control.Monad
import Control.Monad.Trans.Class (MonadTrans(lift))
import Control.Monad.Trans.Except
import Data.ByteString (ByteString)
import qualified Data.ByteString as B import qualified Data.ByteString as B
import qualified Data.Map.Strict as M import qualified Data.Map.Strict as M
import Data.Maybe (catMaybes) import Data.Maybe (catMaybes)
@ -20,6 +28,9 @@ import qualified Network.GRPC.Unsafe.ByteBuffer as C
import qualified Network.GRPC.Unsafe.Metadata as C import qualified Network.GRPC.Unsafe.Metadata as C
import qualified Network.GRPC.Unsafe.Op as C import qualified Network.GRPC.Unsafe.Op as C
import qualified Network.GRPC.Unsafe.Slice as C (Slice, freeSlice) import qualified Network.GRPC.Unsafe.Slice as C (Slice, freeSlice)
import Pipes ((>->))
import qualified Pipes as P
import qualified Pipes.Core as P
-- | Sum describing all possible send and receive operations that can be batched -- | Sum describing all possible send and receive operations that can be batched
-- and executed by gRPC. Usually these are processed in a handful of -- and executed by gRPC. Usually these are processed in a handful of
@ -144,7 +155,8 @@ withOpArrayAndCtxts ops = bracket setup teardown
data OpRecvResult = data OpRecvResult =
OpRecvInitialMetadataResult MetadataMap OpRecvInitialMetadataResult MetadataMap
| OpRecvMessageResult (Maybe B.ByteString) | OpRecvMessageResult (Maybe B.ByteString)
-- ^ If the client or server dies, we might not receive a response body, in -- ^ If a streaming call is in progress and the stream terminates normally,
-- or If the client or server dies, we might not receive a response body, in
-- which case this will be 'Nothing'. -- which case this will be 'Nothing'.
| OpRecvStatusOnClientResult MetadataMap C.StatusCode B.ByteString | OpRecvStatusOnClientResult MetadataMap C.StatusCode B.ByteString
| OpRecvCloseOnServerResult Bool -- ^ True if call was cancelled. | OpRecvCloseOnServerResult Bool -- ^ True if call was cancelled.
@ -202,7 +214,6 @@ resultFromOpContext _ = do
-- GRPC_CALL_ERROR_TOO_MANY_OPERATIONS error if we use the same 'Op' twice in -- GRPC_CALL_ERROR_TOO_MANY_OPERATIONS error if we use the same 'Op' twice in
-- the same batch, so we might want to change the list to a set. I don't think -- the same batch, so we might want to change the list to a set. I don't think
-- order matters within a batch. Need to check. -- order matters within a batch. Need to check.
runOps :: C.Call runOps :: C.Call
-- ^ 'Call' that this batch is associated with. One call can be -- ^ 'Call' that this batch is associated with. One call can be
-- associated with many batches. -- associated with many batches.
@ -232,6 +243,12 @@ runOps call cq ops =
fmap (Right . catMaybes) $ mapM resultFromOpContext contexts fmap (Right . catMaybes) $ mapM resultFromOpContext contexts
Left err -> return $ Left err Left err -> return $ Left err
runOps' :: C.Call
-> CompletionQueue
-> [Op]
-> ExceptT GRPCIOError IO [OpRecvResult]
runOps' c cq = ExceptT . runOps c cq
-- | If response status info is present in the given 'OpRecvResult's, returns -- | If response status info is present in the given 'OpRecvResult's, returns
-- a tuple of trailing metadata, status code, and status details. -- a tuple of trailing metadata, status code, and status details.
extractStatusInfo :: [OpRecvResult] extractStatusInfo :: [OpRecvResult]
@ -240,3 +257,100 @@ extractStatusInfo [] = Nothing
extractStatusInfo (OpRecvStatusOnClientResult meta code details:_) = extractStatusInfo (OpRecvStatusOnClientResult meta code details:_) =
Just (meta, code, details) Just (meta, code, details)
extractStatusInfo (_:xs) = extractStatusInfo xs extractStatusInfo (_:xs) = extractStatusInfo xs
--------------------------------------------------------------------------------
-- Types and helpers for common ops batches
type SendSingle a
= C.Call
-> CompletionQueue
-> a
-> ExceptT GRPCIOError IO ()
type RecvSingle a
= C.Call
-> CompletionQueue
-> ExceptT GRPCIOError IO a
sendSingle :: SendSingle Op
sendSingle c cq op = void (runOps' c cq [op])
sendInitialMetadata :: SendSingle MetadataMap
sendInitialMetadata c cq = sendSingle c cq . OpSendInitialMetadata
sendStatusFromServer :: SendSingle (MetadataMap, C.StatusCode, StatusDetails)
sendStatusFromServer c cq (md, st, ds) =
sendSingle c cq (OpSendStatusFromServer md st ds)
recvInitialMetadata :: RecvSingle MetadataMap
recvInitialMetadata c cq = runOps' c cq [OpRecvInitialMetadata] >>= \case
[OpRecvInitialMetadataResult md]
-> return md
_ -> throwE (GRPCIOInternalUnexpectedRecv "recvInitialMetadata")
recvStatusOnClient :: RecvSingle (MetadataMap, C.StatusCode, StatusDetails)
recvStatusOnClient c cq = runOps' c cq [OpRecvStatusOnClient] >>= \case
[OpRecvStatusOnClientResult md st ds]
-> return (md, st, StatusDetails ds)
_ -> throwE (GRPCIOInternalUnexpectedRecv "recvStatusOnClient")
--------------------------------------------------------------------------------
-- Streaming types and helpers
-- | Requests use Nothing to denote read, Just to denote
-- write. Right-constructed responses use Just to indicate a successful read,
-- and Nothing to denote end of stream when reading or a successful write.
type Streaming a =
P.Client (Maybe ByteString) (Either GRPCIOError (Maybe ByteString)) IO a
-- | Run the given 'Streaming' operation via an appropriate upstream
-- proxy. I.e., if called on the client side, the given 'Streaming' operation
-- talks to a server proxy, and vice versa.
runStreamingProxy :: String
-- ^ context string for including in errors
-> C.Call
-- ^ the call associated with this streaming operation
-> CompletionQueue
-- ^ the completion queue for ops batches
-> Streaming a
-- ^ the requesting side of the streaming operation
-> ExceptT GRPCIOError IO a
runStreamingProxy nm c cq
= ExceptT . P.runEffect . (streamingProxy nm c cq P.+>>) . fmap Right
streamingProxy :: String
-- ^ context string for including in errors
-> C.Call
-- ^ the call associated with this streaming operation
-> CompletionQueue
-- ^ the completion queue for ops batches
-> Maybe ByteString
-- ^ the request to the proxy
-> P.Server
(Maybe ByteString)
(Either GRPCIOError (Maybe ByteString))
IO (Either GRPCIOError a)
streamingProxy nm c cq = maybe recv send
where
recv = run [OpRecvMessage] >>= \case
RecvMsgRslt mr -> rsp mr >>= streamingProxy nm c cq
Right{} -> err (urecv "recv")
Left e -> err e
send msg = run [OpSendMessage msg] >>= \case
Right [] -> rsp Nothing >>= streamingProxy nm c cq
Right _ -> err (urecv "send")
Left e -> err e
err e = P.respond (Left e) >> return (Left e)
rsp = P.respond . Right
run = lift . runOps c cq
urecv = GRPCIOInternalUnexpectedRecv . (nm ++)
type StreamRecv = Streaming (Either GRPCIOError (Maybe ByteString))
streamRecv :: StreamRecv
streamRecv = P.request Nothing
type StreamSend = ByteString -> Streaming (Either GRPCIOError ())
streamSend :: StreamSend
streamSend = fmap void . P.request . Just
pattern RecvMsgRslt mmsg <- Right [OpRecvMessageResult mmsg]

215
src/Network/GRPC/LowLevel/Server.hs
View file

@ -1,12 +1,22 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PatternSynonyms #-}
{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE ViewPatterns #-}
-- | This module defines data structures and operations pertaining to registered -- | This module defines data structures and operations pertaining to registered
-- servers using registered calls; for unregistered support, see -- servers using registered calls; for unregistered support, see
-- `Network.GRPC.LowLevel.Server.Unregistered`. -- `Network.GRPC.LowLevel.Server.Unregistered`.
module Network.GRPC.LowLevel.Server where module Network.GRPC.LowLevel.Server where
import Control.Arrow
import Control.Exception (bracket, finally) import Control.Exception (bracket, finally)
import Control.Monad import Control.Monad
import Control.Monad.Trans.Class (MonadTrans (lift))
import Control.Monad.Trans.Except
import Data.ByteString (ByteString) import Data.ByteString (ByteString)
import Network.GRPC.LowLevel.Call import Network.GRPC.LowLevel.Call
import Network.GRPC.LowLevel.CompletionQueue (CompletionQueue, import Network.GRPC.LowLevel.CompletionQueue (CompletionQueue,
@ -19,14 +29,19 @@ import Network.GRPC.LowLevel.CompletionQueue (CompletionQueue,
import Network.GRPC.LowLevel.GRPC import Network.GRPC.LowLevel.GRPC
import Network.GRPC.LowLevel.Op import Network.GRPC.LowLevel.Op
import qualified Network.GRPC.Unsafe as C import qualified Network.GRPC.Unsafe as C
import qualified Network.GRPC.Unsafe.Op as C
import qualified Network.GRPC.Unsafe.ChannelArgs as C import qualified Network.GRPC.Unsafe.ChannelArgs as C
import qualified Network.GRPC.Unsafe.Op as C
import qualified Pipes as P
import qualified Pipes.Core as P
-- | Wraps various gRPC state needed to run a server. -- | Wraps various gRPC state needed to run a server.
data Server = Server data Server = Server
{ internalServer :: C.Server { internalServer :: C.Server
, serverCQ :: CompletionQueue , serverCQ :: CompletionQueue
, registeredMethods :: [RegisteredMethod] , normalMethods :: [RegisteredMethod 'Normal]
, sstreamingMethods :: [RegisteredMethod 'ServerStreaming]
, cstreamingMethods :: [RegisteredMethod 'ClientStreaming]
, bidiStreamingMethods :: [RegisteredMethod 'BiDiStreaming]
, serverConfig :: ServerConfig , serverConfig :: ServerConfig
} }
@ -39,13 +54,10 @@ data ServerConfig = ServerConfig
-- ^ Port on which to listen for requests. -- ^ Port on which to listen for requests.
, methodsToRegister :: [(MethodName, GRPCMethodType)] , methodsToRegister :: [(MethodName, GRPCMethodType)]
-- ^ List of (method name, method type) tuples specifying all methods to -- ^ List of (method name, method type) tuples specifying all methods to
-- register. You can also handle other unregistered methods with -- register.
-- `serverHandleNormalCall`.
, serverArgs :: [C.Arg] , serverArgs :: [C.Arg]
-- ^ Optional arguments for setting up the -- ^ Optional arguments for setting up the channel on the server. Supplying an
-- channel on the server. Supplying an empty -- empty list will cause the channel to use gRPC's default options.
-- list will cause the channel to use gRPC's
-- default options.
} }
deriving (Show, Eq) deriving (Show, Eq)
@ -62,34 +74,46 @@ startServer grpc conf@ServerConfig{..} =
error $ "Unable to bind port: " ++ show port error $ "Unable to bind port: " ++ show port
cq <- createCompletionQueue grpc cq <- createCompletionQueue grpc
serverRegisterCompletionQueue server cq serverRegisterCompletionQueue server cq
methods <- forM methodsToRegister $ \(name, mtype) ->
serverRegisterMethod server name e mtype -- Register methods according to their GRPCMethodType kind. It's a bit ugly
-- to partition them this way, but we get very convenient phantom typing
-- elsewhere by doing so.
(ns, ss, cs, bs) <- do
let f (ns, ss, cs, bs) (nm, mt) = do
let reg = serverRegisterMethod server nm e mt
case mt of
Normal -> ( , ss, cs, bs) . (:ns) <$> reg
ServerStreaming -> (ns, , cs, bs) . (:ss) <$> reg
ClientStreaming -> (ns, ss, , bs) . (:cs) <$> reg
BiDiStreaming -> (ns, ss, cs, ) . (:bs) <$> reg
foldM f ([],[],[],[]) methodsToRegister
C.grpcServerStart server C.grpcServerStart server
return $ Server server cq methods conf return $ Server server cq ns ss cs bs conf
stopServer :: Server -> IO () stopServer :: Server -> IO ()
-- TODO: Do method handles need to be freed? -- TODO: Do method handles need to be freed?
stopServer (Server server cq _ _) = do stopServer Server{..} = do
grpcDebug "stopServer: calling shutdownNotify." grpcDebug "stopServer: calling shutdownNotify."
shutdownNotify shutdownNotify
grpcDebug "stopServer: cancelling all calls." grpcDebug "stopServer: cancelling all calls."
C.grpcServerCancelAllCalls server C.grpcServerCancelAllCalls internalServer
grpcDebug "stopServer: call grpc_server_destroy." grpcDebug "stopServer: call grpc_server_destroy."
C.grpcServerDestroy server C.grpcServerDestroy internalServer
grpcDebug "stopServer: shutting down CQ." grpcDebug "stopServer: shutting down CQ."
shutdownCQ shutdownCQ
where shutdownCQ = do where shutdownCQ = do
shutdownResult <- shutdownCompletionQueue cq shutdownResult <- shutdownCompletionQueue serverCQ
case shutdownResult of case shutdownResult of
Left _ -> do putStrLn "Warning: completion queue didn't shut down." Left _ -> do putStrLn "Warning: completion queue didn't shut down."
putStrLn "Trying to stop server anyway." putStrLn "Trying to stop server anyway."
Right _ -> return () Right _ -> return ()
shutdownNotify = do shutdownNotify = do
let shutdownTag = C.tag 0 let shutdownTag = C.tag 0
serverShutdownAndNotify server cq shutdownTag serverShutdownAndNotify internalServer serverCQ shutdownTag
grpcDebug "called serverShutdownAndNotify; plucking." grpcDebug "called serverShutdownAndNotify; plucking."
shutdownEvent <- pluck cq shutdownTag (Just 30) shutdownEvent <- pluck serverCQ shutdownTag (Just 30)
grpcDebug $ "shutdownNotify: got shutdown event" ++ show shutdownEvent grpcDebug $ "shutdownNotify: got shutdown event" ++ show shutdownEvent
case shutdownEvent of case shutdownEvent of
-- This case occurs when we pluck but the queue is already in the -- This case occurs when we pluck but the queue is already in the
@ -100,7 +124,7 @@ stopServer (Server server cq _ _) = do
-- Uses 'bracket' to safely start and stop a server, even if exceptions occur. -- Uses 'bracket' to safely start and stop a server, even if exceptions occur.
withServer :: GRPC -> ServerConfig -> (Server -> IO a) -> IO a withServer :: GRPC -> ServerConfig -> (Server -> IO a) -> IO a
withServer grpc cfg f = bracket (startServer grpc cfg) stopServer f withServer grpc cfg = bracket (startServer grpc cfg) stopServer
-- | Register a method on a server. The 'RegisteredMethod' type can then be used -- | Register a method on a server. The 'RegisteredMethod' type can then be used
-- to wait for a request to arrive. Note: gRPC claims this must be called before -- to wait for a request to arrive. Note: gRPC claims this must be called before
@ -118,25 +142,23 @@ serverRegisterMethod :: C.Server
-> GRPCMethodType -> GRPCMethodType
-- ^ Type of method this will be. In the future, this will -- ^ Type of method this will be. In the future, this will
-- be used to switch to the correct handling logic. -- be used to switch to the correct handling logic.
-- Currently, the only valid choice is 'Normal'. -> IO (RegisteredMethod mt)
-> IO RegisteredMethod serverRegisterMethod internalServer meth e mty =
serverRegisterMethod internalServer meth e Normal = do RegisteredMethod mty meth e <$> do
handle <- C.grpcServerRegisterMethod internalServer h <- C.grpcServerRegisterMethod internalServer
(unMethodName meth) (unEndpoint e) (unMethodName meth) (unEndpoint e) (payloadHandling mty)
grpcDebug $ "registered method to handle " ++ show handle grpcDebug $ "registered method handle: " ++ show h ++ " of type " ++ show mty
return $ RegisteredMethod Normal meth e handle return h
serverRegisterMethod _ _ _ _ = error "Streaming methods not implemented yet."
-- | Create a 'Call' with which to wait for the invocation of a registered -- | Create a 'Call' with which to wait for the invocation of a registered
-- method. -- method.
serverCreateCall :: Server serverCreateCall :: Server
-> RegisteredMethod -> RegisteredMethod mt
-> IO (Either GRPCIOError ServerCall) -> IO (Either GRPCIOError ServerCall)
serverCreateCall Server{..} rm = serverCreateCall Server{..} = serverRequestCall internalServer serverCQ
serverRequestCall internalServer serverCQ rm
withServerCall :: Server withServerCall :: Server
-> RegisteredMethod -> RegisteredMethod mt
-> (ServerCall -> IO (Either GRPCIOError a)) -> (ServerCall -> IO (Either GRPCIOError a))
-> IO (Either GRPCIOError a) -> IO (Either GRPCIOError a)
withServerCall server regmethod f = do withServerCall server regmethod f = do
@ -147,61 +169,112 @@ withServerCall server regmethod f = do
where logDestroy c = grpcDebug "withServerRegisteredCall: destroying." where logDestroy c = grpcDebug "withServerRegisteredCall: destroying."
>> destroyServerCall c >> destroyServerCall c
serverOpsSendNormalRegisteredResponse :: ByteString --------------------------------------------------------------------------------
-> MetadataMap -- serverReader (server side of client streaming mode)
-- ^ initial metadata
-> MetadataMap
-- ^ trailing metadata
-> C.StatusCode
-> StatusDetails
-> [Op]
serverOpsSendNormalRegisteredResponse
body initMetadata trailingMeta code details =
[OpSendInitialMetadata initMetadata,
OpRecvCloseOnServer,
OpSendMessage body,
OpSendStatusFromServer trailingMeta code details]
-- | A handler for an registered server call; bytestring parameter is request type ServerReaderHandler
= ServerCall
-> StreamRecv
-> Streaming (Maybe ByteString, MetadataMap, C.StatusCode, StatusDetails)
serverReader :: Server
-> RegisteredMethod 'ClientStreaming
-> MetadataMap -- ^ initial server metadata
-> ServerReaderHandler
-> IO (Either GRPCIOError ())
serverReader s@Server{ serverCQ = cq } rm initMeta f = withServerCall s rm go
where
go sc@(unServerCall -> c) = runExceptT $ do
lift $ debugServerCall sc
(mmsg, trailMD, st, ds) <-
runStreamingProxy "serverReader" c cq (f sc streamRecv)
runOps' c cq ( OpSendInitialMetadata initMeta
: OpSendStatusFromServer trailMD st ds
: maybe [] ((:[]) . OpSendMessage) mmsg
)
return ()
--------------------------------------------------------------------------------
-- serverWriter (server side of server streaming mode)
type ServerWriterHandler
= ServerCall
-> StreamSend
-> Streaming (MetadataMap, C.StatusCode, StatusDetails)
-- | Wait for and then handle a registered, server-streaming call.
serverWriter :: Server
-> RegisteredMethod 'ServerStreaming
-> MetadataMap
-- ^ Initial server metadata
-> ServerWriterHandler
-> IO (Either GRPCIOError ())
serverWriter s@Server{ serverCQ = cq } rm initMeta f = withServerCall s rm go
where
go sc@ServerCall{ unServerCall = c } = runExceptT $ do
lift (debugServerCall sc)
sendInitialMetadata c cq initMeta
st <- runStreamingProxy "serverWriter" c cq (f sc streamSend)
sendStatusFromServer c cq st
--------------------------------------------------------------------------------
-- serverRW (server side of bidirectional streaming mode)
type ServerRWHandler
= ServerCall
-> StreamRecv
-> StreamSend
-> Streaming (MetadataMap, C.StatusCode, StatusDetails)
serverRW :: Server
-> RegisteredMethod 'BiDiStreaming
-> MetadataMap
-- ^ initial server metadata
-> ServerRWHandler
-> IO (Either GRPCIOError ())
serverRW s@Server{ serverCQ = cq } rm initMeta f = withServerCall s rm go
where
go sc@(unServerCall -> c) = runExceptT $ do
lift $ debugServerCall sc
sendInitialMetadata c cq initMeta
st <- runStreamingProxy "serverRW" c cq (f sc streamRecv streamSend)
sendStatusFromServer c cq st
--------------------------------------------------------------------------------
-- serverHandleNormalCall (server side of normal request/response)
-- | A handler for a registered server call; bytestring parameter is request
-- body, with the bytestring response body in the result tuple. The first -- body, with the bytestring response body in the result tuple. The first
-- metadata parameter refers to the request metadata, with the two metadata -- metadata parameter refers to the request metadata, with the two metadata
-- values in the result tuple being the initial and trailing metadata -- values in the result tuple being the initial and trailing metadata
-- respectively. We pass in the 'ServerCall' so that the server can call -- respectively. We pass in the 'ServerCall' so that the server can call
-- 'serverCallCancel' on it if needed. -- 'serverCallCancel' on it if needed.
-- TODO: make a more rigid type for this with a Maybe MetadataMap for the
-- trailing meta, and use it for both kinds of call handlers.
type ServerHandler type ServerHandler
= ServerCall -> ByteString -> MetadataMap = ServerCall -> ByteString -> MetadataMap
-> IO (ByteString, MetadataMap, C.StatusCode, StatusDetails) -> IO (ByteString, MetadataMap, C.StatusCode, StatusDetails)
-- TODO: we will want to replace this with some more general concept that also
-- works with streaming calls in the future.
-- | Wait for and then handle a normal (non-streaming) call. -- | Wait for and then handle a normal (non-streaming) call.
serverHandleNormalCall :: Server serverHandleNormalCall :: Server
-> RegisteredMethod -> RegisteredMethod 'Normal
-> MetadataMap -> MetadataMap
-- ^ Initial server metadata -- ^ Initial server metadata
-> ServerHandler -> ServerHandler
-> IO (Either GRPCIOError ()) -> IO (Either GRPCIOError ())
serverHandleNormalCall s@Server{..} rm initMeta f = do serverHandleNormalCall s@Server{ serverCQ = cq } rm initMeta f =
withServerCall s rm $ \call -> do withServerCall s rm go
where
go sc@(unServerCall -> call) = do
grpcDebug "serverHandleNormalCall(R): starting batch." grpcDebug "serverHandleNormalCall(R): starting batch."
debugServerCall call debugServerCall sc
let payload = optionalPayload call case optionalPayload sc of
case payload of Nothing -> return (Left GRPCIOInternalMissingExpectedPayload)
--TODO: what should we do with an empty payload? Have the handler take Just pay -> do
-- @Maybe ByteString@? Need to figure out when/why payload would be empty. (rspBody, trailMeta, status, ds) <- f sc pay (requestMetadataRecv sc)
Nothing -> error "serverHandleNormalCall(R): payload empty." eea <- runOps call cq
Just requestBody -> do [ OpSendInitialMetadata initMeta
let requestMeta = requestMetadataRecv call , OpRecvCloseOnServer
(respBody, trailingMeta, status, details) <- f call , OpSendMessage rspBody
requestBody , OpSendStatusFromServer trailMeta status ds
requestMeta ]
let respOps = serverOpsSendNormalRegisteredResponse <* grpcDebug "serverHandleNormalCall(R): finished response ops."
respBody initMeta trailingMeta status details return (void eea)
respOpsResults <- runOps (unServerCall call) serverCQ respOps
grpcDebug "serverHandleNormalCall(R): finished response ops."
case respOpsResults of
Left x -> return $ Left x
Right _ -> return $ Right ()

37
src/Network/GRPC/LowLevel/Server/Unregistered.hs
View file

@ -1,3 +1,4 @@
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE RecordWildCards #-}
module Network.GRPC.LowLevel.Server.Unregistered where module Network.GRPC.LowLevel.Server.Unregistered where
@ -7,7 +8,8 @@ import Data.ByteString (ByteString)
import Network.GRPC.LowLevel.Call.Unregistered import Network.GRPC.LowLevel.Call.Unregistered
import Network.GRPC.LowLevel.CompletionQueue.Unregistered (serverRequestCall) import Network.GRPC.LowLevel.CompletionQueue.Unregistered (serverRequestCall)
import Network.GRPC.LowLevel.GRPC import Network.GRPC.LowLevel.GRPC
import Network.GRPC.LowLevel.Op (Op(..), OpRecvResult (..), runOps) import Network.GRPC.LowLevel.Op (Op (..), OpRecvResult (..),
runOps)
import Network.GRPC.LowLevel.Server (Server (..)) import Network.GRPC.LowLevel.Server (Server (..))
import qualified Network.GRPC.Unsafe.Op as C import qualified Network.GRPC.Unsafe.Op as C
@ -57,24 +59,31 @@ serverHandleNormalCall :: Server
-> MetadataMap -- ^ Initial server metadata. -> MetadataMap -- ^ Initial server metadata.
-> ServerHandler -> ServerHandler
-> IO (Either GRPCIOError ()) -> IO (Either GRPCIOError ())
serverHandleNormalCall s@Server{..} srvMetadata f = do serverHandleNormalCall s@Server{..} srvMetadata f =
withServerCall s $ \call@ServerCall{..} -> do withServerCall s $ \call@ServerCall{..} -> do
grpcDebug "serverHandleNormalCall(U): starting batch." grpcDebug "serverHandleNormalCall(U): starting batch."
let recvOps = serverOpsGetNormalCall srvMetadata runOps unServerCall serverCQ
opResults <- runOps unServerCall serverCQ recvOps [ OpSendInitialMetadata srvMetadata
case opResults of , OpRecvMessage
Left x -> do grpcDebug "serverHandleNormalCall(U): ops failed; aborting" ]
>>= \case
Left x -> do
grpcDebug "serverHandleNormalCall(U): ops failed; aborting"
return $ Left x return $ Left x
Right [OpRecvMessageResult (Just body)] -> do Right [OpRecvMessageResult (Just body)] -> do
grpcDebug $ "got client metadata: " ++ show requestMetadataRecv grpcDebug $ "got client metadata: " ++ show requestMetadataRecv
grpcDebug $ "call_details host is: " ++ show callHost grpcDebug $ "call_details host is: " ++ show callHost
(respBody, respMetadata, status, details) <- f call body (rspBody, rspMeta, status, ds) <- f call body
let respOps = serverOpsSendNormalResponse runOps unServerCall serverCQ
respBody respMetadata status details [ OpRecvCloseOnServer
respOpsResults <- runOps unServerCall serverCQ respOps , OpSendMessage rspBody,
case respOpsResults of OpSendStatusFromServer rspMeta status ds
Left x -> do grpcDebug "serverHandleNormalCall(U): resp failed." ]
>>= \case
Left x -> do
grpcDebug "serverHandleNormalCall(U): resp failed."
return $ Left x return $ Left x
Right _ -> grpcDebug "serverHandleNormalCall(U): ops done." Right _ -> do
>> return (Right ()) grpcDebug "serverHandleNormalCall(U): ops done."
return $ Right ()
x -> error $ "impossible pattern match: " ++ show x x -> error $ "impossible pattern match: " ++ show x

4
src/Network/GRPC/Unsafe.chs
View file

@ -61,6 +61,8 @@ instance Storable Call where
peek p = fmap Call (peek (castPtr p)) peek p = fmap Call (peek (castPtr p))
poke p (Call r) = poke (castPtr p) r poke p (Call r) = poke (castPtr p) r
{#enum grpc_server_register_method_payload_handling as ServerRegisterMethodPayloadHandling {underscoreToCase} deriving (Eq, Show)#}
-- | A 'Tag' is an identifier that is used with a 'CompletionQueue' to signal -- | A 'Tag' is an identifier that is used with a 'CompletionQueue' to signal
-- that the corresponding operation has completed. -- that the corresponding operation has completed.
newtype Tag = Tag {unTag :: Ptr ()} deriving (Show, Eq) newtype Tag = Tag {unTag :: Ptr ()} deriving (Show, Eq)
@ -235,7 +237,7 @@ getPeerPeek cstr = do
{`GrpcChannelArgs',unReserved `Reserved'} -> `Server'#} {`GrpcChannelArgs',unReserved `Reserved'} -> `Server'#}
{#fun grpc_server_register_method_ as ^ {#fun grpc_server_register_method_ as ^
{`Server', `String', `String'} -> `CallHandle' CallHandle#} {`Server', `String', `String', `ServerRegisterMethodPayloadHandling'} -> `CallHandle' CallHandle#}
{#fun grpc_server_register_completion_queue as ^ {#fun grpc_server_register_completion_queue as ^
{`Server', `CompletionQueue', unReserved `Reserved'} -> `()'#} {`Server', `CompletionQueue', unReserved `Reserved'} -> `()'#}

2
stack.yaml
View file

@ -8,7 +8,7 @@ resolver: lts-5.10
packages: packages:
- '.' - '.'
# Packages to be pulled from upstream that are not in the resolver (e.g., acme-missiles-0.3) # Packages to be pulled from upstream that are not in the resolver (e.g., acme-missiles-0.3)
extra-deps: [] extra-deps: [managed-1.0.5]
# Override default flag values for local packages and extra-deps # Override default flag values for local packages and extra-deps
flags: {} flags: {}

219
tests/LowLevelTests.hs
View file

@ -1,13 +1,17 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE LambdaCase #-} {-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedLists #-} {-# LANGUAGE OverloadedLists #-}
{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE ViewPatterns #-}
module LowLevelTests where module LowLevelTests where
import Control.Concurrent (threadDelay) import Control.Concurrent (threadDelay)
import Control.Concurrent.Async import Control.Concurrent.Async
import Control.Monad import Control.Monad
import Control.Monad.Managed
import Data.ByteString (ByteString, import Data.ByteString (ByteString,
isPrefixOf, isPrefixOf,
isSuffixOf) isSuffixOf)
@ -16,6 +20,8 @@ import Network.GRPC.LowLevel
import qualified Network.GRPC.LowLevel.Call.Unregistered as U import qualified Network.GRPC.LowLevel.Call.Unregistered as U
import qualified Network.GRPC.LowLevel.Client.Unregistered as U import qualified Network.GRPC.LowLevel.Client.Unregistered as U
import qualified Network.GRPC.LowLevel.Server.Unregistered as U import qualified Network.GRPC.LowLevel.Server.Unregistered as U
import Pipes ((>->))
import qualified Pipes as P
import Test.Tasty import Test.Tasty
import Test.Tasty.HUnit as HU (Assertion, import Test.Tasty.HUnit as HU (Assertion,
assertBool, assertBool,
@ -42,27 +48,25 @@ lowLevelTests = testGroup "Unit tests of low-level Haskell library"
, testCustomUserAgent , testCustomUserAgent
, testClientCompression , testClientCompression
, testClientServerCompression , testClientServerCompression
, testClientStreaming
, testServerStreaming
, testBiDiStreaming
] ]
testGRPCBracket :: TestTree testGRPCBracket :: TestTree
testGRPCBracket = testGRPCBracket =
testCase "Start/stop GRPC" $ withGRPC nop testCase "Start/stop GRPC" $ runManaged $ void mgdGRPC
testCompletionQueueCreateDestroy :: TestTree testCompletionQueueCreateDestroy :: TestTree
testCompletionQueueCreateDestroy = testCompletionQueueCreateDestroy =
testCase "Create/destroy CQ" $ withGRPC $ \g -> testCase "Create/destroy CQ" $ runManaged $ do
withCompletionQueue g nop g <- mgdGRPC
liftIO (withCompletionQueue g nop)
testClientCreateDestroy :: TestTree testClientCreateDestroy :: TestTree
testClientCreateDestroy = testClientCreateDestroy =
clientOnlyTest "start/stop" nop clientOnlyTest "start/stop" nop
testClientCall :: TestTree
testClientCall =
clientOnlyTest "create/destroy call" $ \c -> do
r <- U.withClientCall c "/foo" 10 $ const $ return $ Right ()
r @?= Right ()
testClientTimeoutNoServer :: TestTree testClientTimeoutNoServer :: TestTree
testClientTimeoutNoServer = testClientTimeoutNoServer =
clientOnlyTest "request timeout when server DNE" $ \c -> do clientOnlyTest "request timeout when server DNE" $ \c -> do
@ -97,13 +101,13 @@ testMixRegisteredUnregistered =
concurrently regThread unregThread concurrently regThread unregThread
return () return ()
where regThread = do where regThread = do
let rm = head (registeredMethods s) let rm = head (normalMethods s)
r <- serverHandleNormalCall s rm dummyMeta $ \_ body _ -> do r <- serverHandleNormalCall s rm dummyMeta $ \_ body _ -> do
body @?= "Hello" body @?= "Hello"
return ("reply test", dummyMeta, StatusOk, StatusDetails "") return ("reply test", dummyMeta, StatusOk, "")
return () return ()
unregThread = do unregThread = do
r1 <- U.serverHandleNormalCall s mempty $ \call _ -> do U.serverHandleNormalCall s mempty $ \call _ -> do
U.callMethod call @?= "/bar" U.callMethod call @?= "/bar"
return ("", mempty, StatusOk, return ("", mempty, StatusOk,
StatusDetails "Wrong endpoint") StatusDetails "Wrong endpoint")
@ -130,13 +134,11 @@ testPayload =
initMD @?= dummyMeta initMD @?= dummyMeta
trailMD @?= dummyMeta trailMD @?= dummyMeta
server s = do server s = do
length (registeredMethods s) @?= 1 let rm = head (normalMethods s)
let rm = head (registeredMethods s)
r <- serverHandleNormalCall s rm dummyMeta $ \_ reqBody reqMD -> do r <- serverHandleNormalCall s rm dummyMeta $ \_ reqBody reqMD -> do
reqBody @?= "Hello!" reqBody @?= "Hello!"
checkMD "Server metadata mismatch" clientMD reqMD checkMD "Server metadata mismatch" clientMD reqMD
return ("reply test", dummyMeta, StatusOk, return ("reply test", dummyMeta, StatusOk, "details string")
StatusDetails "details string")
r @?= Right () r @?= Right ()
testServerCancel :: TestTree testServerCancel :: TestTree
@ -146,21 +148,130 @@ testServerCancel =
client c = do client c = do
rm <- clientRegisterMethod c "/foo" Normal rm <- clientRegisterMethod c "/foo" Normal
res <- clientRequest c rm 10 "" mempty res <- clientRequest c rm 10 "" mempty
res @?= Left (GRPCIOBadStatusCode StatusCancelled res @?= badStatus StatusCancelled
(StatusDetails
"Received RST_STREAM err=8"))
server s = do server s = do
let rm = head (registeredMethods s) let rm = head (normalMethods s)
r <- serverHandleNormalCall s rm mempty $ \c _ _ -> do r <- serverHandleNormalCall s rm mempty $ \c _ _ -> do
serverCallCancel c StatusCancelled "" serverCallCancel c StatusCancelled ""
return (mempty, mempty, StatusCancelled, "") return (mempty, mempty, StatusCancelled, "")
r @?= Right () r @?= Right ()
testServerStreaming :: TestTree
testServerStreaming =
csTest "server streaming" client server [("/feed", ServerStreaming)]
where
clientInitMD = [("client","initmd")]
serverInitMD = [("server","initmd")]
clientPay = "FEED ME!"
pays = ["ONE", "TWO", "THREE", "FOUR"] :: [ByteString]
client c = do
rm <- clientRegisterMethod c "/feed" ServerStreaming
eea <- clientReader c rm 10 clientPay clientInitMD $ \initMD recv -> do
liftIO $ checkMD "Server initial metadata mismatch" serverInitMD initMD
forM_ pays $ \p -> recv `is` Right (Just p)
recv `is` Right Nothing
eea @?= Right (dummyMeta, StatusOk, "dtls")
server s = do
let rm = head (sstreamingMethods s)
eea <- serverWriter s rm serverInitMD $ \sc send -> do
liftIO $ do
checkMD "Client request metadata mismatch"
clientInitMD (requestMetadataRecv sc)
case optionalPayload sc of
Nothing -> assertFailure "expected optional payload"
Just pay -> pay @?= clientPay
forM_ pays $ \p -> send p `is` Right ()
return (dummyMeta, StatusOk, "dtls")
eea @?= Right ()
testClientStreaming :: TestTree
testClientStreaming =
csTest "client streaming" client server [("/slurp", ClientStreaming)]
where
clientInitMD = [("a","b")]
serverInitMD = [("x","y")]
trailMD = dummyMeta
serverRsp = "serverReader reply"
serverDtls = "deets"
serverStatus = StatusOk
pays = ["P_ONE", "P_TWO", "P_THREE"] :: [ByteString]
client c = do
rm <- clientRegisterMethod c "/slurp" ClientStreaming
eea <- clientWriter c rm 10 clientInitMD $ \send -> do
-- liftIO $ checkMD "Server initial metadata mismatch" serverInitMD initMD
forM_ pays $ \p -> send p `is` Right ()
eea @?= Right (Just serverRsp, serverInitMD, trailMD, serverStatus, serverDtls)
server s = do
let rm = head (cstreamingMethods s)
eea <- serverReader s rm serverInitMD $ \sc recv -> do
liftIO $ checkMD "Client request metadata mismatch"
clientInitMD (requestMetadataRecv sc)
forM_ pays $ \p -> recv `is` Right (Just p)
recv `is` Right Nothing
return (Just serverRsp, trailMD, serverStatus, serverDtls)
eea @?= Right ()
testBiDiStreaming :: TestTree
testBiDiStreaming =
csTest "bidirectional streaming" client server [("/bidi", BiDiStreaming)]
where
clientInitMD = [("bidi-streaming","client")]
serverInitMD = [("bidi-streaming","server")]
trailMD = dummyMeta
serverStatus = StatusOk
serverDtls = "deets"
client c = do
rm <- clientRegisterMethod c "/bidi" BiDiStreaming
eea <- clientRW c rm 10 clientInitMD $ \initMD recv send -> do
liftIO $ checkMD "Server initial metadata mismatch"
serverInitMD initMD
send "cw0" `is` Right ()
recv `is` Right (Just "sw0")
send "cw1" `is` Right ()
recv `is` Right (Just "sw1")
recv `is` Right (Just "sw2")
return ()
eea @?= Right (trailMD, serverStatus, serverDtls)
server s = do
let rm = head (bidiStreamingMethods s)
eea <- serverRW s rm serverInitMD $ \sc recv send -> do
liftIO $ checkMD "Client request metadata mismatch"
clientInitMD (requestMetadataRecv sc)
recv `is` Right (Just "cw0")
send "sw0" `is` Right ()
recv `is` Right (Just "cw1")
send "sw1" `is` Right ()
send "sw2" `is` Right ()
recv `is` Right Nothing
return (trailMD, serverStatus, serverDtls)
eea @?= Right ()
--------------------------------------------------------------------------------
-- Unregistered tests
testClientCall :: TestTree
testClientCall =
clientOnlyTest "create/destroy call" $ \c -> do
r <- U.withClientCall c "/foo" 10 $ const $ return $ Right ()
r @?= Right ()
testServerCall :: TestTree
testServerCall =
serverOnlyTest "create/destroy call" [] $ \s -> do
r <- U.withServerCall s $ const $ return $ Right ()
r @?= Left GRPCIOTimeout
testPayloadUnregistered :: TestTree testPayloadUnregistered :: TestTree
testPayloadUnregistered = testPayloadUnregistered =
csTest "unregistered normal request/response" client server [] csTest "unregistered normal request/response" client server []
where where
client c = do client c =
U.clientRequest c "/foo" 10 "Hello!" mempty >>= do U.clientRequest c "/foo" 10 "Hello!" mempty >>= do
checkReqRslt $ \NormalRequestResult{..} -> do checkReqRslt $ \NormalRequestResult{..} -> do
rspCode @?= StatusOk rspCode @?= StatusOk
@ -186,13 +297,13 @@ testGoaway =
clientRequest c rm 10 "" mempty clientRequest c rm 10 "" mempty
lastResult <- clientRequest c rm 1 "" mempty lastResult <- clientRequest c rm 1 "" mempty
assertBool "Client handles server shutdown gracefully" $ assertBool "Client handles server shutdown gracefully" $
lastResult == unavailableStatus lastResult == badStatus StatusUnavailable
|| ||
lastResult == deadlineExceededStatus lastResult == badStatus StatusDeadlineExceeded
|| ||
lastResult == Left GRPCIOTimeout lastResult == Left GRPCIOTimeout
server s = do server s = do
let rm = head (registeredMethods s) let rm = head (normalMethods s)
serverHandleNormalCall s rm mempty dummyHandler serverHandleNormalCall s rm mempty dummyHandler
serverHandleNormalCall s rm mempty dummyHandler serverHandleNormalCall s rm mempty dummyHandler
return () return ()
@ -204,9 +315,9 @@ testSlowServer =
client c = do client c = do
rm <- clientRegisterMethod c "/foo" Normal rm <- clientRegisterMethod c "/foo" Normal
result <- clientRequest c rm 1 "" mempty result <- clientRequest c rm 1 "" mempty
result @?= deadlineExceededStatus result @?= badStatus StatusDeadlineExceeded
server s = do server s = do
let rm = head (registeredMethods s) let rm = head (normalMethods s)
serverHandleNormalCall s rm mempty $ \_ _ _ -> do serverHandleNormalCall s rm mempty $ \_ _ _ -> do
threadDelay (2*10^(6 :: Int)) threadDelay (2*10^(6 :: Int))
return dummyResp return dummyResp
@ -221,7 +332,7 @@ testServerCallExpirationCheck =
result <- clientRequest c rm 3 "" mempty result <- clientRequest c rm 3 "" mempty
return () return ()
server s = do server s = do
let rm = head (registeredMethods s) let rm = head (normalMethods s)
serverHandleNormalCall s rm mempty $ \c _ _ -> do serverHandleNormalCall s rm mempty $ \c _ _ -> do
exp1 <- serverCallIsExpired c exp1 <- serverCallIsExpired c
assertBool "Call isn't expired when handler starts" $ not exp1 assertBool "Call isn't expired when handler starts" $ not exp1
@ -245,7 +356,7 @@ testCustomUserAgent =
result <- clientRequest c rm 4 "" mempty result <- clientRequest c rm 4 "" mempty
return () return ()
server = TestServer (stdServerConf [("/foo", Normal)]) $ \s -> do server = TestServer (stdServerConf [("/foo", Normal)]) $ \s -> do
let rm = head (registeredMethods s) let rm = head (normalMethods s)
serverHandleNormalCall s rm mempty $ \_ _ meta -> do serverHandleNormalCall s rm mempty $ \_ _ meta -> do
let ua = meta M.! "user-agent" let ua = meta M.! "user-agent"
assertBool "User agent prefix is present" $ isPrefixOf "prefix!" ua assertBool "User agent prefix is present" $ isPrefixOf "prefix!" ua
@ -266,8 +377,8 @@ testClientCompression =
result <- clientRequest c rm 1 "hello" mempty result <- clientRequest c rm 1 "hello" mempty
return () return ()
server = TestServer (stdServerConf [("/foo", Normal)]) $ \s -> do server = TestServer (stdServerConf [("/foo", Normal)]) $ \s -> do
let rm = head (registeredMethods s) let rm = head (normalMethods s)
serverHandleNormalCall s rm mempty $ \c body _ -> do serverHandleNormalCall s rm mempty $ \_ body _ -> do
body @?= "hello" body @?= "hello"
return dummyResp return dummyResp
return () return ()
@ -294,8 +405,8 @@ testClientServerCompression =
[("/foo", Normal)] [("/foo", Normal)]
[CompressionAlgArg GrpcCompressDeflate] [CompressionAlgArg GrpcCompressDeflate]
server = TestServer sconf $ \s -> do server = TestServer sconf $ \s -> do
let rm = head (registeredMethods s) let rm = head (normalMethods s)
serverHandleNormalCall s rm dummyMeta $ \c body _ -> do serverHandleNormalCall s rm dummyMeta $ \_sc body _ -> do
body @?= "hello" body @?= "hello"
return ("hello", dummyMeta, StatusOk, StatusDetails "") return ("hello", dummyMeta, StatusOk, StatusDetails "")
return () return ()
@ -303,23 +414,28 @@ testClientServerCompression =
-------------------------------------------------------------------------------- --------------------------------------------------------------------------------
-- Utilities and helpers -- Utilities and helpers
is :: (Eq a, Show a, MonadIO m) => m a -> a -> m ()
is act x = act >>= liftIO . (@?= x)
dummyMeta :: M.Map ByteString ByteString dummyMeta :: M.Map ByteString ByteString
dummyMeta = [("foo","bar")] dummyMeta = [("foo","bar")]
dummyResp :: (ByteString, MetadataMap, StatusCode, StatusDetails)
dummyResp = ("", mempty, StatusOk, StatusDetails "") dummyResp = ("", mempty, StatusOk, StatusDetails "")
dummyHandler :: ServerCall -> ByteString -> MetadataMap dummyHandler :: ServerCall -> ByteString -> MetadataMap
-> IO (ByteString, MetadataMap, StatusCode, StatusDetails) -> IO (ByteString, MetadataMap, StatusCode, StatusDetails)
dummyHandler _ _ _ = return dummyResp dummyHandler _ _ _ = return dummyResp
unavailableStatus :: Either GRPCIOError a dummyResult' :: StatusDetails
unavailableStatus = -> IO (ByteString, MetadataMap, StatusCode, StatusDetails)
Left (GRPCIOBadStatusCode StatusUnavailable (StatusDetails "")) dummyResult' = return . (mempty, mempty, StatusOk, )
deadlineExceededStatus :: Either GRPCIOError a badStatus :: StatusCode -> Either GRPCIOError a
deadlineExceededStatus = badStatus st = Left . GRPCIOBadStatusCode st $ case st of
Left (GRPCIOBadStatusCode StatusDeadlineExceeded StatusDeadlineExceeded -> "Deadline Exceeded"
(StatusDetails "Deadline Exceeded")) StatusCancelled -> "Received RST_STREAM err=8"
_ -> mempty
nop :: Monad m => a -> m () nop :: Monad m => a -> m ()
nop = const (return ()) nop = const (return ())
@ -354,8 +470,8 @@ csTest' nm tc ts =
-- | @checkMD msg expected actual@ fails when keys from @expected@ are not in -- | @checkMD msg expected actual@ fails when keys from @expected@ are not in
-- @actual@, or when values differ for matching keys. -- @actual@, or when values differ for matching keys.
checkMD :: String -> MetadataMap -> MetadataMap -> Assertion checkMD :: String -> MetadataMap -> MetadataMap -> Assertion
checkMD desc expected actual = do checkMD desc expected actual =
when (not $ M.null $ expected `diff` actual) $ do unless (M.null $ expected `diff` actual) $
assertEqual desc expected (actual `M.intersection` expected) assertEqual desc expected (actual `M.intersection` expected)
where where
diff = M.differenceWith $ \a b -> if a == b then Nothing else Just b diff = M.differenceWith $ \a b -> if a == b then Nothing else Just b
@ -363,13 +479,19 @@ checkMD desc expected actual = do
checkReqRslt :: Show a => (b -> Assertion) -> Either a b -> Assertion checkReqRslt :: Show a => (b -> Assertion) -> Either a b -> Assertion
checkReqRslt = either clientFail checkReqRslt = either clientFail
-- | The consumer which asserts that the next value it consumes is equal to the
-- given value; string parameter used as in 'assertEqual'.
assertConsumeEq :: (Eq a, Show a) => String -> a -> P.Consumer a IO ()
assertConsumeEq s v = P.lift . assertEqual s v =<< P.await
clientFail :: Show a => a -> Assertion clientFail :: Show a => a -> Assertion
clientFail = assertFailure . ("Client error: " ++). show clientFail = assertFailure . ("Client error: " ++). show
data TestClient = TestClient ClientConfig (Client -> IO ()) data TestClient = TestClient ClientConfig (Client -> IO ())
runTestClient :: TestClient -> IO () runTestClient :: TestClient -> IO ()
runTestClient (TestClient conf c) = withGRPC $ \g -> withClient g conf c runTestClient (TestClient conf f) =
runManaged $ mgdGRPC >>= mgdClient conf >>= liftIO . f
stdTestClient :: (Client -> IO ()) -> TestClient stdTestClient :: (Client -> IO ()) -> TestClient
stdTestClient = TestClient stdClientConf stdTestClient = TestClient stdClientConf
@ -380,7 +502,8 @@ stdClientConf = ClientConfig "localhost" 50051 []
data TestServer = TestServer ServerConfig (Server -> IO ()) data TestServer = TestServer ServerConfig (Server -> IO ())
runTestServer :: TestServer -> IO () runTestServer :: TestServer -> IO ()
runTestServer (TestServer conf s) = withGRPC $ \g -> withServer g conf s runTestServer (TestServer conf f) =
runManaged $ mgdGRPC >>= mgdServer conf >>= liftIO . f
stdTestServer :: [(MethodName, GRPCMethodType)] -> (Server -> IO ()) -> TestServer stdTestServer :: [(MethodName, GRPCMethodType)] -> (Server -> IO ()) -> TestServer
stdTestServer = TestServer . stdServerConf stdTestServer = TestServer . stdServerConf
@ -388,6 +511,14 @@ stdTestServer = TestServer . stdServerConf
stdServerConf :: [(MethodName, GRPCMethodType)] -> ServerConfig stdServerConf :: [(MethodName, GRPCMethodType)] -> ServerConfig
stdServerConf xs = ServerConfig "localhost" 50051 xs [] stdServerConf xs = ServerConfig "localhost" 50051 xs []
threadDelaySecs :: Int -> IO () threadDelaySecs :: Int -> IO ()
threadDelaySecs = threadDelay . (* 10^(6::Int)) threadDelaySecs = threadDelay . (* 10^(6::Int))
mgdGRPC :: Managed GRPC
mgdGRPC = managed withGRPC
mgdClient :: ClientConfig -> GRPC -> Managed Client
mgdClient conf g = managed $ withClient g conf
mgdServer :: ServerConfig -> GRPC -> Managed Server
mgdServer conf g = managed $ withServer g conf

2
tests/LowLevelTests/Op.hs
View file

@ -76,7 +76,7 @@ withClientServerUnaryCall grpc f = do
crm <- clientRegisterMethod c "/foo" Normal crm <- clientRegisterMethod c "/foo" Normal
withServer grpc serverConf $ \s -> withServer grpc serverConf $ \s ->
withClientCall c crm 10 $ \cc -> do withClientCall c crm 10 $ \cc -> do
let srm = head (registeredMethods s) let srm = head (normalMethods s)
-- NOTE: We need to send client ops here or else `withServerCall` hangs, -- NOTE: We need to send client ops here or else `withServerCall` hangs,
-- because registered methods try to do recv ops immediately when -- because registered methods try to do recv ops immediately when
-- created. If later we want to send payloads or metadata, we'll need -- created. If later we want to send payloads or metadata, we'll need