// Subscribe subscribes for data change
Subscribe subscribes for data change
// Subscribe subscribes for data change
func (provider *DataProvider) Subscribe(path string, authInfo *AuthInfo) (id uint64, channel <-chan interface{}, err error) {
	provider.Lock()
	defer provider.Unlock()
	log.WithFields(log.Fields{"subscribeID": provider.currentSubsID, "path": path}).Debug("Subscribe")
	filter, err := CreatePathFilter(path)
	if err != nil {
		return id, channel, err
	}
	subscribeMap := make(map[DataAdapter][]string)
	for path, sensor := range provider.sensors {
		if filter.Match(path) {
			if err = checkPermissions(sensor.adapter, path, authInfo, "r"); err != nil {
				return id, channel, err
			}
			sensor.subscribeIds.PushBack(provider.currentSubsID)
			if subscribeMap[sensor.adapter] == nil {
				subscribeMap[sensor.adapter] = make([]string, 0, 10)
			}
			subscribeMap[sensor.adapter] = append(subscribeMap[sensor.adapter], path)
		}
	}
	if len(subscribeMap) == 0 {
		return id, channel, errors.New("specified data path does not exist")
	}
	for adapter, pathList := range subscribeMap {
		for _, path := range pathList {
			log.WithFields(log.Fields{"adapter": adapter.GetName(), "path": path}).Debug("Subscribe for adapter data")
		}
		if err = adapter.Subscribe(pathList); err != nil {
			return id, channel, err
		}
	}
	id = provider.currentSubsID
	dataChannel := make(chan interface{}, subscribeChannelSize)
	provider.subscribeInfoMap[id] = &subscribeInfo{channel: dataChannel, path: path}
	provider.currentSubsID++
	return id, dataChannel, nil
}
func ( provider * DataProvider ) Subscribe ( path string , authInfo * AuthInfo ) ( id uint64 , channel <- chan interface { } , err error ) { provider . Lock ( ) defer provider . Unlock ( ) log . WithFields ( log . Fields { " " : provider . currentSubsID , " " : path } ) . Debug ( " " ) filter , err := CreatePathFilter ( path ) if err != nil { return id , channel , err } subscribeMap := make ( map [ DataAdapter ] [ ] string ) for path , sensor := range provider . sensors { if filter . Match ( path ) { if err = checkPermissions ( sensor . adapter , path , authInfo , " " ) ; err != nil { return id , channel , err } sensor . subscribeIds . PushBack ( provider . currentSubsID ) if subscribeMap [ sensor . adapter ] == nil { subscribeMap [ sensor . adapter ] = make ( [ ] string , 0 , 10 ) } subscribeMap [ sensor . adapter ] = append ( subscribeMap [ sensor . adapter ] , path ) } } if len ( subscribeMap ) == 0 { return id , channel , errors . New ( " " ) } for adapter , pathList := range subscribeMap { for _ , path := range pathList { log . WithFields ( log . Fields { " " : adapter . GetName ( ) , " " : path } ) . Debug ( " " ) } if err = adapter . Subscribe ( pathList ) ; err != nil { return id , channel , err } } id = provider . currentSubsID dataChannel := make ( chan interface { } , subscribeChannelSize ) provider . subscribeInfoMap [ id ] = & subscribeInfo { channel : dataChannel , path : path } provider . currentSubsID ++ return id , dataChannel , nil }
Subscribe subscribes for data change
Subscribe subscribes for data change
// Create map of pathes grouped by adapter // Get data from adapter and group it by parent // Add subscribe id to subscribe list // Add path to subscribeMap // Subscribe for adapter data changes