fixed pull consumer start, added pull consumer params into channeltem…

…plate

config/jetstream/302-jsm-channel.yaml

@@ -88,6 +88,19 @@ spec:
                     sampleFrequency:
                       description: SampleFrequency sets the percentage of acknowledgements that should be sampled for observability. Valid values are in the range 0-100 and, for example, allows both formats of "30" and "30%".
                       type: string
+                    consumerType:
+                      description: Push or Pull based subscriptions
+                      type: string
+                    fetchMaxWait:
+                      description: Pull consumer. Time to wait for a batch of messages if there is not enough of messages on a stream
+                      type: integer
+                      format: int64
+                      minimum: 0
+                    fetchBatchSize:
+                      description: Pull consumer. Batch of messages to pull from stream at once for processing
+                      type: integer
+                      format: int32
+                      minimum: 1
                 delivery:
                   description: DeliverySpec contains the default delivery spec for each subscription to this Channelable. Each subscription delivery spec, if any, overrides this global delivery spec.
                   type: object

pkg/channel/jetstream/dispatcher/dispatcher.go

@@ -299,6 +299,11 @@ func (d *Dispatcher) subscribe(ctx context.Context, config ChannelConfig, sub Su
 			logger.Errorw("failed to create pull consumer", zap.Error(err))
 			return SubscriberStatusTypeError, err
 		}
+		err = consumer.(*PullConsumer).Start()
+		if err != nil {
+			logger.Errorw("failed to start pull consumer", zap.Error(err))
+			return SubscriberStatusTypeError, err
+		}
 	}
 
 	d.consumers[sub.UID] = consumer

pkg/channel/jetstream/dispatcher/pull_consumer.go

@@ -140,7 +140,6 @@ func (c *PullConsumer) Start() error {
 	if err := c.checkStart(); err != nil {
 		return err
 	}
-
 	defer close(c.closed)
 
 	ctx := logging.WithLogger(context.Background(), c.logger)
@@ -158,9 +157,9 @@ func (c *PullConsumer) Start() error {
 
 	for {
 		// TODO move 200 into subscription config
-		batch, err := c.natsConsumer.Fetch(FetchBatchSize, nats.MaxWait(200*time.Millisecond))
+		batch, err := c.natsConsumer.FetchBatch(FetchBatchSize, nats.MaxWait(200*time.Millisecond))
 		if err != nil {
-			return err
+			c.logger.Errorw("Failed to fetch messages", zap.Error(err), zap.String("consumer", c.sub.Name))
 		}
 
 		c.consumeMessages(ctx, batch, &wg)
@@ -172,8 +171,8 @@ func (c *PullConsumer) Start() error {
 //
 // This method returns once the MessageBatch has been consumed, or upon a call to Consumer.Close.
 // Returning as a result of Consumer.Close results in an io.EOF error.
-func (c *PullConsumer) consumeMessages(ctx context.Context, batch []*nats.Msg, wg *sync.WaitGroup) {
-	for _, natsMsg := range batch {
+func (c *PullConsumer) consumeMessages(ctx context.Context, batch nats.MessageBatch, wg *sync.WaitGroup) {
+	for natsMsg := range batch.Messages() {
 		wg.Add(1)
 
 		go func() {