Replies: 7 comments 9 replies
-
|
Beta Was this translation helpful? Give feedback.
-
My worker service uses busy polling to retrieve one sequential message and processes it: var options = ConnectionFactory.GetDefaultOptions();
//My complete initialization logic is here: https://github.com/nats-io/nats.net/discussions/627#discussion-4176196
var connection = connectionFactory.CreateConnection(options);
while (NotCancelled)
{
if (connection.IsClosed())
{
throw new Exception($"NATS connection is closed");
}
while (connection.IsReconnecting())
{
Thread.Yield(); //Because how else can you handle this state?
}
var context = connection.CreateJetStreamContext();
var options = PullSubscribeOptions.Builder()
.WithStream("STREAM")
.WithDurable("WORKER")
.Build();
var subscription = context.PullSubscribe(subject, options);
connection.Flush(1000);
var msgs = subscription.Fetch(1, 10000);
if (!msgs.Any())
{
continue;
}
var msg = msgs.First();
if (msg == null || !msg.IsJetStream)
{
continue;
}
msg.Ack();
Process(msg);
}
Sure. This problem is intermittant so I'll prepare for the next time it happens.
Maybe. However I ran Two remaining questions: How should I handle NATS connection in my worker service's while loop?
What does "waiting pulls" mean from my |
Beta Was this translation helpful? Give feedback.
-
|
Waiting pulls means a pull request was issued before another one was completed and they are being queued up. Is there any place other than the fetch that you make a pull request? Can you please supply your stream config? |
Beta Was this translation helpful? Give feedback.
-
|
I would not handle your connection in your loop. worry about it if you get a message in the event listener and cleanup / restart a task.
|
Beta Was this translation helpful? Give feedback.
-
My response:
My reasoning for creating a connection for every loop: After a month the .NET process will throw timeouts instead of invoking the event listeners. Restarting the process was the only way to reconnect to the NATS server. The logic is:
Note: I am using pull-based consumer because 1. I require exactly-once guarantee and 2. my processing system is buggy/has huge downtime. If my Windows Service creates a new consumer wouldn't it start processing from the first message instead of current? That is why I have created the consumer manually using NATS CLI. What is the best practice - the app creates the NATS consumer not the IT manager? What if we're dealing with multiple consumers on the same work group?
|
Beta Was this translation helpful? Give feedback.
-
|
Converted to discussion |
Beta Was this translation helpful? Give feedback.
-
|
@Exagram Can you please repost your latest code, particularly the consumer configuration and your app loop |
Beta Was this translation helpful? Give feedback.
-
|
Hi Scott, I tried Curious is this
Note: I have moved the connection logic inside the while loop because the processes start failing after a month of operation because I believe it is more robust (Another issue to deal with). App code: //Worker.cs
public Worker()
{
this.retryPolicy = Policy
.Handle<Exception>()
.WaitAndRetryForever(
sleepDurationProvider: (retryAttempt) => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt % 10)),
onRetry: (e, t) => {
logger.LogWarning($"Retry NATS connection {t}");
});
}
while (!stoppingToken.IsCancellationRequested)
{
await processor.Process(stoppingToken);
}
//processor.cs
public async Task Process(CancellationToken stoppingToken)
{
var message = retryPolicy.Execute((c) => messageService.Subscribe(c), stoppingToken);
...
}
//service.cs
public IMessage Subscribe(CancellationToken cancellationToken) {
var options = ConnectionFactory.GetDefaultOptions();
options.Url = url;
options.Name = name;
options.Timeout = timeout;
options.Verbose = hostEnvironment.IsDevelopment();
var connection = connectionFactory.CreateConnection(options);
var context = connection.CreateJetStreamContext();
var options = PullSubscribeOptions.BindTo(stream, consumer);
var subscription = context.PullSubscribe(subject, options);
connection.Flush(1000);
while (!cancellationToken.IsCancellationRequested)
{
var msgs = subscription.Fetch(1, 10000);
if (!msgs.Any())
{
continue;
}
var msg = msgs.First();
...
msg.Ack();
}
} |
Beta Was this translation helpful? Give feedback.
-
|
One thing to reduce confusion is to have the stream and subjects be different names. I know it's legal, but it's confusing. |
Beta Was this translation helpful? Give feedback.
-
Summary:
I've also done the following tests on two consoles:
|
Beta Was this translation helpful? Give feedback.
-
|
So the bind issue is likely a subject issue. The subject in the subscribe request, even if it's bind, must exactly match the filter subject.
|
Beta Was this translation helpful? Give feedback.
-
|
Is there an issue of mismatched subject for Consumer filter subject: |
Beta Was this translation helpful? Give feedback.
-
|
Ok just eliminating that. |
Beta Was this translation helpful? Give feedback.
-
|
Take a look at this example and see if it helps: using System;
using System.Text;
using System.Threading;
using NATS.Client;
using NATS.Client.JetStream;
namespace NATSExamples
{
internal static class JetStreamStarter
{
static void Main(string[] args)
{
using (IConnection connection1 = new ConnectionFactory().CreateConnection("nats://localhost:4222"))
using (IConnection connection2 = new ConnectionFactory().CreateConnection("nats://localhost:4222"))
{
IJetStreamManagement jsm = connection1.CreateJetStreamManagementContext();
// start off with a clean stream for the example
try { jsm.DeleteStream("STREAM"); } catch (NATSJetStreamException) { }
jsm.AddStream(StreamConfiguration.Builder()
.WithName("STREAM")
.WithSubjects("SUBJECT.>")
.Build());
jsm.AddOrUpdateConsumer("STREAM", ConsumerConfiguration.Builder()
.WithDurable("WORKER")
.WithFilterSubject("SUBJECT.A")
.Build()
);
Console.WriteLine();
int count = 10;
new Thread(() =>
{
IJetStream context2 = connection2.CreateJetStreamContext();
PullSubscribeOptions options = PullSubscribeOptions.BindTo("STREAM", "WORKER");
// subscribe subject must be an exact match to filter subject
IJetStreamPullSubscription subscription = context2.PullSubscribe("SUBJECT.A", options);
int red = 0;
while (red < count)
{
ConsumerInfo loopInfo = subscription.GetConsumerInformation();
Console.WriteLine(
"\nConsumerInfo{" +
"Name='" + loopInfo.Name + '\'' +
", NumPending=" + loopInfo.NumPending +
", NumWaiting=" + loopInfo.NumWaiting +
", Delivered=" + loopInfo.Delivered +
", AckFloor=" + loopInfo.AckFloor +
'}');
subscription.PullExpiresIn(1, 500);
try
{
Msg msg = subscription.NextMessage(600);
Console.WriteLine("Got Message: " + msg.Subject + " " + Encoding.UTF8.GetString(msg.Data));
red++;
}
catch (NATSTimeoutException)
{
Console.WriteLine("Normal timeout waiting for message. Probably no messages currently.");
}
}
ConsumerInfo afterInfo = subscription.GetConsumerInformation();
Console.WriteLine(
"\nConsumerInfo{" +
"Name='" + afterInfo.Name + '\'' +
", NumPending=" + afterInfo.NumPending +
", NumWaiting=" + afterInfo.NumWaiting +
", Delivered=" + afterInfo.Delivered +
", AckFloor=" + afterInfo.AckFloor +
"}\n\n");
}).Start();
IJetStream context1 = connection2.CreateJetStreamContext();
for (int x = 1; x <= count; x++)
{
context1.Publish("SUBJECT.A", Encoding.UTF8.GetBytes("DATA-A-" + x));
context1.Publish("SUBJECT.B", Encoding.UTF8.GetBytes("DATA-B-" + x));
Thread.Sleep(2000);
}
}
}
}
} |
Beta Was this translation helpful? Give feedback.
-
|
Good news! I was able to get the above console tests to pass as well as my .NET web app consumer:
Unfortunately I wasn't able to figure out the root cause. |
Beta Was this translation helpful? Give feedback.
-
|
Update - I might not be receiving messages because of exceeding NATS Server resources: NATS Client 0.14.5 error: NATS Server error: Diagnosis:
|
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
-
I am keeping a NATS connection for a couple of hours (~1 day). Suddenly I'm getting issues connecting to it. Restarting both NATS server and my .NET application doesn't fix it. Thoughts?
In Powershell I ran the server:
My .NET code using the default connection logic:
Exception thrown on line
return connectionFactory.CreateConnection(options);:I moved away from .NET to two Powershell windows:
nats publish "filter.blah.test" "{}"nats consumer next STREAM WORKERObservations through
nats consumer info STREAM WORKER:nats consumer next STREAM WORKER:Waiting Pullsgoes up from 70 to 72 (up by 2). What are Waiting Pulls? I notice that on eachnextthey increment by random factors.Beta Was this translation helpful? Give feedback.
All reactions