Safe Haskell	Safe-Inferred
Language	Haskell2010

Hasura.GraphQL.Execute.Subscription.State

Description

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 details.

Synopsis

data SubscriptionsState = SubscriptionsState {
}
initSubscriptionsState :: SubscriptionPostPollHook -> IO SubscriptionsState
dumpSubscriptionsState :: Bool -> LiveQueriesOptions -> StreamQueriesOptions -> SubscriptionsState -> IO Value
data SubscriberDetails a
type LiveQuerySubscriberDetails = SubscriberDetails CohortKey
type StreamingSubscriberDetails = SubscriberDetails (CohortKey, TVar CursorVariableValues)
addLiveQuery :: forall b. BackendTransport b => Logger Hasura -> ServerMetrics -> PrometheusMetrics -> SubscriberMetadata -> SubscriptionsState -> IO (LiveQueriesOptions, StreamQueriesOptions) -> SourceName -> ParameterizedQueryHash -> Maybe OperationName -> RequestId -> SubscriptionQueryPlan b (MultiplexedQuery b) -> IO GranularPrometheusMetricsState -> OnChange -> Maybe (Endo Value) -> IO LiveQuerySubscriberDetails
addStreamSubscriptionQuery :: forall b. BackendTransport b => Logger Hasura -> ServerMetrics -> PrometheusMetrics -> SubscriberMetadata -> SubscriptionsState -> IO (LiveQueriesOptions, StreamQueriesOptions) -> SourceName -> ParameterizedQueryHash -> Maybe OperationName -> RequestId -> Name -> SubscriptionQueryPlan b (MultiplexedQuery b) -> IO GranularPrometheusMetricsState -> OnChange -> Maybe (Endo Value) -> IO StreamingSubscriberDetails
removeLiveQuery :: Logger Hasura -> ServerMetrics -> PrometheusMetrics -> SubscriptionsState -> LiveQuerySubscriberDetails -> IO GranularPrometheusMetricsState -> Maybe OperationName -> IO ()
removeStreamingQuery :: Logger Hasura -> ServerMetrics -> PrometheusMetrics -> SubscriptionsState -> StreamingSubscriberDetails -> IO GranularPrometheusMetricsState -> Maybe OperationName -> IO ()
data LiveAsyncActionQueryOnSource = LiveAsyncActionQueryOnSource {
}
data LiveAsyncActionQueryWithNoRelationships = LiveAsyncActionQueryWithNoRelationships {
- _laaqwnrSendResponse :: !(ActionLogResponseMap -> IO ())
- _laaqwnrSendCompleted :: !(IO ())
}
data LiveAsyncActionQuery
- = LAAQNoRelationships LiveAsyncActionQueryWithNoRelationships
- | LAAQOnSourceDB LiveAsyncActionQueryOnSource
data AsyncActionQueryLive = AsyncActionQueryLive {
- _aaqlActionIds :: NonEmpty ActionId
- _aaqlOnException :: QErr -> IO ()
- _aaqlLiveExecution :: LiveAsyncActionQuery
}
type AsyncActionSubscriptionState = TMap OperationId AsyncActionQueryLive
addAsyncActionLiveQuery :: AsyncActionSubscriptionState -> OperationId -> NonEmpty ActionId -> (QErr -> IO ()) -> LiveAsyncActionQuery -> IO ()
removeAsyncActionLiveQuery :: AsyncActionSubscriptionState -> OperationId -> IO ()

Documentation

data SubscriptionsState Source #

The top-level datatype that holds the state for all active subscriptions.

NOTE!: This must be kept consistent with a websocket connection's OperationMap, in onClose and onStart.

Constructors

SubscriptionsState
Fields _ssLiveQueryMap :: PollerMap () _ssStreamQueryMap :: PollerMap (TVar CursorVariableValues) _ssPostPollHook :: SubscriptionPostPollHook A hook function which is run after each fetch cycle _ssAsyncActions :: AsyncActionSubscriptionState

initSubscriptionsState :: SubscriptionPostPollHook -> IO SubscriptionsState Source #

dumpSubscriptionsState :: Bool -> LiveQueriesOptions -> StreamQueriesOptions -> SubscriptionsState -> IO Value Source #

For dev debugging, output subject to change.

data SubscriberDetails a Source #

SubscriberDetails contains the data required to locate a subscriber in the correct cohort within the correct poller in the operation map.

Instances

Instances details

Show a => Show (SubscriberDetails a) Source #
Instance details Defined in Hasura.GraphQL.Execute.Subscription.State Methods showsPrec :: Int -> SubscriberDetails a -> ShowS # show :: SubscriberDetails a -> String # showList :: [SubscriberDetails a] -> ShowS #

type LiveQuerySubscriberDetails = SubscriberDetails CohortKey Source #

type StreamingSubscriberDetails = SubscriberDetails (CohortKey, TVar CursorVariableValues) Source #

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.

addLiveQuery Source #

Arguments

:: forall b. BackendTransport b
=> Logger Hasura
-> ServerMetrics
-> PrometheusMetrics
-> SubscriberMetadata
-> SubscriptionsState
-> IO (LiveQueriesOptions, StreamQueriesOptions)
-> SourceName
-> ParameterizedQueryHash
-> Maybe OperationName	operation name of the query
-> RequestId
-> SubscriptionQueryPlan b (MultiplexedQuery b)
-> IO GranularPrometheusMetricsState
-> OnChange	the action to be executed when result changes
-> Maybe (Endo Value)
-> IO LiveQuerySubscriberDetails

Fork a thread handling a regular (live query) subscription

addStreamSubscriptionQuery Source #

Arguments

:: forall b. BackendTransport b
=> Logger Hasura
-> ServerMetrics
-> PrometheusMetrics
-> SubscriberMetadata
-> SubscriptionsState
-> IO (LiveQueriesOptions, StreamQueriesOptions)
-> SourceName
-> ParameterizedQueryHash
-> Maybe OperationName	operation name of the query
-> RequestId
-> Name	root field name
-> SubscriptionQueryPlan b (MultiplexedQuery b)
-> IO GranularPrometheusMetricsState
-> OnChange	the action to be executed when result changes
-> Maybe (Endo Value)	the modifier for adding typename for namespaced queries
-> IO StreamingSubscriberDetails

Fork a thread handling a streaming subscription

removeLiveQuery :: Logger Hasura -> ServerMetrics -> PrometheusMetrics -> SubscriptionsState -> LiveQuerySubscriberDetails -> IO GranularPrometheusMetricsState -> Maybe OperationName -> IO () Source #

removeStreamingQuery :: Logger Hasura -> ServerMetrics -> PrometheusMetrics -> SubscriptionsState -> StreamingSubscriberDetails -> IO GranularPrometheusMetricsState -> Maybe OperationName -> IO () Source #

data LiveAsyncActionQueryOnSource Source #

An async action query whose relationships are referred 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]

Constructors

LiveAsyncActionQueryOnSource

Fields

_laaqpCurrentLqId :: !LiveQuerySubscriberDetails
_laaqpPrevActionLogMap :: !ActionLogResponseMap
_laaqpRestartLq :: !(LiveQuerySubscriberDetails -> ActionLogResponseMap -> IO (Maybe LiveQuerySubscriberDetails))
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.

data LiveAsyncActionQueryWithNoRelationships Source #

Constructors

LiveAsyncActionQueryWithNoRelationships
Fields _laaqwnrSendResponse :: !(ActionLogResponseMap -> IO ()) An IO action to send response to the websocket client _laaqwnrSendCompleted :: !(IO ()) An IO action to send "completed" message to the websocket client

data LiveAsyncActionQuery Source #

Constructors

LAAQNoRelationships LiveAsyncActionQueryWithNoRelationships
LAAQOnSourceDB LiveAsyncActionQueryOnSource

data AsyncActionQueryLive Source #

Constructors

AsyncActionQueryLive
Fields _aaqlActionIds :: NonEmpty ActionId _aaqlOnException :: QErr -> IO () An IO action to send error message (in case of any exception) to the websocket client _aaqlLiveExecution :: LiveAsyncActionQuery

type AsyncActionSubscriptionState = TMap OperationId AsyncActionQueryLive Source #

A share-able state map which stores an async action live query with it's subscription operation id

addAsyncActionLiveQuery :: AsyncActionSubscriptionState -> OperationId -> NonEmpty ActionId -> (QErr -> IO ()) -> LiveAsyncActionQuery -> IO () Source #

removeAsyncActionLiveQuery :: AsyncActionSubscriptionState -> OperationId -> IO () Source #