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
}
func ( h * Handler ) ObjectChange ( ctx context . Context , tkn , objectType string , objectIDs [ ] string ) ( * rpc . Subscription , error ) { logger := h . logger . Add ( " " , " " , " " , objectType , " " , objectIDs ) if ! h . token . Check ( tkn ) { logger . Warn ( " " ) 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 ( " " ) 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.
ObjectChange subscribes to changes for objects of a given type .