Skip to content

feature/issue-566-suave #616

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
njlr wants to merge 5 commits into
base: dev
Choose a base branch
from
Open

feature/issue-566-suave #616

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ee8f99b
Adds Suave package and example
njlr Sep 23, 2026
c390637
Use `[<Struct>] ?` optional parameters and `voption` in executor midd…
xperiandri Oct 1, 2026
8891cce
Adds Suave package and example
njlr Sep 23, 2026
d74a147
Bumps to latest
njlr Oct 2, 2026
c23907a
Merge branch 'feature/issue-566-suave' of https://github.com/njlr/FSh…
njlr Oct 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Bumps to latest
  • Loading branch information
@njlr
njlr committed Oct 2, 2026
commit d74a14726a31bdcbe178441bf9babad42f90bf97
3 changes: 2 additions & 1 deletion src/FSharp.Data.GraphQL.Server.Suave/GraphQLHttpHandler.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,8 @@ let private jsonResponse (serializerOptions : JsonSerializerOptions) (statusCode

let private toResponse ({ DocumentId = documentId; Content = content } : GQLExecutionResult) =
match content with
| Direct (data, errs) -> GQLResponse.Direct (documentId, data, errs)
// `GQLResponse` carries a null `data` for a result whose non-null root field failed
| Direct (data, errs) -> GQLResponse.Direct (documentId, data |> ValueOption.toObj, errs)
| Deferred (data, errs, _deferred) -> GQLResponse.Direct (documentId, data, errs)
| Stream _stream -> GQLResponse.Stream documentId
| RequestError errs -> GQLResponse.RequestError (documentId, errs)
Expand Down
50 changes: 24 additions & 26 deletions src/FSharp.Data.GraphQL.Server.Suave/GraphQLSubscriptionsManagement.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,39 +1,37 @@
module internal FSharp.Data.GraphQL.Server.Suave.GraphQLSubscriptionsManagement

open System
open System.Collections.Generic

open FSharp.Data.GraphQL.Shared.WebSockets

/// The active subscriptions of one connection, keyed by the id the client gave each of them, with the handle that
/// unsubscribes from its source.
type SubscriptionsDict = Dictionary<SubscriptionId, IDisposable>

// `subscriptions` is mutated both from the WebSocket's sequential message loop and from the `IObserver` callbacks of
// each active subscription's stream, which can fire on an arbitrary thread. `IDictionary`'s backing `Dictionary` is
// not safe for concurrent access, so every operation here locks on the dictionary instance itself - the same
// instance is shared by every caller for a given connection, so this serializes all of them against each other.
let addSubscription
(id : SubscriptionId, unsubscriber : SubscriptionUnsubscriber, onUnsubscribe : OnUnsubscribeAction)
(subscriptions : SubscriptionsDict)
=
lock subscriptions (fun () -> subscriptions.Add (id, (unsubscriber, onUnsubscribe)))

let isIdTaken (id : SubscriptionId) (subscriptions : SubscriptionsDict) = lock subscriptions (fun () -> subscriptions.ContainsKey (id))

let executeOnUnsubscribeAndDispose (id : SubscriptionId) (subscription : SubscriptionUnsubscriber * OnUnsubscribeAction) =
match subscription with
| unsubscriber, onUnsubscribe ->
try
id |> onUnsubscribe
finally
unsubscriber.Dispose ()
// each active subscription's stream, which can fire on an arbitrary thread. `Dictionary` is not safe for concurrent
// access, so every operation here locks on the dictionary instance itself - the same instance is shared by every
// caller for a given connection, so this serializes all of them against each other.
let createSubscriptions () = SubscriptionsDict (StringComparer.Ordinal)

let addSubscription (id : SubscriptionId, unsubscriber : IDisposable) (subscriptions : SubscriptionsDict) =
lock subscriptions (fun () -> subscriptions.Add (id, unsubscriber))

let isIdTaken (id : SubscriptionId) (subscriptions : SubscriptionsDict) = lock subscriptions (fun () -> subscriptions.ContainsKey id)

let removeSubscription (id : SubscriptionId) (subscriptions : SubscriptionsDict) =
lock subscriptions (fun () ->
match subscriptions.TryGetValue id with
| true, sub ->
sub |> executeOnUnsubscribeAndDispose id
subscriptions.Remove (id) |> ignore
| true, unsubscriber ->
subscriptions.Remove id |> ignore
unsubscriber.Dispose ()
| false, _ -> ())

let removeAllSubscriptions (subscriptions : SubscriptionsDict) =
lock subscriptions (fun () ->
subscriptions
|> Seq.iter (fun subscription ->
subscription.Value
|> executeOnUnsubscribeAndDispose subscription.Key)
subscriptions.Clear ())
try
for unsubscriber in subscriptions.Values do
unsubscriber.Dispose ()
finally
subscriptions.Clear ())
79 changes: 50 additions & 29 deletions src/FSharp.Data.GraphQL.Server.Suave/GraphQLWebSocketHandler.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ let private serializeServerMessage (serializerOptions : JsonSerializerOptions) (
Payload = ValueSome (ExecutionResult payload)
}
| Complete id -> { Id = ValueSome id; Type = "complete"; Payload = ValueNone }
| Error (id, errMessages) -> {
| ServerError (id, errMessages) -> {
Id = ValueSome id
Type = "error"
Payload = ValueSome (ErrorMessages errMessages)
Expand Down Expand Up @@ -151,61 +151,82 @@ let private waitForConnectionInitAndRespondToClient (options : GraphQLOptions<'R
return Result.Error $"{nameof ConnectionInit} failed (not because of timeout)"
}

/// <summary>
/// Converts an exception raised by a subscription's source into the errors of its terminal <c>error</c> message,
/// exposing only messages of GraphQL-facing errors.
/// </summary>
let private problemDetailsOfSourceError (ex : exn) : GQLProblemDetails list =
match box ex with
| :? IGQLError as error -> [ GQLProblemDetails.OfError error ]
| _ -> [ GQLProblemDetails.Create "Unexpected error during subscription" ]

/// Runs the <c>graphql-transport-ws</c> message loop for an already-initialized connection.
let private handleMessages (options : GraphQLOptions<'Root>) (ctx : HttpContext) (webSocket : WebSocket) : Async<unit> =
let serializerOptions = options.SerializerOptions
let subscriptions = Dictionary<SubscriptionId, SubscriptionUnsubscriber * OnUnsubscribeAction>()
let subscriptions = GraphQLSubscriptionsManagement.createSubscriptions ()
let sendMsg msg = sendMessage serializerOptions webSocket msg

let sendOutput id (output : SubscriptionExecutionResult) = sendMsg (Next (id, output))

let sendSubscriptionResponseOutput id subscriptionResult =
match subscriptionResult with
| SubscriptionResult output -> sendOutput id { Data = ValueSome output; Errors = [] }
| SubscriptionErrors (_, errors) -> sendOutput id { Data = ValueNone; Errors = errors }

let sendDeferredResponseOutput id deferredResult =
match deferredResult with
| DeferredResult (data, _path) -> sendOutput id { Data = ValueSome (data :?> Output); Errors = [] }
| DeferredErrors (_, errors, _path) -> sendOutput id { Data = ValueNone; Errors = errors }

let sendDeferredResultDelayedBy (ms : int) id deferredResult = async {
do! Async.Sleep ms
do! sendDeferredResponseOutput id deferredResult
}
| SubscriptionResult output -> sendOutput id (SubscriptionExecutionResult.Create (ValueSome output, []))
// The executor may still have resolved partial data alongside the field errors; it is forwarded as-is
| SubscriptionErrors (ValueSome output, errors) -> sendOutput id (SubscriptionExecutionResult.Create (ValueSome output, errors))
| SubscriptionErrors (ValueNone, errors) -> sendOutput id (SubscriptionExecutionResult.CreateErrors errors)

let addClientSubscription
(id : SubscriptionId)
(howToSendDataOnNext : SubscriptionId -> 'ResponseContent -> Async<unit>)
(streamSource : IObservable<'ResponseContent>)
=
// Set once the source has ended, so that a source ending synchronously inside `Subscribe` does not leave its id
// registered after the end has already tried to remove it
let mutable ended = false

let endSubscription (message : ServerMessage) =
ended <- true
sendMsg message |> Async.RunSynchronously
subscriptions
|> GraphQLSubscriptionsManagement.removeSubscription id

let observer = {
new IObserver<'ResponseContent> with
member _.OnNext value = howToSendDataOnNext id value |> Async.RunSynchronously
member _.OnError ex =
ctx.runtime.logger.info (Suave.Logging.Message.eventX $"Error on subscription with Id = '{id}': {ex}")
member _.OnCompleted () =
sendMsg (Complete id) |> Async.RunSynchronously
subscriptions
|> GraphQLSubscriptionsManagement.removeSubscription id
endSubscription (ServerError (id, problemDetailsOfSourceError ex))
member _.OnCompleted () = endSubscription (Complete id)
}

let unsubscriber = streamSource.Subscribe observer
subscriptions
|> GraphQLSubscriptionsManagement.addSubscription (id, unsubscriber, ignore)
lock subscriptions (fun () ->
if ended then
unsubscriber.Dispose ()
else
subscriptions
|> GraphQLSubscriptionsManagement.addSubscription (id, unsubscriber))

let applyPlanExecutionResult (id : SubscriptionId) (executionResult : GQLExecutionResult) = async {
match executionResult with
match executionResult.Content with
| Stream observableOutput ->
observableOutput
|> addClientSubscription id sendSubscriptionResponseOutput
| Deferred (data, errors, observableOutput) ->
do! sendOutput id { Data = ValueSome data; Errors = [] }
if errors.IsEmpty then
observableOutput
|> addClientSubscription id (sendDeferredResultDelayedBy 5000)
| Direct (data, _) -> do! sendOutput id { Data = ValueSome data; Errors = [] }
| RequestError problemDetails -> do! sendOutput id { Data = ValueNone; Errors = problemDetails }
| Deferred (data, errors, _deferred) ->
// TODO: deliver the deferred and streamed payloads in the incremental delivery format, as the ASP.NET Core
// integration does through its `IncrementalDelivery`; until then only the initial result is sent
do! sendOutput id (SubscriptionExecutionResult.Create (ValueSome data, errors))
do! sendMsg (Complete id)
| Direct (data, errors) ->
// An execution result, whose data is null when a non-null root field failed during execution; still a
// result, so it is sent as Next + Complete like any other, not as the terminal Error
do! sendOutput id (SubscriptionExecutionResult.Create (data, errors))
// The graphql-transport-ws protocol requires Complete after the single Next of a query or mutation
do! sendMsg (Complete id)
| RequestError problemDetails ->
// The request was rejected before execution, so it is not a result: the protocol requires it to be sent
// as the terminal Error message instead of a Next followed by Complete
do! sendMsg (ServerError (id, problemDetails))
}

let handleSubscribe (id : SubscriptionId) (query : GQLRequestContent) = async {
Expand All @@ -221,7 +242,7 @@ let private handleMessages (options : GraphQLOptions<'Root>) (ctx : HttpContext)
do! applyPlanExecutionResult id planExecutionResult
with ex ->
ctx.runtime.logger.info (Suave.Logging.Message.eventX $"Unexpected error during subscription with id '{id}': {ex}")
do! sendMsg (Error (id, [ NameValueLookup ([ ("subscription", "Unexpected error during subscription" :> obj) ]) ]))
do! sendMsg (ServerError (id, [ GQLProblemDetails.Create "Unexpected error during subscription" ]))
}

let rec loop () = async {
Expand Down