How to use IJetStreamPullAsyncSubscription, Handler is not invoked by NATS on publish

[rdkumavat1]

I am trying to use pull consumer with Handler, I am expecting NATS should call that handler on new message publish in the subject, but it is not happening in my case, Can you please share document or example for the same?

I have this sample code.

var consumerConfigModel = Config.ProcessConfigurationObject(consumerConfigObject);

    var configBuilder = ConsumerConfiguration.Builder();

    configBuilder
                         .WithDurable(consumerName)
                         .WithDeliverPolicy(DeliverPolicy.All)
                         .WithAckPolicy(AckPolicy.Explicit)
                         .WithReplayPolicy(ReplayPolicy.Instant)
                         .WithAckWait(consumerConfigModel.max_ack_wait_duration)
                         .WithMaxDeliver(consumerConfigModel.max_deliver)
                         .WithMaxAckPending(consumerConfigModel.max_pending_ack)
                         .WithFilterSubject(subject);
                        //.WithMaxBatch(100);

    // .WithRateLimitBps(subscribers.rateLimitsBps);

    PullSubscribeOptions pullOptions = PullSubscribeOptions.Builder()
            .WithConfiguration(configBuilder.Build())
            .WithStream(streamName)
            
            .Build();

    var subscriptionInfo = jetStream.PullSubscribeAsync(subject, handler, pullOptions);

[mtmk]

cc @scottf

[scottf]

A pull subscription does not work like a push subscription, meaning the server doesn't just start sending messages, you must send a pull request to tell the server to send messages.

Would you please look at all the samples that start with "Simplification" in this folder:

https://github.com/nats-io/nats.net.v1/tree/main/src/Samples

and try using the simplification api. It has consume api which use pull under the covers and provides continuous message retrieval from the server, much like a push consumer would.

[rdkumavat1]

Thank you, but can you please show me the sample that shows usage of - IJetStreamPullAsyncSubscription with hanlder, I want to know how to implement it

Thanks

[scottf]

I found some example code for you. This works fine, but isn't really supported. We much prefer that you use Simplification.

    Options opts = ConnectionFactory.GetDefaultOptions();
    using (IConnection c = new ConnectionFactory().CreateConnection(opts))
    {
        IJetStreamManagement jsm = c.CreateJetStreamManagementContext();
        IJetStream js = c.CreateJetStreamContext();
        
        StreamConfiguration sc = StreamConfiguration.Builder()
            .WithName("stream")
            .WithStorageType(StorageType.Memory)
            .WithSubjects("subject.>")
            .Build();

        jsm.AddStream(sc);
        
        jsm.AddOrUpdateConsumer("stream", ConsumerConfiguration.Builder()
            .WithDurable("durGt")
            .WithFilterSubject(">")
            .Build());
        jsm.AddOrUpdateConsumer("stream", ConsumerConfiguration.Builder()
            .WithDurable("durA")
            .WithFilterSubject("subject.A")
            .Build());

        int countGt = 0;
        PullSubscribeOptions psoGt = PullSubscribeOptions.BindTo("stream", "durGt");
        IJetStreamPullAsyncSubscription subGt = js.PullSubscribeAsync(null,
            (s, e) =>
            {
                countGt++;
                Console.WriteLine($"Received on durGt => {e.Message.Subject} {Encoding.UTF8.GetString(e.Message.Data)}");
            },
            psoGt);
        subGt.PullExpiresIn(10, 1000);

        int countA = 0;
        PullSubscribeOptions psoA = PullSubscribeOptions.BindTo("stream", "durA");
        IJetStreamPullAsyncSubscription subA = js.PullSubscribeAsync(null,
            (s, e) =>
            {
                countA++;
                Console.WriteLine($"Received on durA => {e.Message.Subject} {Encoding.UTF8.GetString(e.Message.Data)}");
            },
            psoA);
        subA.PullExpiresIn(10, 1000);
        
        for (int x = 0; x < 10; x++) {
            if (x % 2 == 0) {
                js.Publish("subject.A", Encoding.UTF8.GetBytes("" + x));
            }
            else {
                js.Publish("subject.B", Encoding.UTF8.GetBytes("" + x));
            }
        }

        Thread.Sleep(2000); // just give time for the message to come across
        Console.WriteLine($"subGt got {countGt} messages.");
        Console.WriteLine($"subA got {countA} messages.");
    }
}