Method Consume

jetstream/push.go:47–178

(handler MessageHandler, opts ...PushConsumeOpt)

)

func (p *pushConsumer) Consume(handler MessageHandler, opts ...PushConsumeOpt) (ConsumeContext, error) {
	if handler == nil {
		return nil, ErrHandlerRequired
	}
	consumeOpts := &pushConsumeOpts{}
	for _, opt := range opts {
		if err := opt.configurePushConsume(consumeOpts); err != nil {
			return nil, err
		}
	}

	p.Lock()
	defer p.Unlock()

	if p.info == nil {
		return nil, ErrConsumerNotFound
	}

	if p.started.Load() {
		return nil, ErrConsumerAlreadyConsuming
	}

	consumeID := nuid.Next()
	sub := &pushSubscription{
		id:                consumeID,
		errs:              make(chan error, 1),
		done:              make(chan struct{}, 1),
		consumeOpts:       consumeOpts,
		connStatusChanged: p.js.conn.StatusChanged(nats.CONNECTED, nats.RECONNECTING),
		idleHeartbeat:     p.info.Config.IdleHeartbeat,
	}

	sub.hbMonitor = sub.scheduleHeartbeatCheck(sub.idleHeartbeat)
	internalHandler := func(msg *nats.Msg) {
		if sub.hbMonitor != nil {
			sub.hbMonitor.Stop()
		}
		defer func() {
			if sub.hbMonitor != nil {
				sub.hbMonitor.Reset(2 * sub.idleHeartbeat)
			}
		}()
		status, descr := msg.Header.Get("Status"), msg.Header.Get("Description")
		if status == "" {
			jsMsg := p.js.toJSMsg(msg)
			handler(jsMsg)
			return
		}
		sub.Lock()
		if err, terminate := sub.handleStatusMsg(msg, status, descr); err != nil {
			if sub.consumeOpts.ErrHandler != nil {
				sub.consumeOpts.ErrHandler(sub, err)
			}
			if terminate {
				sub.Stop()
			}
		}
		sub.Unlock()