func (h *Handler) ObjectChange(ctx context.Context, tkn, objectType string,
	objectIDs []string) (*rpc.Subscription, error) {
	logger := h.logger.Add("method", "ObjectChange",
		"objectType", objectType, "objectIDs", objectIDs)

	if !h.token.Check(tkn) {
		logger.Warn("access denied")
		return nil, ErrAccessDenied
	}

	table, ok := objectChangeTables[objectType]
	if !ok {
		logger.Warn(ErrBadObjectType.Error())
		return nil, ErrBadObjectType
	}

	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) {
		obj, err := h.db.FindByPrimaryKeyFrom(table, job.RelatedID)
		if err != nil {
			logger.Error(err.Error())
		}

		var odata json.RawMessage
		if obj != nil {
			odata, err = json.Marshal(obj)
			if err != nil {
				logger.Error(err.Error())
			}
		}

		err = ntf.Notify(sub.ID,
			&ObjectChangeResult{odata, job, rpcsrv.ToError(result)})
		if err != nil {
			logger.Warn(err.Error())
		}
	}

	sid := string(sub.ID)
	err := h.queue.Subscribe(objectIDs, sid, cb)
	if 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(objectIDs, sid)
		if err != nil {
			logger.Error(err.Error())
		}
	}()

	return sub, nil
}
func (h *Handler) ObjectChange(ctx context.Context, tkn, objectType string,
	objectIDs []string) (*rpc.Subscription, error) {
	logger := h.logger.Add("method", "ObjectChange",
		"objectType", objectType, "objectIDs", objectIDs)

	if !h.token.Check(tkn) {
		logger.Warn("access denied")
		return nil, ErrAccessDenied
	}

	table, ok := objectChangeTables[objectType]
	if !ok {
		logger.Warn(ErrBadObjectType.Error())
		return nil, ErrBadObjectType
	}

	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) {
		obj, err := h.db.FindByPrimaryKeyFrom(table, job.RelatedID)
		if err != nil {
			logger.Error(err.Error())
		}

		var odata json.RawMessage
		if obj != nil {
			odata, err = json.Marshal(obj)
			if err != nil {
				logger.Error(err.Error())
			}
		}

		err = ntf.Notify(sub.ID,
			&ObjectChangeResult{odata, job, rpcsrv.ToError(result)})
		if err != nil {
			logger.Warn(err.Error())
		}
	}

	sid := string(sub.ID)
	err := h.queue.Subscribe(objectIDs, sid, cb)
	if 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(objectIDs, sid)
		if err != nil {
			logger.Error(err.Error())
		}
	}()

	return sub, nil
}
// ObjectChange subscribes to changes for objects of a given type.