sess/subscribe.go
func (h *Handler) ConnChange(ctx context.Context,
	product, productPassword string) (*rpc.Subscription, error) {
	logger := h.logger.Add("method", "ConnChange", "product", product)

	logger.Info("subscribing to adapter connection changes")

	_, err := h.checkProductPassword(logger, product, productPassword)
	if err != nil {
		return nil, err
	}

	ntf, ok := rpc.NotifierFromContext(ctx)
	if !ok {
		logger.Error("no notifier found in context")
		return nil, ErrInternal
	}

	sub := ntf.CreateSubscription()
	cb := func(job *data.Job, result error) {
		if result == nil {
			h.handleConnChange(product, logger, ntf, sub, job)
		}
	}
	jobTypes := []string{
		data.JobClientPreServiceSuspend,
		data.JobClientPreServiceUnsuspend,
		data.JobClientPreServiceTerminate,
	}

	// TODO: testing. fix it later.
	sid := string(sub.ID)
	if err = h.queue.Subscribe(jobTypes, sid, cb); err != nil {
		logger.Error(err.Error())
		return nil, ErrInternal
	}

	go func() {
		for err, ok := <-sub.Err(); ok; {
			if err != nil {
				logger.Warn(err.Error())
			}
		}

		err := h.queue.Unsubscribe(jobTypes, sid)
		if err != nil {
			logger.Error(err.Error())
		}
	}()

	return sub, nil
}
func (h *Handler) ConnChange(ctx context.Context,
	product, productPassword string) (*rpc.Subscription, error) {
	logger := h.logger.Add("method", "ConnChange", "product", product)

	logger.Info("subscribing to adapter connection changes")

	_, err := h.checkProductPassword(logger, product, productPassword)
	if err != nil {
		return nil, err
	}

	ntf, ok := rpc.NotifierFromContext(ctx)
	if !ok {
		logger.Error("no notifier found in context")
		return nil, ErrInternal
	}

	sub := ntf.CreateSubscription()
	cb := func(job *data.Job, result error) {
		if result == nil {
			h.handleConnChange(product, logger, ntf, sub, job)
		}
	}
	jobTypes := []string{
		data.JobClientPreServiceSuspend,
		data.JobClientPreServiceUnsuspend,
		data.JobClientPreServiceTerminate,
	}

	// TODO: testing. fix it later.
	sid := string(sub.ID)
	if err = h.queue.Subscribe(jobTypes, sid, cb); err != nil {
		logger.Error(err.Error())
		return nil, ErrInternal
	}

	go func() {
		for err, ok := <-sub.Err(); ok; {
			if err != nil {
				logger.Warn(err.Error())
			}
		}

		err := h.queue.Unsubscribe(jobTypes, sid)
		if err != nil {
			logger.Error(err.Error())
		}
	}()

	return sub, nil
}
// ConnChange subscribes to changes for adapter connection changes.