| | | 1 | | using Microsoft.Extensions.Hosting; |
| | | 2 | | using Microsoft.Extensions.Logging; |
| | | 3 | | using Microsoft.Extensions.Options; |
| | | 4 | | using NATS.Client.Core; |
| | | 5 | | using NATS.Net; |
| | | 6 | | |
| | | 7 | | namespace AsyncResponse.Transports.NATS; |
| | | 8 | | |
| | | 9 | | /// <summary> |
| | | 10 | | /// Base hosted service that consumes a JetStream subject through a durable consumer and routes each |
| | | 11 | | /// message to the AsyncResponse ingress with the configured acknowledgement/redelivery/dead-letter |
| | | 12 | | /// policy. A failed consume loop is retried with bounded backoff so a transient NATS outage does not |
| | | 13 | | /// kill the subscriber. |
| | | 14 | | /// </summary> |
| | | 15 | | internal abstract class NatsSubscriberService : BackgroundService |
| | | 16 | | { |
| | | 17 | | /// <summary> |
| | | 18 | | /// Re-arm period of the idle long-poll fetch (the NATS.Net default fetch period). Purely how |
| | | 19 | | /// often an empty wait returns to re-check cancellation — not a delivery deadline. |
| | | 20 | | /// </summary> |
| | 6 | 21 | | private static readonly TimeSpan LongPollExpires = TimeSpan.FromSeconds(30); |
| | | 22 | | |
| | | 23 | | /// <summary> |
| | | 24 | | /// A long poll that comes back empty sooner than this did not expire: the server holds a pull |
| | | 25 | | /// request for the whole <see cref="LongPollExpires"/> when there is nothing to deliver. Half |
| | | 26 | | /// the period leaves room for a poll cut short by a reconnect without ever mistaking a |
| | | 27 | | /// millisecond answer for an expiry. |
| | | 28 | | /// </summary> |
| | 6 | 29 | | private static readonly TimeSpan FastEmptyPollThreshold = LongPollExpires / 2; |
| | | 30 | | |
| | | 31 | | /// <summary> |
| | | 32 | | /// Consecutive fast-empty long polls after which the attempt is handed back to the supervisor, |
| | | 33 | | /// so the stream/consumer provisioning runs again. |
| | | 34 | | /// </summary> |
| | | 35 | | private const int MaxConsecutiveFastEmptyPolls = 5; |
| | | 36 | | |
| | | 37 | | private readonly INatsJetStreamTransport _jetStream; |
| | | 38 | | private readonly TimeProvider _timeProvider; |
| | | 39 | | |
| | | 40 | | /// <summary>Runs the NatsSubscriberService operation.</summary> |
| | | 41 | | protected NatsSubscriberService( |
| | | 42 | | IOptions<NatsAsyncResponseTransportOptions> options, |
| | | 43 | | INatsConnection connection, |
| | | 44 | | ILogger logger) |
| | 392 | 45 | | : this(options, new NatsJetStreamTransportAdapter(connection.CreateJetStreamContext(), logger, options.Value.Str |
| | | 46 | | { |
| | 392 | 47 | | } |
| | | 48 | | |
| | | 49 | | /// <summary>Runs the NatsSubscriberService operation.</summary> |
| | 410 | 50 | | protected NatsSubscriberService( |
| | 410 | 51 | | IOptions<NatsAsyncResponseTransportOptions> options, |
| | 410 | 52 | | INatsJetStreamTransport jetStream, |
| | 410 | 53 | | ILogger logger, |
| | 410 | 54 | | TimeProvider? timeProvider = null) |
| | | 55 | | { |
| | 410 | 56 | | Options = options.Value; |
| | 410 | 57 | | NatsTransportOptionsValidator.ValidateCommon(Options); |
| | 410 | 58 | | _jetStream = jetStream; |
| | 410 | 59 | | Logger = logger; |
| | 410 | 60 | | Schema = new NatsTransportSubjectSchema(Options); |
| | | 61 | | |
| | | 62 | | // Times the idle long poll and paces its fast-empty backoff. The system clock in |
| | | 63 | | // production — what is measured is a real server's answer; the seam exists for tests. |
| | 410 | 64 | | _timeProvider = timeProvider ?? TimeProvider.System; |
| | 410 | 65 | | } |
| | | 66 | | |
| | 8783 | 67 | | protected NatsAsyncResponseTransportOptions Options { get; } |
| | 813 | 68 | | protected ILogger Logger { get; } |
| | 4668 | 69 | | protected NatsTransportSubjectSchema Schema { get; } |
| | | 70 | | |
| | | 71 | | protected abstract string Subject { get; } |
| | | 72 | | protected abstract string Stream { get; } |
| | | 73 | | protected abstract string Consumer { get; } |
| | | 74 | | protected abstract NatsSubscriberOptions SubscriberOptions { get; } |
| | | 75 | | protected abstract NatsSubscriberRole Role { get; } |
| | | 76 | | /// <summary>Handles the delivered message.</summary> |
| | | 77 | | protected abstract Task HandleMessageAsync(NatsJobDelivery delivery, CancellationToken cancellationToken); |
| | | 78 | | |
| | | 79 | | /// <summary>Runs this background operation until cancellation is requested.</summary> |
| | | 80 | | /// <summary> |
| | | 81 | | /// Validates subscriber options here rather than at the top of <c>ExecuteAsync</c>: since |
| | | 82 | | /// Microsoft.Extensions.Hosting.Abstractions 10.0.10, <c>BackgroundService.StartAsync</c> no |
| | | 83 | | /// longer runs <c>ExecuteAsync</c> inline, so a throw there surfaces only through the host's |
| | | 84 | | /// background-exception handling — or never, when a fast stop discards the queued work — |
| | | 85 | | /// instead of failing host startup synchronously. |
| | | 86 | | /// </summary> |
| | | 87 | | public override Task StartAsync(CancellationToken cancellationToken) |
| | | 88 | | { |
| | 402 | 89 | | NatsTransportOptionsValidator.ValidateSubscriber(Options, SubscriberOptions, Role.ToString()); |
| | 400 | 90 | | return base.StartAsync(cancellationToken); |
| | | 91 | | } |
| | | 92 | | |
| | | 93 | | protected override async Task ExecuteAsync(CancellationToken stoppingToken) |
| | | 94 | | { |
| | | 95 | | // The dispatcher — and with it the ACK-after-enqueue queue and its workers — belongs to |
| | | 96 | | // the hosted service, not to one supervised attempt. Disposing it IS the stop-time drain |
| | | 97 | | // (wait BackgroundDrainTimeout, then cancel and dead-letter whatever is still queued), so |
| | | 98 | | // owning it per attempt ran that drain on every fetch-loop failure of a host that was NOT |
| | | 99 | | // stopping: a NATS blip or a JetStream leader election paused consumption for the drain |
| | | 100 | | // budget and then buried queued, already-ACKed work as "drain budget lapsed" — or lost it |
| | | 101 | | // outright when the dead-letter publish rode the same failing connection. Nothing in it |
| | | 102 | | // is per attempt (the JetStream adapter wraps the host's reconnecting connection, and each |
| | | 103 | | // delivery carries its own settlement handles), so every rebuilt attempt feeds this one |
| | | 104 | | // instance and only the host stop drains it. |
| | 400 | 105 | | await using var dispatcher = new NatsMessageDispatcher( |
| | 400 | 106 | | HandleMessageAsync, |
| | 400 | 107 | | _jetStream, |
| | 400 | 108 | | Options, |
| | 400 | 109 | | SubscriberOptions, |
| | 400 | 110 | | Schema, |
| | 400 | 111 | | Logger, |
| | 400 | 112 | | Role, |
| | 400 | 113 | | Consumer); |
| | | 114 | | |
| | 400 | 115 | | await SubscriberSupervisor.RunAsync( |
| | 409 | 116 | | attemptToken => RunSubscriberAsync(dispatcher, attemptToken), |
| | 400 | 117 | | stoppingToken, |
| | 9 | 118 | | failures => AsyncResponseRetry.Backoff(failures, Options.SubscriberRetryBaseDelay, Options.SubscriberRetryMa |
| | 409 | 119 | | (ex, retryDelay) => Logger.LogWarning(ex, "NATS subscriber failed for subject {Subject} ({Role}); retrying i |
| | 400 | 120 | | } |
| | | 121 | | |
| | | 122 | | private async Task RunSubscriberAsync(NatsMessageDispatcher dispatcher, CancellationToken stoppingToken) |
| | | 123 | | { |
| | 409 | 124 | | if (Options.CreateStreams) |
| | | 125 | | { |
| | 409 | 126 | | await _jetStream.EnsureStreamAsync(Stream, Subject, Options.StreamMaxMessages, stoppingToken).ConfigureAwait |
| | 402 | 127 | | if (Options.DeadLetterEnabled) |
| | 402 | 128 | | await _jetStream.EnsureDeadLetterStreamAsync(Schema.DeadLetterStream, Schema.DeadLetterSubject, Options. |
| | | 129 | | } |
| | | 130 | | |
| | 402 | 131 | | await _jetStream.EnsureConsumerAsync(Stream, Consumer, Options.AckWait, stoppingToken).ConfigureAwait(false); |
| | | 132 | | |
| | 400 | 133 | | Logger.LogInformation( |
| | 400 | 134 | | "NATS subscriber started. Subject: {Subject}. Stream: {Stream}. Consumer: {Consumer}. Role: {Role}. AckMode: |
| | 400 | 135 | | Subject, Stream, Consumer, Role, SubscriberOptions.AckMode); |
| | | 136 | | |
| | | 137 | | // JetStream counts a delivery when it hands the message over, not when a handler starts. |
| | | 138 | | // ACK-after-handler runs its batch serially, so every message prefetched behind a handler |
| | | 139 | | // that kills the process (stack overflow, OOM, FailFast) came back with its count bumped |
| | | 140 | | // without ever having run — and MaxDeliveryAttempts crashes later the pre-execution cap |
| | | 141 | | // dead-lettered up to BatchSize-1 healthy batch-mates along with the poison one. One |
| | | 142 | | // message per fetch leaves the rest on the stream, where nothing is counted and any peer |
| | | 143 | | // can take it. ACK-after-enqueue settles each message as it is accepted, so it keeps the |
| | | 144 | | // batch. |
| | 400 | 145 | | var fetchSize = SubscriberOptions.AckMode is NatsAckMode.AckAfterEnqueue ? SubscriberOptions.BatchSize : 1; |
| | 400 | 146 | | var batch = new List<NatsJobDelivery>(fetchSize); |
| | 400 | 147 | | var fastEmptyPolls = 0; |
| | 1218 | 148 | | while (!stoppingToken.IsCancellationRequested) |
| | | 149 | | { |
| | 851 | 150 | | batch.Clear(); |
| | | 151 | | |
| | | 152 | | // Drain whatever is already available, up to the fetch size. The batch is |
| | | 153 | | // materialized out of the client buffer BEFORE dispatch so the in-progress heartbeat |
| | | 154 | | // below can reach every waiting message: the server starts each message's AckWait |
| | | 155 | | // clock at delivery, and an open-ended consume buffered the whole prefetch |
| | | 156 | | // client-side — a serial batch whose handlers together outlast AckWait had its tail |
| | | 157 | | // redelivered to a competing consumer (and NumDelivered climbed toward the Term cap) |
| | | 158 | | // while it was still queued here. |
| | 2200 | 159 | | await foreach (var delivery in _jetStream.FetchNoWaitAsync(Stream, Consumer, fetchSize, stoppingToken).Confi |
| | 249 | 160 | | batch.Add(delivery); |
| | | 161 | | |
| | 833 | 162 | | if (batch.Count == 0) |
| | | 163 | | { |
| | | 164 | | // Nothing waiting: long-poll for a single message so idle delivery latency stays |
| | | 165 | | // push-like. Siblings arriving behind the long-polled message stay ON the stream |
| | | 166 | | // — where AckWait has not started — until the next no-wait drain. |
| | 584 | 167 | | var pollStarted = _timeProvider.GetTimestamp(); |
| | 1544 | 168 | | await foreach (var delivery in _jetStream.FetchAsync(Stream, Consumer, maxMessages: 1, LongPollExpires, |
| | 188 | 169 | | batch.Add(delivery); |
| | | 170 | | |
| | 569 | 171 | | if (batch.Count == 0) |
| | | 172 | | { |
| | 381 | 173 | | fastEmptyPolls = await BackOffAfterEmptyLongPollAsync(_timeProvider.GetElapsedTime(pollStarted), fas |
| | 381 | 174 | | continue; |
| | | 175 | | } |
| | | 176 | | } |
| | | 177 | | |
| | 437 | 178 | | fastEmptyPolls = 0; |
| | 437 | 179 | | await DispatchBatchAsync(dispatcher, batch, stoppingToken).ConfigureAwait(false); |
| | | 180 | | } |
| | 367 | 181 | | } |
| | | 182 | | |
| | | 183 | | /// <summary> |
| | | 184 | | /// Tells an expired long poll (re-arm at once) from one the server never held. A pull request |
| | | 185 | | /// that reaches no live consumer — the durable or its stream was deleted, or JetStream has no |
| | | 186 | | /// leader for it — is answered "503 no responders", which the client ends WITHOUT an |
| | | 187 | | /// exception: exactly what an expiry looks like, only in a millisecond. Re-arming on that spun |
| | | 188 | | /// the loop thousands of times a second against a cluster that was already in trouble, and |
| | | 189 | | /// since nothing ever threw, the supervisor never reran the provisioning that would have |
| | | 190 | | /// recreated the consumer — the subscriber stayed dead until the process restarted. (A poll |
| | | 191 | | /// in flight when the consumer is deleted does throw; the silent case is the replica that was |
| | | 192 | | /// inside a handler at that moment.) Returns the updated consecutive fast-empty count. |
| | | 193 | | /// </summary> |
| | | 194 | | private async Task<int> BackOffAfterEmptyLongPollAsync(TimeSpan pollDuration, int fastEmptyPolls, CancellationToken |
| | | 195 | | { |
| | 381 | 196 | | if (pollDuration >= FastEmptyPollThreshold || stoppingToken.IsCancellationRequested) |
| | 381 | 197 | | return 0; // the long poll expired empty; re-arm |
| | | 198 | | |
| | 0 | 199 | | fastEmptyPolls++; |
| | 0 | 200 | | if (fastEmptyPolls >= MaxConsecutiveFastEmptyPolls) |
| | | 201 | | { |
| | 0 | 202 | | throw new InvalidOperationException( |
| | 0 | 203 | | $"NATS consumer '{Consumer}' on stream '{Stream}' answered {fastEmptyPolls} consecutive long polls empty |
| | 0 | 204 | | $"(the last one returned after {pollDuration.TotalMilliseconds:F0} ms of a {LongPollExpires.TotalSeconds |
| | 0 | 205 | | "the pull requests are not reaching a live consumer — it or its stream was deleted, or JetStream has no |
| | | 206 | | } |
| | | 207 | | |
| | 0 | 208 | | var delay = AsyncResponseRetry.Backoff(fastEmptyPolls, Options.SubscriberRetryBaseDelay, Options.SubscriberRetry |
| | 0 | 209 | | Logger.LogDebug( |
| | 0 | 210 | | "NATS long poll for {Role} returned empty after {PollDuration} instead of being held; backing off {Delay} be |
| | 0 | 211 | | Role, |
| | 0 | 212 | | pollDuration, |
| | 0 | 213 | | delay, |
| | 0 | 214 | | fastEmptyPolls, |
| | 0 | 215 | | MaxConsecutiveFastEmptyPolls); |
| | 0 | 216 | | await Task.Delay(delay, _timeProvider, stoppingToken).ConfigureAwait(false); |
| | 0 | 217 | | return fastEmptyPolls; |
| | 381 | 218 | | } |
| | | 219 | | |
| | | 220 | | private async Task DispatchBatchAsync( |
| | | 221 | | NatsMessageDispatcher dispatcher, |
| | | 222 | | List<NatsJobDelivery> batch, |
| | | 223 | | CancellationToken stoppingToken) |
| | | 224 | | { |
| | | 225 | | // The batch is dispatched serially, so a slow handler lets the server-side AckWait of the |
| | | 226 | | // later (still unsettled) messages lapse into redelivery. While the batch is in flight, a |
| | | 227 | | // heartbeat signals in-progress for every unsettled message to reset its AckWait window. |
| | | 228 | | // The heartbeat is NOT tied to the stop token: a handler takes no token, so it outlives |
| | | 229 | | // the stop signal, and cancelling its renewal there let AckWait lapse under the live |
| | | 230 | | // handler on every rolling deploy — the job was redelivered to a peer and ran twice. It |
| | | 231 | | // ends only when this loop has let go of every message. |
| | 437 | 232 | | var progress = new BatchProgress(); |
| | 437 | 233 | | using var renewalCancellation = new CancellationTokenSource(); |
| | 437 | 234 | | var renewalTask = RenewInProgressLoopAsync(batch, progress, renewalCancellation.Token); |
| | 437 | 235 | | var next = 0; |
| | | 236 | | try |
| | | 237 | | { |
| | 1311 | 238 | | for (; next < batch.Count; next++) |
| | | 239 | | { |
| | | 240 | | // Stopping: do not start what has not started. The rest of the batch used to run |
| | | 241 | | // on, handler after handler, past the stop signal. |
| | 437 | 242 | | if (stoppingToken.IsCancellationRequested) |
| | | 243 | | break; |
| | | 244 | | |
| | | 245 | | try |
| | | 246 | | { |
| | 437 | 247 | | await dispatcher.HandleAsync(batch[next], stoppingToken).ConfigureAwait(false); |
| | 437 | 248 | | } |
| | | 249 | | finally |
| | | 250 | | { |
| | 437 | 251 | | progress.MarkSettled(); |
| | | 252 | | } |
| | | 253 | | } |
| | | 254 | | } |
| | | 255 | | finally |
| | | 256 | | { |
| | | 257 | | // Hand back whatever never started (a stop, or a handler cancelled by it, cut the |
| | | 258 | | // batch short) while the heartbeat still covers it. |
| | 874 | 259 | | for (var i = Math.Max(next, progress.SettledCount); i < batch.Count; i++) |
| | 0 | 260 | | await ReleaseUnstartedAsync(batch[i]).ConfigureAwait(false); |
| | | 261 | | |
| | 437 | 262 | | renewalCancellation.Cancel(); |
| | | 263 | | try |
| | | 264 | | { |
| | | 265 | | // Cancellation exits the sweep between messages and aborts the in-flight |
| | | 266 | | // heartbeat (the token reaches the SDK call), so this normally completes at once. |
| | | 267 | | // The bound is the hard backstop for a heartbeat the client cannot abort — a |
| | | 268 | | // write wedged on a dead socket: an unbounded join here held the loop after every |
| | | 269 | | // message in the batch had settled, so no further batch was fetched and a stop |
| | | 270 | | // never completed, with nothing for the supervisor to restart. Past one heartbeat |
| | | 271 | | // interval the loop is abandoned; the server's AckWait settles whatever it left. |
| | 437 | 272 | | await renewalTask.WaitAsync(RenewalInterval).ConfigureAwait(false); |
| | 435 | 273 | | } |
| | 2 | 274 | | catch (TimeoutException) |
| | | 275 | | { |
| | 2 | 276 | | Logger.LogWarning( |
| | 2 | 277 | | "NATS in-progress heartbeat for {Role} did not stop within {RenewalInterval} after its batch settled |
| | 2 | 278 | | Role, |
| | 2 | 279 | | RenewalInterval); |
| | 2 | 280 | | _ = renewalTask.ContinueWith( |
| | 0 | 281 | | static (task, state) => ((ILogger)state!).LogWarning(task.Exception, "Abandoned NATS in-progress hea |
| | 2 | 282 | | Logger, |
| | 2 | 283 | | CancellationToken.None, |
| | 2 | 284 | | TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, |
| | 2 | 285 | | TaskScheduler.Default); |
| | 2 | 286 | | } |
| | | 287 | | } |
| | 437 | 288 | | } |
| | | 289 | | |
| | | 290 | | /// <summary> |
| | | 291 | | /// Hands a prefetched message that never started back to the server, with no redelivery delay |
| | | 292 | | /// so an idle peer can take it at once. Leaving it unsettled instead would pin it for the rest |
| | | 293 | | /// of its AckWait window — the stop that cut the batch short is usually a rolling deploy, and |
| | | 294 | | /// the messages behind the handler are exactly the work the surviving replicas should pick up. |
| | | 295 | | /// A failure here is not worth failing the stop over: the window lapses and the server |
| | | 296 | | /// redelivers anyway, which is the same outcome one AckWait later. |
| | | 297 | | /// </summary> |
| | | 298 | | private async Task ReleaseUnstartedAsync(NatsJobDelivery delivery) |
| | | 299 | | { |
| | | 300 | | try |
| | | 301 | | { |
| | 0 | 302 | | await delivery.NakAsync(TimeSpan.Zero).ConfigureAwait(false); |
| | 0 | 303 | | } |
| | 0 | 304 | | catch (Exception ex) |
| | | 305 | | { |
| | 0 | 306 | | Logger.LogDebug( |
| | 0 | 307 | | ex, |
| | 0 | 308 | | "Failed to hand back an unstarted NATS message on subject {Subject} ({Role}); it redelivers when its Ack |
| | 0 | 309 | | delivery.Subject, |
| | 0 | 310 | | Role); |
| | 0 | 311 | | } |
| | 0 | 312 | | } |
| | | 313 | | |
| | | 314 | | /// <summary> |
| | | 315 | | /// ~AckWait/3: two chances to land a renewal inside every AckWait window even when one sweep |
| | | 316 | | /// is delayed by a slow round trip. Also the bound on joining the renewal loop after a batch. |
| | | 317 | | /// </summary> |
| | 876 | 318 | | private TimeSpan RenewalInterval => TimeSpan.FromMilliseconds(Math.Max(1, Options.AckWait.TotalMilliseconds / 3)); |
| | | 319 | | |
| | | 320 | | private async Task RenewInProgressLoopAsync( |
| | | 321 | | List<NatsJobDelivery> batch, |
| | | 322 | | BatchProgress progress, |
| | | 323 | | CancellationToken cancellationToken) |
| | | 324 | | { |
| | 437 | 325 | | var interval = RenewalInterval; |
| | | 326 | | try |
| | | 327 | | { |
| | 4 | 328 | | while (true) |
| | | 329 | | { |
| | 441 | 330 | | await Task.Delay(interval, cancellationToken).ConfigureAwait(false); |
| | | 331 | | |
| | | 332 | | // Renew from the first unsettled message onward: that covers the message |
| | | 333 | | // currently in the handler plus everything still waiting its turn. A settle |
| | | 334 | | // racing this sweep is harmless — Ack/Nak/Term has already consumed the delivery |
| | | 335 | | // server-side, and a late in-progress signal for it is ignored rather than |
| | | 336 | | // un-settling anything, so no suppression mark is needed (unlike the SQS twin, |
| | | 337 | | // whose failure path re-arms a visibility a late renewal could stretch). |
| | 20 | 338 | | for (var i = progress.SettledCount; i < batch.Count; i++) |
| | | 339 | | { |
| | | 340 | | // The batch finished or the subscriber is stopping: exit quietly between |
| | | 341 | | // messages instead of spending a round trip on each remaining renewal. |
| | 6 | 342 | | if (cancellationToken.IsCancellationRequested) |
| | 0 | 343 | | return; |
| | | 344 | | |
| | 6 | 345 | | if (i < progress.SettledCount) |
| | | 346 | | continue; |
| | | 347 | | |
| | 6 | 348 | | var delivery = batch[i]; |
| | | 349 | | try |
| | | 350 | | { |
| | 6 | 351 | | await delivery.ProgressAsync(cancellationToken).ConfigureAwait(false); |
| | 4 | 352 | | } |
| | 2 | 353 | | catch (Exception ex) when (ex is not OperationCanceledException) |
| | | 354 | | { |
| | 0 | 355 | | Logger.LogWarning( |
| | 0 | 356 | | ex, |
| | 0 | 357 | | "Failed to signal in-progress for NATS message on subject {Subject} ({Role}); its AckWait ma |
| | 0 | 358 | | delivery.Subject, |
| | 0 | 359 | | Role); |
| | 0 | 360 | | } |
| | 4 | 361 | | } |
| | | 362 | | } |
| | | 363 | | } |
| | 437 | 364 | | catch (OperationCanceledException) |
| | | 365 | | { |
| | | 366 | | // The batch finished or the subscriber is stopping. |
| | 437 | 367 | | } |
| | 437 | 368 | | } |
| | | 369 | | |
| | | 370 | | private sealed class BatchProgress |
| | | 371 | | { |
| | | 372 | | // Settled only ever increments, so a monotonic volatile read is enough — no lock, and a |
| | | 373 | | // stale read only renews an already-settled message once more (which the server ignores). |
| | | 374 | | private int _settledCount; |
| | | 375 | | |
| | 449 | 376 | | public int SettledCount => Volatile.Read(ref _settledCount); |
| | | 377 | | |
| | 437 | 378 | | public void MarkSettled() => Interlocked.Increment(ref _settledCount); |
| | | 379 | | } |
| | | 380 | | } |
| | | 381 | | |
| | | 382 | | /// <summary>Consumes worker-job messages and executes them through the AsyncResponse ingress.</summary> |
| | | 383 | | internal sealed class NatsWorkerSubscriber : NatsSubscriberService |
| | | 384 | | { |
| | | 385 | | private readonly IAsyncResponseIngress _ingress; |
| | | 386 | | |
| | | 387 | | /// <summary>Runs the NatsWorkerSubscriber operation.</summary> |
| | | 388 | | public NatsWorkerSubscriber( |
| | | 389 | | IOptions<NatsAsyncResponseTransportOptions> options, |
| | | 390 | | INatsConnection connection, |
| | | 391 | | IAsyncResponseIngress ingress, |
| | | 392 | | ILogger<NatsWorkerSubscriber> logger) |
| | | 393 | | : base(options, connection, logger) |
| | | 394 | | => _ingress = ingress; |
| | | 395 | | |
| | | 396 | | internal NatsWorkerSubscriber( |
| | | 397 | | IOptions<NatsAsyncResponseTransportOptions> options, |
| | | 398 | | INatsJetStreamTransport jetStream, |
| | | 399 | | IAsyncResponseIngress ingress, |
| | | 400 | | ILogger<NatsWorkerSubscriber> logger) |
| | | 401 | | : base(options, jetStream, logger) |
| | | 402 | | => _ingress = ingress; |
| | | 403 | | |
| | | 404 | | protected override string Subject => Schema.WorkerSubject; |
| | | 405 | | protected override string Stream => Schema.WorkerStream; |
| | | 406 | | protected override string Consumer => Options.WorkerConsumer; |
| | | 407 | | protected override NatsSubscriberOptions SubscriberOptions => Options.WorkerSubscriber; |
| | | 408 | | protected override NatsSubscriberRole Role => NatsSubscriberRole.Worker; |
| | | 409 | | |
| | | 410 | | /// <summary>Handles the delivered message.</summary> |
| | | 411 | | protected override Task HandleMessageAsync(NatsJobDelivery delivery, CancellationToken cancellationToken) |
| | | 412 | | => _ingress.HandleWorkerMessageAsync(delivery.Payload); |
| | | 413 | | } |
| | | 414 | | |
| | | 415 | | /// <summary>Consumes response messages and feeds them into the AsyncResponse ingress, correlated by header or JSON body |
| | | 416 | | internal sealed class NatsResponseIngressSubscriber : NatsSubscriberService |
| | | 417 | | { |
| | | 418 | | private readonly IAsyncResponseIngress _ingress; |
| | | 419 | | |
| | | 420 | | /// <summary>Runs the NatsResponseIngressSubscriber operation.</summary> |
| | | 421 | | public NatsResponseIngressSubscriber( |
| | | 422 | | IOptions<NatsAsyncResponseTransportOptions> options, |
| | | 423 | | INatsConnection connection, |
| | | 424 | | IAsyncResponseIngress ingress, |
| | | 425 | | ILogger<NatsResponseIngressSubscriber> logger) |
| | | 426 | | : base(options, connection, logger) |
| | | 427 | | => _ingress = ingress; |
| | | 428 | | |
| | | 429 | | internal NatsResponseIngressSubscriber( |
| | | 430 | | IOptions<NatsAsyncResponseTransportOptions> options, |
| | | 431 | | INatsJetStreamTransport jetStream, |
| | | 432 | | IAsyncResponseIngress ingress, |
| | | 433 | | ILogger<NatsResponseIngressSubscriber> logger) |
| | | 434 | | : base(options, jetStream, logger) |
| | | 435 | | => _ingress = ingress; |
| | | 436 | | |
| | | 437 | | protected override string Subject => Schema.ResponseSubject; |
| | | 438 | | protected override string Stream => Schema.ResponseStream; |
| | | 439 | | protected override string Consumer => Options.ResponseConsumer; |
| | | 440 | | protected override NatsSubscriberOptions SubscriberOptions => Options.ResponseSubscriber; |
| | | 441 | | protected override NatsSubscriberRole Role => NatsSubscriberRole.ResponseIngress; |
| | | 442 | | |
| | | 443 | | /// <summary>Handles the delivered message.</summary> |
| | | 444 | | protected override Task HandleMessageAsync(NatsJobDelivery delivery, CancellationToken cancellationToken) |
| | | 445 | | { |
| | | 446 | | var correlationId = !_ingress.IsOverInboundBudget(delivery.Payload) |
| | | 447 | | ? NatsCorrelationIdExtractor.Extract(delivery.Headers, delivery.Payload, Options) |
| | | 448 | | : null; |
| | | 449 | | return _ingress.HandleResponseMessageAsync(delivery.Payload, correlationId); |
| | | 450 | | } |
| | | 451 | | } |