Function upsertPushConsumer

jetstream/consumer.go:269–285

(ctx context.Context, js *jetStream, stream string, cfg ConsumerConfig, action string)

}

func upsertPushConsumer(ctx context.Context, js *jetStream, stream string, cfg ConsumerConfig, action string) (PushConsumer, error) {
	if cfg.DeliverSubject == "" {
		return nil, ErrNotPushConsumer
	}

	resp, err := upsertConsumer(ctx, js, stream, cfg, action)
	if err != nil {
		return nil, err
	}

	return &pushConsumer{
		js:     js,
		stream: stream,
		name:   resp.Name,
		info:   resp.ConsumerInfo,
	}, nil
}

func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg ConsumerConfig, action string) (*consumerInfoResponse, error) {
	ctx, cancel := js.wrapContextWithoutDeadline(ctx)