-
Notifications
You must be signed in to change notification settings - Fork 55
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
JetStream pull consumer redesign (#115)
* JetStream pull consumer redesign * Consume and fetch all * Max bytes * Header parser multi span bug fix * Msg size fix * Fixed tests and async sync bug * Fixed tests * Fixed tests * Fixed tests * Fixed tests * Trace test issue * Fixed warnings * Removed flappy no-ACK test This test was checking if the server was resending a message which was not ACKed, after the ack_wait times out. Resending unacknowledged messages is a server function we could test under more controlled test setup and it's this part of the test isn't testing NATS NET codebase. For some reason resend function isn't working most of the time when run under the GitHub Actions environment and creating false positives. Hence I decided to remove this part of the test to avoid unnecessary noise for the time being. * End fetch with internal clock This is a case where server request timeout might not reach the client because of network anomalies. * Default to fast ACK * Options -> Opts refactor * Ignore extra messages on consumer dispose * Propagate JS options for ACK * Test asserts * Flappy test trace * Public API fixes * JSMsg hides the NatsMsg and proxy relevant fields * Moved RawData to example * JSContext NewInbox is now internal
- Loading branch information
Showing
61 changed files
with
1,325 additions
and
1,034 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
14 changes: 14 additions & 0 deletions
14
sandbox/Example.JetStream.PullConsumer/Example.JetStream.PullConsumer.csproj
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <OutputType>Exe</OutputType> | ||
| <TargetFramework>net6.0</TargetFramework> | ||
| <ImplicitUsings>enable</ImplicitUsings> | ||
| <Nullable>enable</Nullable> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\..\src\NATS.Client.JetStream\NATS.Client.JetStream.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,145 @@ | ||
| using System.Diagnostics; | ||
| using Example.JetStream.PullConsumer; | ||
| using Microsoft.Extensions.Logging; | ||
| using NATS.Client.Core; | ||
| using NATS.Client.JetStream; | ||
|
|
||
| var cts = new CancellationTokenSource(); | ||
|
|
||
| Console.CancelKeyPress += (_, e) => | ||
| { | ||
| e.Cancel = true; | ||
| cts.Cancel(); | ||
| }; | ||
|
|
||
| var options = NatsOpts.Default with { LoggerFactory = new MinimumConsoleLoggerFactory(LogLevel.Error) }; | ||
|
|
||
| await using var nats = new NatsConnection(options); | ||
|
|
||
| var js = new NatsJSContext(nats); | ||
|
|
||
| var consumer = await js.CreateConsumerAsync("s1", "c1"); | ||
|
|
||
| var idle = TimeSpan.FromSeconds(15); | ||
| var expires = TimeSpan.FromSeconds(30); | ||
|
|
||
| // int? maxMsgs = null; | ||
| // int? maxBytes = 128; | ||
| int? maxMsgs = 1000; | ||
| int? maxBytes = null; | ||
|
|
||
| static void ErrorHandler(NatsJSNotification notification) | ||
| { | ||
| Console.WriteLine($"Error: {notification}"); | ||
| } | ||
|
|
||
| void Report(int i, Stopwatch sw, string data) | ||
| { | ||
| Console.WriteLine(data); | ||
| if (i % 10000 == 0) | ||
| Console.WriteLine($"Received: {i / sw.Elapsed.TotalSeconds:f2} msgs/s [{i}] {sw.Elapsed}"); | ||
| } | ||
|
|
||
| var consumeOpts = new NatsJSConsumeOpts | ||
| { | ||
| MaxMsgs = maxMsgs, | ||
| MaxBytes = maxBytes, | ||
| Expires = expires, | ||
| IdleHeartbeat = idle, | ||
| Serializer = new RawDataSerializer(), | ||
| ErrorHandler = ErrorHandler, | ||
| }; | ||
|
|
||
| var fetchOpts = new NatsJSFetchOpts | ||
| { | ||
| MaxMsgs = maxMsgs, | ||
| MaxBytes = maxBytes, | ||
| Expires = expires, | ||
| IdleHeartbeat = idle, | ||
| Serializer = new RawDataSerializer(), | ||
| ErrorHandler = ErrorHandler, | ||
| }; | ||
|
|
||
| var nextOpts = new NatsJSNextOpts | ||
| { | ||
| Expires = expires, IdleHeartbeat = idle, Serializer = new RawDataSerializer(), ErrorHandler = ErrorHandler, | ||
| }; | ||
|
|
||
| var stopwatch = Stopwatch.StartNew(); | ||
| var count = 0; | ||
|
|
||
| try | ||
| { | ||
| if (args.Length > 0 && args[0] == "fetch") | ||
| { | ||
| while (!cts.Token.IsCancellationRequested) | ||
| { | ||
| Console.WriteLine($"___\nFETCH {maxMsgs}"); | ||
| await using var sub = await consumer.FetchAsync<RawData>(fetchOpts, cts.Token); | ||
| await foreach (var msg in sub.Msgs.ReadAllAsync(cts.Token)) | ||
| { | ||
| await msg.AckAsync(cancellationToken: cts.Token); | ||
| Report(++count, stopwatch, $"data: {msg.Data}"); | ||
| } | ||
| } | ||
| } | ||
| else if (args.Length > 0 && args[0] == "fetch-all") | ||
| { | ||
| while (!cts.Token.IsCancellationRequested) | ||
| { | ||
| Console.WriteLine($"___\nFETCH {maxMsgs}"); | ||
| await foreach (var msg in consumer.FetchAllAsync<RawData>(fetchOpts, cts.Token)) | ||
| { | ||
| await msg.AckAsync(cancellationToken: cts.Token); | ||
| Report(++count, stopwatch, $"data: {msg.Data}"); | ||
| } | ||
| } | ||
| } | ||
| else if (args.Length > 0 && args[0] == "next") | ||
| { | ||
| while (!cts.Token.IsCancellationRequested) | ||
| { | ||
| Console.WriteLine("___\nNEXT"); | ||
| var next = await consumer.NextAsync<RawData>(nextOpts, cts.Token); | ||
| if (next is { } msg) | ||
| { | ||
| await msg.AckAsync(cancellationToken: cts.Token); | ||
| Report(++count, stopwatch, $"data: {msg.Data}"); | ||
| } | ||
| } | ||
| } | ||
| else if (args.Length > 0 && args[0] == "consume") | ||
| { | ||
| Console.WriteLine("___\nCONSUME"); | ||
| await using var sub = await consumer.ConsumeAsync<RawData>( | ||
| consumeOpts, | ||
| cts.Token); | ||
|
|
||
| await foreach (var msg in sub.Msgs.ReadAllAsync(cts.Token)) | ||
| { | ||
| await msg.AckAsync(cancellationToken: cts.Token); | ||
| Report(++count, stopwatch, $"data: {msg.Data}"); | ||
| } | ||
|
|
||
| // Console.WriteLine($"took {stopwatch.Elapsed}"); | ||
| // await nats.PingAsync(cts.Token); | ||
| } | ||
| else if (args.Length > 0 && args[0] == "consume-all") | ||
| { | ||
| Console.WriteLine("___\nCONSUME-ALL"); | ||
| await foreach (var msg in consumer.ConsumeAllAsync<RawData>(consumeOpts, cts.Token)) | ||
| { | ||
| await msg.AckAsync(cancellationToken: cts.Token); | ||
| Report(++count, stopwatch, $"data: {msg.Data}"); | ||
| } | ||
| } | ||
| else | ||
| { | ||
| Console.WriteLine("Usage: dotnet run -- <consume|consume-all|fetch|fetch-all|next>"); | ||
| } | ||
| } | ||
| catch (OperationCanceledException) | ||
| { | ||
| } | ||
|
|
||
| Console.WriteLine("Bye"); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| using System.Text; | ||
|
|
||
| namespace Example.JetStream.PullConsumer; | ||
|
|
||
| public class RawData | ||
| { | ||
| public RawData(byte[] buffer) => Buffer = buffer; | ||
|
|
||
| public byte[] Buffer { get; } | ||
|
|
||
| public override string ToString() => Encoding.ASCII.GetString(Buffer); | ||
| } |
28 changes: 28 additions & 0 deletions
28
sandbox/Example.JetStream.PullConsumer/RawDataSerializer.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| using System.Buffers; | ||
| using NATS.Client.Core; | ||
|
|
||
| namespace Example.JetStream.PullConsumer; | ||
|
|
||
| public class RawDataSerializer : INatsSerializer | ||
| { | ||
| public int Serialize<T>(ICountableBufferWriter bufferWriter, T? value) | ||
| { | ||
| if (value is RawData data) | ||
| { | ||
| bufferWriter.Write(data.Buffer); | ||
| return data.Buffer.Length; | ||
| } | ||
|
|
||
| throw new Exception($"Can only work with '{typeof(RawData)}'"); | ||
| } | ||
|
|
||
| public T? Deserialize<T>(in ReadOnlySequence<byte> buffer) => (T?)Deserialize(buffer, typeof(T)); | ||
|
|
||
| public object? Deserialize(in ReadOnlySequence<byte> buffer, Type type) | ||
| { | ||
| if (type != typeof(RawData)) | ||
| throw new Exception($"Can only work with '{typeof(RawData)}'"); | ||
|
|
||
| return new RawData(buffer.ToArray()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.