2022-03-16 03:39:21 +03:00
|
|
|
{-# LANGUAGE TemplateHaskell #-}
|
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
-- | Top-level management of subscription poller threads.
|
|
|
|
-- The implementation of the polling itself is
|
|
|
|
-- in "Hasura.GraphQL.Execute.Subscription.Poll". See "Hasura.GraphQL.Execute.Subscription" for high-level
|
2019-08-28 15:19:21 +03:00
|
|
|
-- details.
|
2022-03-21 13:39:49 +03:00
|
|
|
module Hasura.GraphQL.Execute.Subscription.State
|
|
|
|
( SubscriptionsState (..),
|
|
|
|
initSubscriptionsState,
|
|
|
|
dumpSubscriptionsState,
|
|
|
|
SubscriberDetails,
|
|
|
|
SubscriptionPostPollHook,
|
2021-09-24 01:56:37 +03:00
|
|
|
addLiveQuery,
|
2022-04-07 17:41:43 +03:00
|
|
|
addStreamSubscriptionQuery,
|
2021-09-24 01:56:37 +03:00
|
|
|
removeLiveQuery,
|
2022-04-07 17:41:43 +03:00
|
|
|
removeStreamingQuery,
|
2021-09-24 01:56:37 +03:00
|
|
|
LiveAsyncActionQueryOnSource (..),
|
|
|
|
LiveAsyncActionQueryWithNoRelationships (..),
|
|
|
|
LiveAsyncActionQuery (..),
|
|
|
|
AsyncActionQueryLive (..),
|
|
|
|
AsyncActionSubscriptionState,
|
|
|
|
addAsyncActionLiveQuery,
|
|
|
|
removeAsyncActionLiveQuery,
|
2022-03-21 13:39:49 +03:00
|
|
|
LiveQuerySubscriberDetails,
|
2022-04-07 17:41:43 +03:00
|
|
|
StreamingSubscriberDetails,
|
2021-09-24 01:56:37 +03:00
|
|
|
)
|
|
|
|
where
|
|
|
|
|
|
|
|
import Control.Concurrent.Extended (forkImmortal, sleep)
|
|
|
|
import Control.Concurrent.STM qualified as STM
|
|
|
|
import Control.Exception (mask_)
|
|
|
|
import Control.Immortal qualified as Immortal
|
|
|
|
import Data.Aeson.Extended qualified as J
|
|
|
|
import Data.String
|
|
|
|
import Data.Text.Extended
|
|
|
|
import Data.UUID.V4 qualified as UUID
|
|
|
|
import GHC.AssertNF.CPP
|
|
|
|
import Hasura.Base.Error
|
|
|
|
import Hasura.GraphQL.Execute.Backend
|
2022-03-21 13:39:49 +03:00
|
|
|
import Hasura.GraphQL.Execute.Subscription.Options
|
|
|
|
import Hasura.GraphQL.Execute.Subscription.Plan
|
|
|
|
import Hasura.GraphQL.Execute.Subscription.Poll
|
|
|
|
import Hasura.GraphQL.Execute.Subscription.TMap qualified as TMap
|
2021-09-24 01:56:37 +03:00
|
|
|
import Hasura.GraphQL.ParameterizedQueryHash (ParameterizedQueryHash)
|
|
|
|
import Hasura.GraphQL.Transport.Backend
|
|
|
|
import Hasura.GraphQL.Transport.HTTP.Protocol (OperationName)
|
|
|
|
import Hasura.GraphQL.Transport.WebSocket.Protocol (OperationId)
|
|
|
|
import Hasura.Logging qualified as L
|
|
|
|
import Hasura.Prelude
|
|
|
|
import Hasura.RQL.Types.Action
|
|
|
|
import Hasura.RQL.Types.Common (SourceName, unNonNegativeDiffTime)
|
|
|
|
import Hasura.Server.Metrics (ServerMetrics (..))
|
|
|
|
import Hasura.Server.Types (RequestId)
|
2022-04-07 17:41:43 +03:00
|
|
|
import Language.GraphQL.Draft.Syntax qualified as G
|
2021-09-24 01:56:37 +03:00
|
|
|
import StmContainers.Map qualified as STMMap
|
|
|
|
import System.Metrics.Gauge qualified as EKG.Gauge
|
2021-05-11 18:18:31 +03:00
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
-- | The top-level datatype that holds the state for all active subscriptions.
|
2020-03-05 20:59:26 +03:00
|
|
|
--
|
2020-06-04 20:25:21 +03:00
|
|
|
-- NOTE!: This must be kept consistent with a websocket connection's
|
|
|
|
-- 'OperationMap', in 'onClose' and 'onStart'.
|
2022-03-21 13:39:49 +03:00
|
|
|
data SubscriptionsState = SubscriptionsState
|
2022-04-07 17:41:43 +03:00
|
|
|
{ _ssLiveQueryOptions :: LiveQueriesOptions,
|
|
|
|
_ssStreamQueryOptions :: StreamQueriesOptions,
|
|
|
|
_ssLiveQueryMap :: PollerMap (),
|
|
|
|
_ssStreamQueryMap :: PollerMap (STM.TVar CursorVariableValues),
|
2021-09-24 01:56:37 +03:00
|
|
|
-- | A hook function which is run after each fetch cycle
|
2022-04-22 22:53:12 +03:00
|
|
|
_ssPostPollHook :: SubscriptionPostPollHook,
|
|
|
|
_ssAsyncActions :: AsyncActionSubscriptionState
|
2019-08-28 15:19:21 +03:00
|
|
|
}
|
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
initSubscriptionsState ::
|
2022-04-07 17:41:43 +03:00
|
|
|
LiveQueriesOptions -> StreamQueriesOptions -> SubscriptionPostPollHook -> IO SubscriptionsState
|
|
|
|
initSubscriptionsState liveQOptions streamQOptions pollHook =
|
2021-09-24 01:56:37 +03:00
|
|
|
STM.atomically $
|
2022-04-07 17:41:43 +03:00
|
|
|
SubscriptionsState liveQOptions <$> pure streamQOptions <*> STMMap.new <*> STMMap.new <*> pure pollHook <*> TMap.new
|
2019-08-28 15:19:21 +03:00
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
dumpSubscriptionsState :: Bool -> SubscriptionsState -> IO J.Value
|
2022-04-07 17:41:43 +03:00
|
|
|
dumpSubscriptionsState extended (SubscriptionsState liveQOpts streamQOpts lqMap streamMap _ _) = do
|
2019-08-28 15:19:21 +03:00
|
|
|
lqMapJ <- dumpPollerMap extended lqMap
|
2022-04-07 17:41:43 +03:00
|
|
|
streamMapJ <- dumpPollerMap extended streamMap
|
2021-09-24 01:56:37 +03:00
|
|
|
return $
|
|
|
|
J.object
|
2022-03-21 13:39:49 +03:00
|
|
|
[ "options" J..= liveQOpts,
|
2022-04-07 17:41:43 +03:00
|
|
|
"live_queries_map" J..= lqMapJ,
|
|
|
|
"stream_queries_map" J..= streamMapJ,
|
|
|
|
"stream_queries_options" J..= streamQOpts
|
2021-09-24 01:56:37 +03:00
|
|
|
]
|
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
-- | SubscriberDetails contains the data required to locate a subscriber
|
|
|
|
-- in the correct cohort within the correct poller in the operation map.
|
|
|
|
data SubscriberDetails a = SubscriberDetails
|
|
|
|
{ _sdPoller :: !PollerKey,
|
|
|
|
_sdCohort :: !a,
|
|
|
|
_sdSubscriber :: !SubscriberId
|
2021-09-24 01:56:37 +03:00
|
|
|
}
|
|
|
|
deriving (Show)
|
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
type LiveQuerySubscriberDetails = SubscriberDetails CohortKey
|
|
|
|
|
2022-04-07 17:41:43 +03:00
|
|
|
-- | The `CohortKey` contains the variables with which the subscription was started
|
|
|
|
-- and which will remain unchanged. The second type contains the mutable reference
|
|
|
|
-- through which we can get the latest value of the cursor and using both the `CohortKey`
|
|
|
|
-- and the latest cursor value, we locate the subscriber in the operation map to find its
|
|
|
|
-- details and then stop it.
|
|
|
|
type StreamingSubscriberDetails = SubscriberDetails (CohortKey, STM.TVar CursorVariableValues)
|
|
|
|
|
2022-04-22 22:53:12 +03:00
|
|
|
-- | `findPollerForSubscriber` places a subscriber in the correct poller.
|
|
|
|
-- If the poller doesn't exist then we create one otherwise we return the
|
|
|
|
-- existing one.
|
|
|
|
findPollerForSubscriber ::
|
|
|
|
Subscriber ->
|
|
|
|
CohortId ->
|
|
|
|
PollerMap streamCursorVars ->
|
|
|
|
PollerKey ->
|
|
|
|
CohortKey ->
|
|
|
|
(Subscriber -> Cohort streamCursorVars -> STM.STM streamCursorVars) ->
|
|
|
|
(Subscriber -> CohortId -> Poller streamCursorVars -> STM.STM streamCursorVars) ->
|
|
|
|
STM.STM ((Maybe (Poller streamCursorVars)), streamCursorVars)
|
|
|
|
findPollerForSubscriber subscriber cohortId pollerMap pollerKey cohortKey addToCohort addToPoller =
|
|
|
|
-- a handler is returned only when it is newly created
|
|
|
|
STMMap.lookup pollerKey pollerMap >>= \case
|
|
|
|
Just poller -> do
|
|
|
|
-- Found a poller, now check if a cohort also exists
|
|
|
|
cursorVars <-
|
|
|
|
TMap.lookup cohortKey (_pCohorts poller) >>= \case
|
|
|
|
-- cohort found too! Simply add the subscriber to the cohort
|
|
|
|
Just cohort -> addToCohort subscriber cohort
|
|
|
|
-- cohort not found. Create a cohort with the subscriber and add
|
|
|
|
-- the cohort to the poller
|
|
|
|
Nothing -> addToPoller subscriber cohortId poller
|
|
|
|
return (Nothing, cursorVars)
|
|
|
|
Nothing -> do
|
|
|
|
-- no poller found, so create one with the cohort
|
|
|
|
-- and the subscriber within it.
|
|
|
|
!poller <- Poller <$> TMap.new <*> STM.newEmptyTMVar
|
|
|
|
cursorVars <- addToPoller subscriber cohortId poller
|
|
|
|
STMMap.insert poller pollerKey pollerMap
|
|
|
|
return $ (Just poller, cursorVars)
|
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
-- | Fork a thread handling a regular (live query) subscription
|
2021-09-24 01:56:37 +03:00
|
|
|
addLiveQuery ::
|
|
|
|
forall b.
|
|
|
|
BackendTransport b =>
|
|
|
|
L.Logger L.Hasura ->
|
|
|
|
ServerMetrics ->
|
|
|
|
SubscriberMetadata ->
|
2022-03-21 13:39:49 +03:00
|
|
|
SubscriptionsState ->
|
2021-09-24 01:56:37 +03:00
|
|
|
SourceName ->
|
|
|
|
ParameterizedQueryHash ->
|
|
|
|
-- | operation name of the query
|
|
|
|
Maybe OperationName ->
|
|
|
|
RequestId ->
|
2022-03-21 13:39:49 +03:00
|
|
|
SubscriptionQueryPlan b (MultiplexedQuery b) ->
|
2021-09-24 01:56:37 +03:00
|
|
|
-- | the action to be executed when result changes
|
|
|
|
OnChange ->
|
2022-03-21 13:39:49 +03:00
|
|
|
IO LiveQuerySubscriberDetails
|
2019-08-28 15:19:21 +03:00
|
|
|
addLiveQuery
|
2021-09-24 01:56:37 +03:00
|
|
|
logger
|
|
|
|
serverMetrics
|
|
|
|
subscriberMetadata
|
2022-03-21 13:39:49 +03:00
|
|
|
subscriptionState
|
2021-09-24 01:56:37 +03:00
|
|
|
source
|
|
|
|
parameterizedQueryHash
|
|
|
|
operationName
|
|
|
|
requestId
|
|
|
|
plan
|
|
|
|
onResultAction = do
|
|
|
|
-- CAREFUL!: It's absolutely crucial that we can't throw any exceptions here!
|
|
|
|
|
|
|
|
-- disposable UUIDs:
|
|
|
|
cohortId <- newCohortId
|
|
|
|
subscriberId <- newSubscriberId
|
|
|
|
|
|
|
|
let !subscriber = Subscriber subscriberId subscriberMetadata requestId operationName onResultAction
|
|
|
|
|
|
|
|
$assertNFHere subscriber -- so we don't write thunks to mutable vars
|
2022-04-22 22:53:12 +03:00
|
|
|
(pollerMaybe, ()) <-
|
2021-09-24 01:56:37 +03:00
|
|
|
STM.atomically $
|
2022-04-22 22:53:12 +03:00
|
|
|
findPollerForSubscriber
|
|
|
|
subscriber
|
|
|
|
cohortId
|
|
|
|
lqMap
|
|
|
|
handlerId
|
|
|
|
cohortKey
|
|
|
|
addToCohort
|
|
|
|
addToPoller
|
2021-09-24 01:56:37 +03:00
|
|
|
|
|
|
|
-- we can then attach a polling thread if it is new the livequery can only be
|
|
|
|
-- cancelled after putTMVar
|
2022-04-22 22:53:12 +03:00
|
|
|
onJust pollerMaybe $ \poller -> do
|
2021-09-24 01:56:37 +03:00
|
|
|
pollerId <- PollerId <$> UUID.nextRandom
|
2022-03-21 13:39:49 +03:00
|
|
|
threadRef <- forkImmortal ("pollLiveQuery." <> show pollerId) logger $
|
2021-09-24 01:56:37 +03:00
|
|
|
forever $ do
|
2022-04-22 22:53:12 +03:00
|
|
|
pollLiveQuery @b pollerId lqOpts (source, sourceConfig) role parameterizedQueryHash query (_pCohorts poller) postPollHook
|
2021-09-24 01:56:37 +03:00
|
|
|
sleep $ unNonNegativeDiffTime $ unRefetchInterval refetchInterval
|
|
|
|
let !pState = PollerIOState threadRef pollerId
|
|
|
|
$assertNFHere pState -- so we don't write thunks to mutable vars
|
2022-04-22 22:53:12 +03:00
|
|
|
STM.atomically $ STM.putTMVar (_pIOState poller) pState
|
2021-09-24 01:56:37 +03:00
|
|
|
|
|
|
|
liftIO $ EKG.Gauge.inc $ smActiveSubscriptions serverMetrics
|
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
pure $ SubscriberDetails handlerId cohortKey subscriberId
|
2021-09-24 01:56:37 +03:00
|
|
|
where
|
2022-04-07 17:41:43 +03:00
|
|
|
SubscriptionsState lqOpts _ lqMap _ postPollHook _ = subscriptionState
|
2022-03-21 13:39:49 +03:00
|
|
|
SubscriptionsOptions _ refetchInterval = lqOpts
|
|
|
|
SubscriptionQueryPlan (ParameterizedSubscriptionQueryPlan role query) sourceConfig cohortKey _ = plan
|
2021-09-24 01:56:37 +03:00
|
|
|
|
|
|
|
handlerId = PollerKey source role $ toTxt query
|
|
|
|
|
|
|
|
addToCohort subscriber handlerC =
|
|
|
|
TMap.insert subscriber (_sId subscriber) $ _cNewSubscribers handlerC
|
|
|
|
|
|
|
|
addToPoller subscriber cohortId handler = do
|
2022-03-21 13:39:49 +03:00
|
|
|
!newCohort <-
|
|
|
|
Cohort cohortId
|
|
|
|
<$> STM.newTVar Nothing
|
|
|
|
<*> TMap.new
|
|
|
|
<*> TMap.new
|
2022-04-07 17:41:43 +03:00
|
|
|
<*> pure ()
|
2021-09-24 01:56:37 +03:00
|
|
|
addToCohort subscriber newCohort
|
|
|
|
TMap.insert newCohort cohortKey $ _pCohorts handler
|
|
|
|
|
2022-04-07 17:41:43 +03:00
|
|
|
-- | Fork a thread handling a streaming subscription
|
|
|
|
addStreamSubscriptionQuery ::
|
|
|
|
forall b.
|
|
|
|
BackendTransport b =>
|
|
|
|
L.Logger L.Hasura ->
|
|
|
|
ServerMetrics ->
|
|
|
|
SubscriberMetadata ->
|
|
|
|
SubscriptionsState ->
|
|
|
|
SourceName ->
|
|
|
|
ParameterizedQueryHash ->
|
|
|
|
-- | operation name of the query
|
|
|
|
Maybe OperationName ->
|
|
|
|
RequestId ->
|
|
|
|
-- | root field name
|
|
|
|
G.Name ->
|
|
|
|
SubscriptionQueryPlan b (MultiplexedQuery b) ->
|
|
|
|
-- | the action to be executed when result changes
|
|
|
|
OnChange ->
|
|
|
|
IO StreamingSubscriberDetails
|
|
|
|
addStreamSubscriptionQuery
|
|
|
|
logger
|
|
|
|
serverMetrics
|
|
|
|
subscriberMetadata
|
|
|
|
subscriptionState
|
|
|
|
source
|
|
|
|
parameterizedQueryHash
|
|
|
|
operationName
|
|
|
|
requestId
|
|
|
|
rootFieldName
|
|
|
|
plan
|
|
|
|
onResultAction = do
|
|
|
|
-- CAREFUL!: It's absolutely crucial that we can't throw any exceptions here!
|
|
|
|
|
|
|
|
-- disposable UUIDs:
|
|
|
|
cohortId <- newCohortId
|
|
|
|
subscriberId <- newSubscriberId
|
|
|
|
|
|
|
|
let !subscriber = Subscriber subscriberId subscriberMetadata requestId operationName onResultAction
|
|
|
|
|
|
|
|
$assertNFHere subscriber -- so we don't write thunks to mutable vars
|
|
|
|
(handlerM, cohortCursorTVar) <-
|
|
|
|
STM.atomically $
|
2022-04-22 22:53:12 +03:00
|
|
|
findPollerForSubscriber
|
|
|
|
subscriber
|
|
|
|
cohortId
|
|
|
|
streamQueryMap
|
|
|
|
handlerId
|
|
|
|
cohortKey
|
|
|
|
addToCohort
|
|
|
|
addToPoller
|
2022-04-07 17:41:43 +03:00
|
|
|
|
|
|
|
-- we can then attach a polling thread if it is new the subscription can only be
|
|
|
|
-- cancelled after putTMVar
|
|
|
|
onJust handlerM $ \handler -> do
|
|
|
|
pollerId <- PollerId <$> UUID.nextRandom
|
|
|
|
threadRef <- forkImmortal ("pollStreamingQuery." <> show (unPollerId pollerId)) logger $
|
|
|
|
forever $ do
|
|
|
|
pollStreamingQuery @b pollerId streamQOpts (source, sourceConfig) role parameterizedQueryHash query (_pCohorts handler) rootFieldName postPollHook Nothing
|
|
|
|
sleep $ unNonNegativeDiffTime $ unRefetchInterval refetchInterval
|
|
|
|
let !pState = PollerIOState threadRef pollerId
|
|
|
|
$assertNFHere pState -- so we don't write thunks to mutable vars
|
|
|
|
STM.atomically $ STM.putTMVar (_pIOState handler) pState
|
|
|
|
|
|
|
|
liftIO $ EKG.Gauge.inc $ smActiveSubscriptions serverMetrics
|
|
|
|
|
|
|
|
pure $ SubscriberDetails handlerId (cohortKey, cohortCursorTVar) subscriberId
|
|
|
|
where
|
|
|
|
SubscriptionsState _ streamQOpts _ streamQueryMap postPollHook _ = subscriptionState
|
|
|
|
SubscriptionsOptions _ refetchInterval = streamQOpts
|
|
|
|
SubscriptionQueryPlan (ParameterizedSubscriptionQueryPlan role query) sourceConfig cohortKey _ = plan
|
|
|
|
|
|
|
|
handlerId = PollerKey source role $ toTxt query
|
|
|
|
|
|
|
|
addToCohort subscriber handlerC = do
|
|
|
|
TMap.insert subscriber (_sId subscriber) $ _cNewSubscribers handlerC
|
|
|
|
pure $ _cStreamCursorVariables handlerC
|
|
|
|
|
|
|
|
addToPoller subscriber cohortId handler = do
|
|
|
|
latestCursorValues <-
|
|
|
|
STM.newTVar (CursorVariableValues (_unValidatedVariables (_cvCursorVariables cohortKey)))
|
|
|
|
!newCohort <- Cohort cohortId <$> STM.newTVar Nothing <*> TMap.new <*> TMap.new <*> pure latestCursorValues
|
|
|
|
cohortCursorVals <- addToCohort subscriber newCohort
|
|
|
|
TMap.insert newCohort cohortKey $ _pCohorts handler
|
|
|
|
pure cohortCursorVals
|
|
|
|
|
2021-09-24 01:56:37 +03:00
|
|
|
removeLiveQuery ::
|
|
|
|
L.Logger L.Hasura ->
|
|
|
|
ServerMetrics ->
|
2022-03-21 13:39:49 +03:00
|
|
|
SubscriptionsState ->
|
2019-08-28 15:19:21 +03:00
|
|
|
-- the query and the associated operation
|
2022-03-21 13:39:49 +03:00
|
|
|
LiveQuerySubscriberDetails ->
|
2021-09-24 01:56:37 +03:00
|
|
|
IO ()
|
2022-03-21 13:39:49 +03:00
|
|
|
removeLiveQuery logger serverMetrics lqState lqId@(SubscriberDetails handlerId cohortId sinkId) = mask_ $ do
|
2020-03-05 20:59:26 +03:00
|
|
|
mbCleanupIO <- STM.atomically $ do
|
2022-03-21 13:39:49 +03:00
|
|
|
detM <- getQueryDet lqMap
|
2021-09-24 01:56:37 +03:00
|
|
|
fmap join $
|
|
|
|
forM detM $ \(Poller cohorts ioState, cohort) ->
|
|
|
|
cleanHandlerC cohorts ioState cohort
|
2020-03-05 20:59:26 +03:00
|
|
|
sequence_ mbCleanupIO
|
2021-05-05 07:24:12 +03:00
|
|
|
liftIO $ EKG.Gauge.dec $ smActiveSubscriptions serverMetrics
|
2019-08-28 15:19:21 +03:00
|
|
|
where
|
2022-03-21 13:39:49 +03:00
|
|
|
lqMap = _ssLiveQueryMap lqState
|
2019-08-28 15:19:21 +03:00
|
|
|
|
2022-03-21 13:39:49 +03:00
|
|
|
getQueryDet subMap = do
|
|
|
|
pollerM <- STMMap.lookup handlerId subMap
|
2021-09-24 01:56:37 +03:00
|
|
|
fmap join $
|
|
|
|
forM pollerM $ \poller -> do
|
|
|
|
cohortM <- TMap.lookup cohortId (_pCohorts poller)
|
|
|
|
return $ (poller,) <$> cohortM
|
2019-08-28 15:19:21 +03:00
|
|
|
|
|
|
|
cleanHandlerC cohortMap ioState handlerC = do
|
|
|
|
let curOps = _cExistingSubscribers handlerC
|
|
|
|
newOps = _cNewSubscribers handlerC
|
|
|
|
TMap.delete sinkId curOps
|
|
|
|
TMap.delete sinkId newOps
|
2021-09-24 01:56:37 +03:00
|
|
|
cohortIsEmpty <-
|
|
|
|
(&&)
|
|
|
|
<$> TMap.null curOps
|
|
|
|
<*> TMap.null newOps
|
2019-08-28 15:19:21 +03:00
|
|
|
when cohortIsEmpty $ TMap.delete cohortId cohortMap
|
|
|
|
handlerIsEmpty <- TMap.null cohortMap
|
2020-03-05 20:59:26 +03:00
|
|
|
-- when there is no need for handler i.e, this happens to be the last
|
|
|
|
-- operation, take the ref for the polling thread to cancel it
|
2019-08-28 15:19:21 +03:00
|
|
|
if handlerIsEmpty
|
|
|
|
then do
|
|
|
|
STMMap.delete handlerId lqMap
|
2020-03-05 20:59:26 +03:00
|
|
|
threadRefM <- fmap _pThread <$> STM.tryReadTMVar ioState
|
2021-09-24 01:56:37 +03:00
|
|
|
return $
|
|
|
|
Just $ -- deferred IO:
|
|
|
|
case threadRefM of
|
|
|
|
Just threadRef -> Immortal.stop threadRef
|
|
|
|
-- This would seem to imply addLiveQuery broke or a bug
|
|
|
|
-- elsewhere. Be paranoid and log:
|
|
|
|
Nothing ->
|
|
|
|
L.unLogger logger $
|
|
|
|
L.UnstructuredLog L.LevelError $
|
|
|
|
fromString $
|
|
|
|
"In removeLiveQuery no worker thread installed. Please report this as a bug: "
|
|
|
|
<> show lqId
|
2019-08-28 15:19:21 +03:00
|
|
|
else return Nothing
|
2021-03-31 13:39:01 +03:00
|
|
|
|
2022-04-07 17:41:43 +03:00
|
|
|
removeStreamingQuery ::
|
|
|
|
L.Logger L.Hasura ->
|
|
|
|
ServerMetrics ->
|
|
|
|
SubscriptionsState ->
|
|
|
|
-- the query and the associated operation
|
|
|
|
StreamingSubscriberDetails ->
|
|
|
|
IO ()
|
|
|
|
removeStreamingQuery logger serverMetrics subscriptionState (SubscriberDetails handlerId (cohortId, cursorVariableTV) sinkId) = mask_ $ do
|
|
|
|
mbCleanupIO <- STM.atomically $ do
|
|
|
|
detM <- getQueryDet streamQMap
|
|
|
|
fmap join $
|
|
|
|
forM detM $ \(Poller cohorts ioState, currentCohortId, cohort) ->
|
|
|
|
cleanHandlerC cohorts ioState (cohort, currentCohortId)
|
|
|
|
sequence_ mbCleanupIO
|
|
|
|
liftIO $ EKG.Gauge.dec $ smActiveSubscriptions serverMetrics
|
|
|
|
where
|
|
|
|
streamQMap = _ssStreamQueryMap subscriptionState
|
|
|
|
|
|
|
|
getQueryDet subMap = do
|
|
|
|
pollerM <- STMMap.lookup handlerId subMap
|
|
|
|
(CursorVariableValues currentCohortCursorVal) <- STM.readTVar cursorVariableTV
|
|
|
|
let updatedCohortId = modifyCursorCohortVariables (mkUnsafeValidateVariables currentCohortCursorVal) cohortId
|
|
|
|
fmap join $
|
|
|
|
forM pollerM $ \poller -> do
|
|
|
|
cohortM <- TMap.lookup updatedCohortId (_pCohorts poller)
|
|
|
|
return $ (poller,updatedCohortId,) <$> cohortM
|
|
|
|
|
|
|
|
cleanHandlerC cohortMap ioState (handlerC, currentCohortId) = do
|
|
|
|
let curOps = _cExistingSubscribers handlerC
|
|
|
|
newOps = _cNewSubscribers handlerC
|
|
|
|
TMap.delete sinkId curOps
|
|
|
|
TMap.delete sinkId newOps
|
|
|
|
cohortIsEmpty <-
|
|
|
|
(&&)
|
|
|
|
<$> TMap.null curOps
|
|
|
|
<*> TMap.null newOps
|
|
|
|
when cohortIsEmpty $ TMap.delete currentCohortId cohortMap
|
|
|
|
handlerIsEmpty <- TMap.null cohortMap
|
|
|
|
-- when there is no need for handler i.e,
|
|
|
|
-- operation, take the ref for the polling thread to cancel it
|
|
|
|
if handlerIsEmpty
|
|
|
|
then do
|
|
|
|
STMMap.delete handlerId streamQMap
|
|
|
|
threadRefM <- fmap _pThread <$> STM.tryReadTMVar ioState
|
|
|
|
return $
|
|
|
|
Just $ -- deferred IO:
|
|
|
|
case threadRefM of
|
|
|
|
Just threadRef -> Immortal.stop threadRef
|
|
|
|
-- This would seem to imply addStreamSubscriptionQuery broke or a bug
|
|
|
|
-- elsewhere. Be paranoid and log:
|
|
|
|
Nothing ->
|
|
|
|
L.unLogger logger $
|
|
|
|
L.UnstructuredLog L.LevelError $
|
|
|
|
fromString $
|
|
|
|
"In removeLiveQuery no worker thread installed. Please report this as a bug: "
|
|
|
|
<> " poller_id: "
|
|
|
|
<> show handlerId
|
|
|
|
<> ", cohort_id: "
|
|
|
|
<> show cohortId
|
|
|
|
<> ", subscriber_id:"
|
|
|
|
<> show sinkId
|
|
|
|
else return Nothing
|
|
|
|
|
2021-03-31 13:39:01 +03:00
|
|
|
-- | An async action query whose relationships are refered to table in a source.
|
|
|
|
-- We need to generate an SQL statement with the action response and execute it
|
|
|
|
-- in the source database so as to fetch response joined with relationship rows.
|
|
|
|
-- For more details see Note [Resolving async action query]
|
2021-09-24 01:56:37 +03:00
|
|
|
data LiveAsyncActionQueryOnSource = LiveAsyncActionQueryOnSource
|
2022-03-21 13:39:49 +03:00
|
|
|
{ _laaqpCurrentLqId :: !LiveQuerySubscriberDetails,
|
2021-09-24 01:56:37 +03:00
|
|
|
_laaqpPrevActionLogMap :: !ActionLogResponseMap,
|
|
|
|
-- | An IO action to restart the live query poller with updated action log responses fetched from metadata storage
|
|
|
|
-- Restarting a live query re-generates the SQL statement with new action log responses to send latest action
|
|
|
|
-- response to the client.
|
2022-03-21 13:39:49 +03:00
|
|
|
_laaqpRestartLq :: !(LiveQuerySubscriberDetails -> ActionLogResponseMap -> IO (Maybe LiveQuerySubscriberDetails))
|
2021-03-31 13:39:01 +03:00
|
|
|
}
|
|
|
|
|
2021-09-24 01:56:37 +03:00
|
|
|
data LiveAsyncActionQueryWithNoRelationships = LiveAsyncActionQueryWithNoRelationships
|
|
|
|
{ -- | An IO action to send response to the websocket client
|
|
|
|
_laaqwnrSendResponse :: !(ActionLogResponseMap -> IO ()),
|
|
|
|
-- | An IO action to send "completed" message to the websocket client
|
|
|
|
_laaqwnrSendCompleted :: !(IO ())
|
2021-03-31 13:39:01 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
data LiveAsyncActionQuery
|
|
|
|
= LAAQNoRelationships !LiveAsyncActionQueryWithNoRelationships
|
|
|
|
| LAAQOnSourceDB !LiveAsyncActionQueryOnSource
|
|
|
|
|
2021-09-24 01:56:37 +03:00
|
|
|
data AsyncActionQueryLive = AsyncActionQueryLive
|
|
|
|
{ _aaqlActionIds :: !(NonEmpty ActionId),
|
|
|
|
-- | An IO action to send error message (in case of any exception) to the websocket client
|
|
|
|
_aaqlOnException :: !(QErr -> IO ()),
|
|
|
|
_aaqlLiveExecution :: !LiveAsyncActionQuery
|
2021-03-31 13:39:01 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
-- | A share-able state map which stores an async action live query with it's subscription operation id
|
|
|
|
type AsyncActionSubscriptionState = TMap.TMap OperationId AsyncActionQueryLive
|
|
|
|
|
2021-09-24 01:56:37 +03:00
|
|
|
addAsyncActionLiveQuery ::
|
|
|
|
AsyncActionSubscriptionState ->
|
|
|
|
OperationId ->
|
|
|
|
NonEmpty ActionId ->
|
|
|
|
(QErr -> IO ()) ->
|
|
|
|
LiveAsyncActionQuery ->
|
|
|
|
IO ()
|
2021-03-31 13:39:01 +03:00
|
|
|
addAsyncActionLiveQuery queriesState opId actionIds onException liveQuery =
|
|
|
|
STM.atomically $
|
2021-09-24 01:56:37 +03:00
|
|
|
TMap.insert (AsyncActionQueryLive actionIds onException liveQuery) opId queriesState
|
2021-03-31 13:39:01 +03:00
|
|
|
|
2021-09-24 01:56:37 +03:00
|
|
|
removeAsyncActionLiveQuery ::
|
|
|
|
AsyncActionSubscriptionState -> OperationId -> IO ()
|
2021-03-31 13:39:01 +03:00
|
|
|
removeAsyncActionLiveQuery queriesState opId =
|
|
|
|
STM.atomically $ TMap.delete opId queriesState
|