Skip to content
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

Add support for SSE-based subscriptions for caliban-quick #2141

Merged
merged 5 commits into from
Feb 29, 2024
Merged

Add support for SSE-based subscriptions for caliban-quick #2141

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c15748a
Add support for SSE-based subscriptions in the quick adapter
kyri-petrou Feb 26, 2024
7848c5f
Cleanup `AcceptsGqlEncodings`
kyri-petrou Feb 26, 2024
1cf019d
One more cleanup
kyri-petrou Feb 26, 2024
b30ccba
Always send a 200 code for SSE (Tapir)
kyri-petrou Feb 26, 2024
6b0ed9d
Unify SSE response handling
kyri-petrou Feb 27, 2024
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
32 changes: 22 additions & 10 deletions adapters/quick/src/main/scala/caliban/QuickRequestHandler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,8 @@ package caliban

import caliban.Configurator.ExecutionConfiguration
import caliban.HttpUtils.DeferMultipart
import caliban.ResponseValue.StreamValue
import caliban.ResponseValue.{ ObjectValue, StreamValue }
import caliban.Value.NullValue
import caliban.interop.jsoniter.ValueJsoniter
import caliban.uploads.{ FileMeta, GraphQLUploadRequest, Uploads }
import caliban.wrappers.Caching
Expand Down Expand Up @@ -148,13 +149,7 @@ final private class QuickRequestHandler[-R](interpreter: GraphQLInterpreter[R, A
cacheDirective.fold(headers)(headers.addHeader(Header.CacheControl.name, _))

private def transformResponse(httpReq: Request, resp: GraphQLResponse[Any])(implicit trace: Trace): Response = {

val acceptsGqlJson: Boolean =
httpReq.headers.get(Header.Accept.name).exists { h =>
// Better performance than having to parse the Accept header
h.length >= 33 && h.toLowerCase.contains("application/graphql-response+json")
}

val accepts = new HttpUtils.AcceptsGqlEncodings(httpReq.headers.get(Header.Accept.name))
val cacheDirective = HttpUtils.computeCacheDirective(resp.extensions)

resp match {
Expand All @@ -164,7 +159,9 @@ final private class QuickRequestHandler[-R](interpreter: GraphQLInterpreter[R, A
headers = responseHeaders(ContentTypeMultipart, None),
body = Body.fromStream(encodeMultipartMixedResponse(resp, stream))
)
case resp if acceptsGqlJson =>
case resp if accepts.serverSentEvents =>
Response.fromServerSentEvents(encodeTextEventStream(resp))
case resp if accepts.graphQLJson =>
Response(
status = resp.errors.collectFirst { case _: CalibanError.ParsingError | _: CalibanError.ValidationError =>
Status.BadRequest
Expand Down Expand Up @@ -205,11 +202,24 @@ final private class QuickRequestHandler[-R](interpreter: GraphQLInterpreter[R, A
.mapConcatChunk(Chunk.fromArray)
}

private def encodeTextEventStream(resp: GraphQLResponse[Any])(implicit trace: Trace) = {
val stream = (resp.data match {
case ObjectValue((fieldName, StreamValue(stream)) :: Nil) =>
stream.either.map {
case Right(r) => GraphQLResponse(ObjectValue(List(fieldName -> r)), resp.errors)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Isn't this saying: for all response bodies return the original errors? Would we want errors to be null here perhaps?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm I think you're right. It's probably better to send an initial one with all the errors (if they exist) and then the subsequent ones not containing errors

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I made the the GraphQLResponse stream handling generic and placed it in HttpUtils in the core module so that both the caliban-quick and tapir adapters use the same one. Also the errors now are reported in the first event only

case Left(err) => GraphQLResponse(ObjectValue(List(fieldName -> NullValue)), List(err))
}
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@paulpdaniels Is this error handling sound? Also, is there a better way to merge the error and success channels? I couldn't for the life of me figure it out 😕

case _ =>
ZStream.succeed(resp)
}).map(resp => ServerSentEvent(writeToString(resp.toResponseValue), eventType = Some("next")))

stream ++ ZStream.succeed(CompleteSse)
}

private def isFtv1Request(req: Request) =
req.headers
.get(GraphQLRequest.`apollo-federation-include-trace`)
.exists(_.equalsIgnoreCase(GraphQLRequest.ftv1))

}

object QuickRequestHandler {
Expand All @@ -225,6 +235,8 @@ object QuickRequestHandler {
private val ContentTypeMultipart =
Headers(Header.ContentType(MediaType.multipart.mixed.copy(parameters = DeferMultipart.DeferHeaderParams)).untyped)

private val CompleteSse = ServerSentEvent("", Some("complete"))

private val BodyDecodeErrorResponse =
badRequest("Failed to decode json body")

Expand Down
3 changes: 1 addition & 2 deletions adapters/quick/src/test/scala/caliban/QuickAdapterSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,7 @@ object QuickAdapterSpec extends ZIOSpecDefault {
"QuickAdapterSpec",
uri"http://localhost:8090/api/graphql",
wsUri = None,
uploadUri = Some(uri"http://localhost:8090/upload/graphql"),
sseSupport = None
uploadUri = Some(uri"http://localhost:8090/upload/graphql")
)
suite.provideShared(
apiLayer,
Expand Down
18 changes: 16 additions & 2 deletions core/src/main/scala/caliban/HttpUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,20 @@ private[caliban] object HttpUtils {
extensions
.flatMap(_.fields.collectFirst { case (Caching.DirectiveName, ResponseValue.ObjectValue(fields)) =>
fields.collectFirst { case ("httpHeader", Value.StringValue(cacheHeader)) => cacheHeader }
})
.flatten
}.flatten)

final class AcceptsGqlEncodings(header0: Option[String]) {
private val isEmpty = header0.isEmpty
private val length: Int = if (isEmpty) 0 else header0.get.length
private lazy val header = if (isEmpty) "" else header0.get.toLowerCase

/**
* NOTE: From 1st January 2025 this should be changed to `true` as the default
*
* @see [[https://graphql.github.io/graphql-over-http/draft/#sec-Legacy-watershed]]
*/
def graphQLJson: Boolean = length >= 33 && header.contains("application/graphql-response+json")

def serverSentEvents: Boolean = length >= 17 && header.contains("text/event-stream")
}
}
23 changes: 3 additions & 20 deletions interop/tapir/src/main/scala/caliban/interop/tapir/TapirAdapter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,24 +106,7 @@ object TapirAdapter {
streamConstructor: StreamConstructor[BS],
responseCodec: JsonCodec[ResponseValue]
): (MediaType, StatusCode, Option[String], CalibanBody[BS]) = {

/**
* NOTE: From 1st January 2025 this logic should be changed to use `application/graphql-response+json` as the
* default content-type when the client does not specify an accept header.
*
* @see [[https://graphql.github.io/graphql-over-http/draft/#sec-Legacy-watershed]]
*/
def accepts = request.acceptsContentTypes
.fold(
_ => None,
_.find {
case ContentTypeRange("application", "graphql-response+json", _, _) => true
case ContentTypeRange("text", "event-stream", _, _) => true
case _ => false
}
)
.map(ct => MediaType(ct.mainType, ct.subType))
.getOrElse(MediaType.ApplicationJson)
val accepts = new HttpUtils.AcceptsGqlEncodings(request.header(HeaderNames.Accept))

response match {
case resp @ GraphQLResponse(StreamValue(stream), _, _, _) =>
Expand All @@ -133,7 +116,7 @@ object TapirAdapter {
None,
encodeMultipartMixedResponse(resp, stream)
)
case resp if accepts == GraphqlResponseJson.mediaType =>
case resp if accepts.graphQLJson =>
val code =
response.errors.collectFirst { case _: CalibanError.ParsingError | _: CalibanError.ValidationError =>
StatusCode.BadRequest
Expand All @@ -149,7 +132,7 @@ object TapirAdapter {
excludeExtensions = cacheDirective.map(_ => Set(Caching.DirectiveName))
)
)
case resp if accepts == GraphqlServerSentEvent.mediaType =>
case resp if accepts.serverSentEvents =>
val code = response.errors.collectFirst { case HttpRequestMethod.MutationOverGetError => StatusCode.BadRequest }
.getOrElse(StatusCode.Ok)
(
Expand Down
8 changes: 4 additions & 4 deletions interop/tapir/src/test/scala/caliban/interop/tapir/TapirAdapterSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ object TapirAdapterSpec {
httpUri: Uri,
uploadUri: Option[Uri] = None,
wsUri: Option[Uri] = None,
sseSupport: Option[Boolean] = Some(true)
sseSupport: Boolean = true
)(implicit
requestCodec: JsonCodec[GraphQLRequest],
responseCodec: JsonCodec[GraphQLResponse[CalibanError]],
Expand Down Expand Up @@ -269,8 +269,8 @@ object TapirAdapterSpec {
}
)
),
sseSupport.map(_ =>
suite("SSE")(
Some(
suite("server-sent events")(
test("TextEventStream") {
for {
res <- runSSERequest(
Expand All @@ -290,7 +290,7 @@ object TapirAdapterSpec {
)
)
} @@ TestAspect.timeout(10.seconds)
)
).when(sseSupport)
),
runUpload.map(runUpload =>
suite("uploads")(
Expand Down