Server.hs

-- | Node with server role (i.e., its peer is a client)
module Network.GRPC.Util.Session.Server (
    ConnectionToClient(..)
  , setupResponseChannel
  ) where

import Control.Concurrent
import Control.Exception
import Network.HTTP.Semantics qualified as HTTP
import Network.HTTP.Semantics.Server qualified as Server

import Network.GRPC.Common.Exception
import Network.GRPC.Util.Imports
import Network.GRPC.Util.ServerStream
import Network.GRPC.Util.Session.API
import Network.GRPC.Util.Session.Channel
import Network.GRPC.Util.Stream
import Network.GRPC.Util.Thread

{-------------------------------------------------------------------------------
  Connection
-------------------------------------------------------------------------------}

-- | Connection to the client, as provided by @http2@
data ConnectionToClient = ConnectionToClient {
      request :: Server.Request
    , respond :: Server.Response -> IO ()
    }

{-------------------------------------------------------------------------------
  Internal auxiliary: constructing responses
-------------------------------------------------------------------------------}

respondStreamingWithResult :: forall a.
     HasCallStack
  => ConnectionToClient
  -> Server.TrailersMaker
  -> ResponseInfo
  -> (OutputStream -> IO a)
  -> IO a
respondStreamingWithResult conn trailers responseInfo body = do
    resultVar :: MVar (Either ExactException a) <- newEmptyMVar
    let resp :: Server.Response
        resp = flip Server.setResponseTrailersMaker trailers
             . Server.responseStreamingIface
                 (responseStatus  responseInfo)
                 (responseHeaders responseInfo)
             $ auxThreadBody resultVar

    respond conn resp

    -- Any exception thrown here is thrown in the context of a 'Thread'.
    either throwExact return =<< takeMVar resultVar
  where
    -- This will be running in an auxiliary thread, spawned by http2. Any
    -- exceptions that are thrown by that thread will remain uncaught, and
    -- will trigger the top-level uncaught exception handler. We therefore
    -- catch all of these and store them in 'resultVar', which the main
    -- grapesy outbound thread is waiting on.
    --
    -- When the connection is closed, this thread will be cleaned up by
    -- the ThreadManager in http2.
    auxThreadBody ::
         MVar (Either ExactException a)
      -> HTTP.OutBodyIface
      -> IO ()
    auxThreadBody resultVar iface = do
        result <- try $ do
          -- It is important we create the output stream inside the body of the
          -- exception handler, since that too can fail.
          ostrm <- serverOutputStream iface
          body ostrm
        putMVar resultVar result

respondNoBody :: ConnectionToClient -> ResponseInfo -> IO ()
respondNoBody conn responseInfo =
    respond conn resp
  where
    resp :: Server.Response
    resp = Server.responseNoBody
             (responseStatus  responseInfo)
             (responseHeaders responseInfo)

{-------------------------------------------------------------------------------
  Initiate response
-------------------------------------------------------------------------------}

-- | Setup response channel
--
-- Notes:
--
-- * The actual response will not immediately be initiated; see below.
-- * We assume that the client is allowed to close their outbound stream to us.
-- * 'setupResponseChannel' will not throw any exceptions.
setupResponseChannel :: forall sess.
     (HasCallStack, IsSession sess)
  => sess
  -> ConnectionToClient
  -> FlowStart (Inbound sess)
  -> IO (FlowStart (Outbound sess), ResponseInfo)
  -- ^ Construct headers for the initial response
  --
  -- This function is allowed to block. If it does, no response will not be
  -- initiated until it returns.
  --
  -- If this function throws an exception, the response is never initiated;
  -- this is treated the same was as when we fail to set up the outbound
  -- connection due to a network failure.
  -> IO (Channel sess)
setupResponseChannel sess
                     conn
                     inboundStart
                     startOutbound
                   = do
    channel <- initChannel "server"
    monitorInbound channel

    forkThread "grapesy:serverInbound" (channelInbound channel) $ \unmask ctxt -> unmask $
      case inboundStart of
        FlowStartRegular headers -> do
          regular <- initFlowStateRegular headers
          stream  <- serverInputStream (request conn)
          threadMainBody ctxt regular $ \markDone ->
            markDone =<< recvMessageLoop sess regular stream
        FlowStartNoMessages trailers ->
          -- The client sent a request with an empty body
          threadTrivial ctxt trailers

    forkThread "grapesy:serverOutbound" (channelOutbound channel) $ \unmask ctxt -> unmask $ do
      (outboundStart, responseInfo) <- startOutbound
      case outboundStart of
        FlowStartRegular headers -> do
          regular <- initFlowStateRegular headers
          threadMainBody ctxt regular $ \markDone -> do
            respondStreamingWithResult
                conn
                (outboundTrailersMaker sess channel regular)
                responseInfo $ \stream ->
              sendMessageLoop sess regular stream markDone
        FlowStartNoMessages trailers -> do
          respondNoBody conn responseInfo
          threadTrivial ctxt trailers

    return channel

-- | Make outbound thread monitor the input thread
--
-- Monitoring here is a bit subtle: http2 will spawn an auxiliary thread, which
-- will be the one that will actually run the 'sendMessageLoop'. Meanwhile, our
-- \"outbound thread\" will simply be waiting for that auxiliary thread to
-- terminate; if the monitor fires, is it /this/ thread that receives it, not
-- the auxiliary http2 thread.
monitorInbound :: forall sess. Channel sess -> IO ()
monitorInbound channel = do
    _monitorRef <-
      threadMonitor
        (channelOutbound channel)
        (channelInbound channel)
        monitorPred
    return ()
  where
    -- Unlike on the client-side, if the input thread terminates normally, the
    -- server can continue to run normally (the stream can be half-closed from
    -- the client). We only only terminate if the inbound thread threw an
    -- exception, indicating that the client disconnected abruptly.
    --
    -- See also 'Network.GRPC.Util.Session.Client.setupRequestChannel'.
    monitorPred ::
          Either
            ThreadException
            ( Either
                (NoMessages (Inbound sess))
                (Trailers   (Inbound sess))
            )
       -> Maybe ExactException
    monitorPred = \case
        Left  e         -> Just $ threadException e
        Right _trailers -> Nothing