-
-
Notifications
You must be signed in to change notification settings - Fork 922
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.
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
Support subscriptions of value types #3876
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -70,6 +70,14 @@ public ChatSubscriptions(IChat chat) | |
Type = typeof(StringGraphType), | ||
StreamResolver = new SourceStreamResolver<string>(context => Subscribe(context).Select(message => message.Content)) | ||
}); | ||
|
||
int counter = 0; | ||
AddField(new FieldType | ||
{ | ||
Name = "messageCounter", | ||
Type = typeof(IntGraphType), | ||
StreamResolver = new SourceStreamResolver<int>(context => Subscribe(context).Select(_ => ++counter)) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Threw exception here previously |
||
}); | ||
} | ||
|
||
private IObservable<Message> SubscribeById(IResolveFieldContext context) | ||
|
@@ -145,7 +153,7 @@ public MessageInputType() | |
{ | ||
Field<StringGraphType>("fromId"); | ||
Field<StringGraphType>("content"); | ||
Field<DateGraphType>("sentAt"); | ||
Field<DateTimeOffsetGraphType>("sentAt"); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. So the serialized responses would be repeatable as they would carry timezone info which can be set to UTC. |
||
} | ||
} | ||
|
||
|
@@ -164,7 +172,7 @@ public class Message | |
|
||
public string Content { get; set; } | ||
|
||
public DateTime SentAt { get; set; } | ||
public DateTimeOffset SentAt { get; set; } | ||
} | ||
|
||
public class MessageFrom | ||
|
@@ -180,7 +188,7 @@ public class ReceivedMessage | |
|
||
public string Content { get; set; } | ||
|
||
public DateTime SentAt { get; set; } | ||
public DateTimeOffset SentAt { get; set; } | ||
} | ||
|
||
public interface IChat | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -5,6 +5,8 @@ namespace GraphQL.Tests.Subscription; | |
|
||
public class SubscriptionTests | ||
{ | ||
private readonly DateTimeOffset DateConst = new DateTimeOffset(2024, 1, 1, 0, 0, 0, TimeSpan.Zero); | ||
|
||
protected async Task<ExecutionResult> ExecuteSubscribeAsync(ExecutionOptions options) | ||
{ | ||
var executer = new DocumentExecuter(); | ||
|
@@ -28,7 +30,7 @@ public async Task SubscribeGetAll() | |
DisplayName = "test", | ||
Id = "1" | ||
}, | ||
SentAt = DateTime.Now.Date | ||
SentAt = DateConst | ||
}; | ||
|
||
var chat = new Chat(); | ||
|
@@ -67,7 +69,7 @@ public async Task SubscribeToContent(bool useMiddleware) | |
DisplayName = "test", | ||
Id = "1" | ||
}, | ||
SentAt = DateTime.Now.Date | ||
SentAt = DateConst | ||
}; | ||
|
||
var chat = new Chat(); | ||
|
@@ -112,7 +114,7 @@ public async Task Subscribe() | |
DisplayName = "test", | ||
Id = "1" | ||
}, | ||
SentAt = DateTime.Now.Date | ||
SentAt = DateConst | ||
}; | ||
var chat = new Chat(); | ||
var schema = new ChatSchema(chat); | ||
|
@@ -127,13 +129,47 @@ public async Task Subscribe() | |
chat.AddMessage(addedMessage); | ||
|
||
/* Then */ | ||
var stream = result.Streams!.Values.First(); | ||
var stream = result.Streams.ShouldNotBeNull().Values.First(); | ||
var message = await stream.FirstOrDefaultAsync(); | ||
|
||
message.ShouldNotBeNull(); | ||
message.ShouldBeOfType<ExecutionResult>(); | ||
message.Data.ShouldNotBeNull(); | ||
message.Data.ShouldNotBeAssignableTo<Task>(); | ||
message.ShouldBeSimilarTo(""" | ||
{"data":{"messageAdded":{"from":{"id":"1","displayName":"test"},"content":"test","sentAt":"2024-01-01T00:00:00\u002B00:00"}}} | ||
"""); | ||
} | ||
|
||
[Fact] | ||
public async Task SubscribeInt() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Tests in this case direct usage of SourceStreamResolver, which is what all the field builders use. |
||
{ | ||
/* Given */ | ||
var addedMessage = new Message | ||
{ | ||
Content = "test", | ||
From = new MessageFrom | ||
{ | ||
DisplayName = "test", | ||
Id = "1" | ||
}, | ||
SentAt = DateConst | ||
}; | ||
var chat = new Chat(); | ||
var schema = new ChatSchema(chat); | ||
|
||
/* When */ | ||
var result = await ExecuteSubscribeAsync(new ExecutionOptions | ||
{ | ||
Query = "subscription { messageCounter }", | ||
Schema = schema | ||
}); | ||
|
||
chat.AddMessage(addedMessage); | ||
|
||
/* Then */ | ||
var stream = result.Streams.ShouldNotBeNull().Values.First(); | ||
var message = await stream.FirstOrDefaultAsync(); | ||
|
||
message.ShouldBeSimilarTo(""" | ||
{"data":{"messageCounter":1}} | ||
"""); | ||
} | ||
|
||
[Fact] | ||
|
@@ -148,7 +184,7 @@ public async Task SubscribeAsync() | |
DisplayName = "test", | ||
Id = "1" | ||
}, | ||
SentAt = DateTime.Now.Date | ||
SentAt = DateConst | ||
}; | ||
var chat = new Chat(); | ||
var schema = new ChatSchema(chat); | ||
|
@@ -163,13 +199,12 @@ public async Task SubscribeAsync() | |
chat.AddMessage(addedMessage); | ||
|
||
/* Then */ | ||
var stream = result.Streams!.Values.First(); | ||
var stream = result.Streams.ShouldNotBeNull().Values.First(); | ||
var message = await stream.FirstOrDefaultAsync(); | ||
|
||
message.ShouldNotBeNull(); | ||
message.ShouldBeOfType<ExecutionResult>(); | ||
message.Data.ShouldNotBeNull(); | ||
message.Data.ShouldNotBeAssignableTo<Task>(); | ||
message.ShouldBeSimilarTo(""" | ||
{"data":{"messageAddedAsync":{"from":{"id":"1","displayName":"test"},"content":"test","sentAt":"2024-01-01T00:00:00\u002B00:00"}}} | ||
"""); | ||
} | ||
|
||
[Fact] | ||
|
@@ -184,7 +219,7 @@ public async Task SubscribeWithArgument() | |
DisplayName = "test", | ||
Id = "1" | ||
}, | ||
SentAt = DateTime.Now.Date | ||
SentAt = DateConst | ||
}; | ||
var chat = new Chat(); | ||
var schema = new ChatSchema(chat); | ||
|
@@ -203,12 +238,12 @@ public async Task SubscribeWithArgument() | |
chat.AddMessage(addedMessage); | ||
|
||
/* Then */ | ||
var stream = result.Streams!.Values.First(); | ||
var stream = result.Streams.ShouldNotBeNull().Values.First(); | ||
var message = await stream.FirstOrDefaultAsync(); | ||
|
||
message.ShouldNotBeNull(); | ||
message.ShouldBeOfType<ExecutionResult>(); | ||
message.Data.ShouldNotBeNull(); | ||
message.ShouldBeSimilarTo(""" | ||
{"data":{"messageAddedByUser":{"from":{"id":"1","displayName":"test"},"content":"test","sentAt":"2024-01-01T00:00:00\u002B00:00"}}} | ||
"""); | ||
} | ||
|
||
[Fact] | ||
|
@@ -223,7 +258,7 @@ public async Task SubscribeWithArgumentAsync() | |
DisplayName = "test", | ||
Id = "1" | ||
}, | ||
SentAt = DateTime.Now.Date | ||
SentAt = DateConst | ||
}; | ||
var chat = new Chat(); | ||
var schema = new ChatSchema(chat); | ||
|
@@ -242,12 +277,12 @@ public async Task SubscribeWithArgumentAsync() | |
chat.AddMessage(addedMessage); | ||
|
||
/* Then */ | ||
var stream = result.Streams!.Values.First(); | ||
var stream = result.Streams.ShouldNotBeNull().Values.First(); | ||
var message = await stream.FirstOrDefaultAsync(); | ||
|
||
message.ShouldNotBeNull(); | ||
message.ShouldBeOfType<ExecutionResult>(); | ||
message.Data.ShouldNotBeNull(); | ||
message.ShouldBeSimilarTo(""" | ||
{"data":{"messageAddedByUserAsync":{"from":{"id":"1","displayName":"test"},"content":"test","sentAt":"2024-01-01T00:00:00\u002B00:00"}}} | ||
"""); | ||
} | ||
|
||
[Fact] | ||
|
@@ -267,7 +302,7 @@ public async Task OnError() | |
chat.AddError(new Exception("test")); | ||
|
||
/* Then */ | ||
var stream = result.Streams!.Values.First(); | ||
var stream = result.Streams.ShouldNotBeNull().Values.First(); | ||
var error = await Should.ThrowAsync<ExecutionError>(async () => await stream.FirstOrDefaultAsync()); | ||
error.InnerException!.Message.ShouldBe("test"); | ||
error.Path.ShouldBe(new[] { "messageAdded" }); | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,28 @@ | ||
namespace GraphQL.Resolvers; | ||
|
||
/// <summary> | ||
/// Converts an <see cref="IObservable{T}"/> for value types into an <see cref="IObservable{T}">IObservable<object?></see>. | ||
/// </summary> | ||
internal sealed class ObservableAdapter<T> : IObservable<object?> | ||
{ | ||
private readonly IObservable<T> _observable; | ||
|
||
public ObservableAdapter(IObservable<T> observable) | ||
{ | ||
_observable = observable; | ||
} | ||
|
||
public IDisposable Subscribe(IObserver<object?> observer) => _observable.Subscribe(new ObserverAdapter(observer)); | ||
|
||
private sealed class ObserverAdapter : IObserver<T> | ||
{ | ||
private readonly IObserver<object?> _observer; | ||
public ObserverAdapter(IObserver<object?> observer) | ||
{ | ||
_observer = observer; | ||
} | ||
public void OnCompleted() => _observer.OnCompleted(); | ||
public void OnError(Exception error) => _observer.OnError(error); | ||
public void OnNext(T value) => _observer.OnNext(value); // note: boxing here | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -16,9 +16,13 @@ | |
throw new ArgumentNullException(nameof(sourceStreamResolver)); | ||
|
||
if (typeof(TReturnType).IsValueType) | ||
throw new InvalidOperationException("The generic type TReturnType must not be a value type."); | ||
|
||
_sourceStreamResolver = context => new ValueTask<IObservable<object?>>((IObservable<object?>)sourceStreamResolver(context)); | ||
{ | ||
_sourceStreamResolver = context => new(new ObservableAdapter<TReturnType?>(sourceStreamResolver(context))); | ||
} | ||
else | ||
{ | ||
_sourceStreamResolver = context => new((IObservable<object?>)sourceStreamResolver(context)); | ||
} | ||
Comment on lines
18
to
+25
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The changes in this file are the only "real" changes in the project |
||
} | ||
|
||
/// <inheritdoc cref="SourceStreamResolver{TReturnType}(Func{IResolveFieldContext, IObservable{TReturnType}})"/> | ||
|
@@ -28,9 +32,13 @@ | |
throw new ArgumentNullException(nameof(sourceStreamResolver)); | ||
|
||
if (typeof(TReturnType).IsValueType) | ||
throw new InvalidOperationException("The generic type TReturnType must not be a value type."); | ||
|
||
_sourceStreamResolver = async context => (IObservable<object?>)await sourceStreamResolver(context).ConfigureAwait(false); | ||
{ | ||
_sourceStreamResolver = async context => new ObservableAdapter<TReturnType?>(await sourceStreamResolver(context).ConfigureAwait(false)); | ||
} | ||
else | ||
{ | ||
_sourceStreamResolver = async context => (IObservable<object?>)await sourceStreamResolver(context).ConfigureAwait(false); | ||
} | ||
} | ||
|
||
/// <inheritdoc/> | ||
|
@@ -50,21 +58,31 @@ | |
throw new ArgumentNullException(nameof(sourceStreamResolver)); | ||
|
||
if (typeof(TReturnType).IsValueType) | ||
throw new InvalidOperationException("The generic type TReturnType must not be a value type."); | ||
|
||
_sourceStreamResolver = context => new ValueTask<IObservable<object?>>((IObservable<object?>)sourceStreamResolver(context.As<TSourceType>())); | ||
{ | ||
_sourceStreamResolver = context => new(new ObservableAdapter<TReturnType?>(sourceStreamResolver(context.As<TSourceType>()))); | ||
} | ||
else | ||
{ | ||
_sourceStreamResolver = context => new((IObservable<object?>)sourceStreamResolver(context.As<TSourceType>())); | ||
} | ||
} | ||
|
||
/// <inheritdoc cref="SourceStreamResolver{TSourceType, TReturnType}(Func{IResolveFieldContext{TSourceType}, IObservable{TReturnType}})"/> | ||
/// | ||
public SourceStreamResolver(Func<IResolveFieldContext<TSourceType>, ValueTask<IObservable<TReturnType?>>> sourceStreamResolver) | ||
{ | ||
if (sourceStreamResolver == null) | ||
throw new ArgumentNullException(nameof(sourceStreamResolver)); | ||
|
||
if (typeof(TReturnType).IsValueType) | ||
throw new InvalidOperationException("The generic type TReturnType must not be a value type."); | ||
|
||
_sourceStreamResolver = async context => (IObservable<object?>)await sourceStreamResolver(context.As<TSourceType>()).ConfigureAwait(false); | ||
if (typeof(TReturnType).IsValueType) | ||
{ | ||
_sourceStreamResolver = async context => new ObservableAdapter<TReturnType?>(await sourceStreamResolver(context.As<TSourceType>()).ConfigureAwait(false)); | ||
} | ||
else | ||
{ | ||
_sourceStreamResolver = async context => (IObservable<object?>)await sourceStreamResolver(context.As<TSourceType>()).ConfigureAwait(false); | ||
} | ||
Comment on lines
+78
to
+85
Check notice Code scanning / CodeQL Missed ternary opportunity Note
Both branches of this 'if' statement write to the same variable - consider using '?' to express intent better.
|
||
} | ||
|
||
/// <inheritdoc/> | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This tests the auto-registering types and worked without changes.