JetStream pull consumer redesign #115
Conversation
mtmk
changed the title
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.
This is a case where server request timeout might not reach the client because of network anomalies.
* Options -> Opts refactor * Ignore extra messages on consumer dispose * Propagate JS options for ACK
| { | ||
| var isVersionLineRead = false; | ||
|
|
||
| while (!reader.End) | ||
| { | ||
| var span = reader.UnreadSpan; | ||
| while (span.Length > 0) | ||
| if (reader.TryReadTo(out ReadOnlySpan<byte> line, ByteCRLF)) |
There was a problem hiding this comment.
public bool TryReadTo (out ReadOnlySpan<T> span ... will copy a multi-span header wheras public bool TryReadTo (out System.Buffers.ReadOnlySequence<T> sequence ... would not.
If it's easy to change TryParseHeaderLine to operate on a ReadOnlySequence<byte> instead of ReadOnlySpan<byte> then that might be a worthwhile optimization
I suppose this is why Kestrel has separate paths for single-span and multi-span headers
It would be fine to shelf this optimization and do it later, the copy for a multi-span header does make the code significantly less complicated
There was a problem hiding this comment.
I agree. Let's do it in another PR.
src/NATS.Client.Core/RawData.cs
Outdated
|
|
||
| namespace NATS.Client.Core; | ||
|
|
||
| public class RawData |
There was a problem hiding this comment.
I think we should evaluate RawData a little more before committing it to the public API. I think it is only used in the Example project, maybe move it there for now?
|
|
||
| internal NatsJSOpts Opts { get; } | ||
|
|
||
| public string NewInbox() => $"{Opts.InboxPrefix}.{Guid.NewGuid():n}"; |
There was a problem hiding this comment.
Should this be internal? Confusing to have a public INatsConnection.NewInbox and NatsJSContext.NewInbox, one using the Mux Subscription and one not.
| public NatsJSMsg(NatsMsg<T> msg, NatsJSContext jsContext) | ||
| { | ||
| Msg = msg; | ||
| JSContext = jsContext; | ||
| } |
There was a problem hiding this comment.
I can't think of a case where NatsJSMsg needs a public constructor. Shouldn't a JS Msg always come from a consumer?
| JSContext = jsContext; | ||
| } | ||
|
|
||
| public NatsJSContext JSContext { get; } | ||
|
|
||
| public NatsMsg<T> Msg { get; } |
There was a problem hiding this comment.
Maybe instead of passing through NatsMsg<T>, we should proxy all of the public properties that are interesting:
string Subject,
int Size,
NatsHeaders? Headers,
T? Data,
INatsConnection? Connection
We could leave out the string? replyTo property, and all of the Reply methods as well... Since the only thing that should be replied with are the Ack methods which are already on the NatsJSMsg<T>
* JSMsg hides the NatsMsg and proxy relevant fields * Moved RawData to example * JSContext NewInbox is now internal
JetStream pull consumers are implemented as direct subclasses of
NatsSubBaseinNatsJSSubConsumeandNatsJSSubFetchclasses. All the pending message tracking logic implemented in the subclasses.Other fixes:
Optsas short nameRawDatatype introduced to potentially removebyte[]like calls (API discussion)