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,
	}

	
	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 ( " " , " " , " " , product ) logger . Info ( " " ) _ , err := h . checkProductPassword ( logger , product , productPassword ) if err != nil { return nil , err } ntf , ok := rpc . NotifierFromContext ( ctx ) if ! ok { logger . Error ( " " ) 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.
ConnChange subscribes to changes for adapter connection changes .