how to read pullsubscribe messages to specific subject

[lisablack1]

Hi I am trying to fetch the messages only for one subject , but getting all the messages from all subject

Below is my code , is there any issue with my code or do i have to change anything

any suggestions are appreciated

published to different subject
ex : sub.1, sub.2
Q1 . receiving messages from all subjects , expecting form one subject
Q2 . with different stream and same subject i am receiving the messages , we should not receive the different stream subject data right ?

package main

import (
    "log"
    "time"
    nats "github.com/nats-io/nats.go"
)

func PrepareNatsConnection(streamName, consumerName, subjectName string) (*nats.Conn, nats.JetStreamContext, error) {
      reconnectOpts := nats.Options{
	      Servers: []string{"0.0.0.0:4222"},
	      Name:    "local_connection",
      }

    connection, err := reconnectOpts.Connect()
    if err != nil {
	    return nil, nil, err
    }
    streamContext, err := connection.JetStream()
    if err != nil {
	    return nil, nil, err
    }
    streamInfo, err := streamContext.StreamInfo(streamName)
    if streamInfo == nil || err != nil {
	    streamInfo, err = streamContext.AddStream(&nats.StreamConfig{
		    Name:     streamName,
		    Subjects: []string{subjectName},
	    })
	    if err != nil {
		    log.Println("error while adding the stream to the context", err)
		    return nil, nil, err
	    }
    }

    consumerInfo, err := streamContext.ConsumerInfo(streamName, consumerName)
    if consumerInfo == nil || err != nil {
	    consumerInfo, err = streamContext.AddConsumer(streamName, &nats.ConsumerConfig{
		    Durable:   consumerName,
		    AckPolicy: nats.AckExplicitPolicy,
	    })
	    if err != nil {
		    log.Println("error while adding the consumer to the context", err)
		    return nil, nil, err
	    }
    }
    log.Println("connected to nats")
    return connection, streamContext, nil
}


func main() {
    streamName := "sample_stream"
    consumerName := "sample_consumer1"
    subjectName := "sub.1"

    _, streamContext, err := PrepareNatsConnection(streamName, consumerName, subjectName)
    if err != nil {
	    log.Fatal(err)
    }
    subscription, err := streamContext.PullSubscribe(subjectName,
	    consumerName,
	    nats.ManualAck(),
	    nats.Bind(streamName, consumerName), //for holding the state
    )
    if err != nil {
	    log.Fatal("Subscribe failed,err:", err)
    }

    err = ConsumeMessagesFormServer(subscription) //consuming the messages from the server
    if err != nil {
	    log.Println("error while consuming the messages from nats,err:", err)
    }
}



func ConsumeMessagesFormServer(subscription *nats.Subscription) error {
    count := 0
    for {
	    if count == 100 {
		    break
	    }

	    //fetching the messages using prefetch count
	    msgs, err := subscription.Fetch(1)
	    if err != nil {
		    log.Println("error while fetching the message", err)
		    time.Sleep(time.Millisecond * 100)
		    continue
	    }
	    for _, message := range msgs {

		    log.Printf("Consumer=>Subject:%s, Data:%+v\n", message.Subject, string(message.Data))
		    message.Ack()
		    count += 1
	    }
	    time.Sleep(time.Millisecond * 100)
    }
    return nil
}

[derekcollison]

Could you print out the ConsumerInfo and make sure it has a FilterSubject in the config that matches what your program wants to filter on.

[lisablack1]

Hi @derekcollison

sample consumer information

MaxDeliver: -1,
FilterSubject: "",
ReplayPolicy: github.com/nats-io/nats.go.ReplayInstantPolicy (0),
RateLimit: 0,

I have added filter subject while adding the consumer

    consumerInfo, err = streamContext.AddConsumer(streamName, &nats.ConsumerConfig{
	    Durable:   consumerName,
	    AckPolicy: nats.AckExplicitPolicy,
               FilterSubject : subjectName,
    })

Its working for sample subjects : sub.1,

but when i am trying for the subj.1.*.900 something like this its not filtering

am i doing something wrong , pls suggest