aggregation

Documentation

Overview

Package aggregation implements Aggregation API

Index

• Constants
• Variables
• func NtfListener(s Storage, ntChan <-chan common.Notification)
• type API
  • func NewAPI(registryClient registry.Client, storage Storage) *API
  • func (api *API) Aggregations() (map[string]Aggregation, error)
  • func (api *API) Filter(w http.ResponseWriter, r *http.Request)
  • func (api *API) Index(w http.ResponseWriter, r *http.Request)
  • func (api *API) Query(w http.ResponseWriter, r *http.Request)
• type Aggregation
• type DataEntry
  • func NewDataEntry() DataEntry
  • func (e *DataEntry) MarshalJSON() ([]byte, error)
• type DataSet
• type Index
• type InfluxAggr
  • func (a *InfluxAggr) NtfCreated(ds registry.DataSource, callback chan error)
  • func (a *InfluxAggr) NtfDeleted(ds registry.DataSource, callback chan error)
  • func (a *InfluxAggr) NtfUpdated(oldDS registry.DataSource, newDS registry.DataSource, callback chan error)
  • func (a *InfluxAggr) Query(aggr registry.Aggregation, q data.Query, page, perPage int, ...) (DataSet, int, error)
• type MongoAggr
  • func (a *MongoAggr) NtfCreated(ds registry.DataSource, callback chan error)
  • func (a *MongoAggr) NtfDeleted(ds registry.DataSource, callback chan error)
  • func (a *MongoAggr) NtfUpdated(oldDS registry.DataSource, newDS registry.DataSource, callback chan error)
  • func (a *MongoAggr) Query(aggr registry.Aggregation, q data.Query, page, perPage int, ...) (DataSet, int, error)
• type RecordSet
• type Storage
  • func NewInfluxAggr(influxStorage *data.InfluxStorage) (Storage, chan<- common.Notification, error)
  • func NewMongoAggr(mongoStorage *data.MongoStorage) (Storage, chan<- common.Notification, error)

Constants

const (
	MaxPerPage = 1000
)

Variables

var (
	ErrNotImplemented = errors.New("API not implemented")
)

Functions

func NtfListener

func NtfListener(s Storage, ntChan <-chan common.Notification)

Handles an incoming notification

Types

type API

type API struct {
	// contains filtered or unexported fields
}

func NewAPI

func NewAPI(registryClient registry.Client, storage Storage) *API

func (*API) Aggregations

func (api *API) Aggregations() (map[string]Aggregation, error)

Retrieve aggregations from registry api

func (*API) Filter

func (api *API) Filter(w http.ResponseWriter, r *http.Request)

func (*API) Index

func (api *API) Index(w http.ResponseWriter, r *http.Request)

func (*API) Query

func (api *API) Query(w http.ResponseWriter, r *http.Request)

type Aggregation

type Aggregation struct {
	ID         string   `json:"id"`
	Interval   string   `json:"interval"`
	Aggregates []string `json:"aggregates"`
	Retention  string   `json:"retention"`
	Sources    []string `json:"sources"`
}

type DataEntry

type DataEntry struct {
	Name       string
	TimeStart  int64
	TimeEnd    int64
	Aggregates map[string]float64
}

func NewDataEntry

func NewDataEntry() DataEntry

func (*DataEntry) MarshalJSON

func (e *DataEntry) MarshalJSON() ([]byte, error)

type DataSet

type DataSet struct {
	BaseName      string      `json:"bn,omitempty"`
	BaseTimeStart int64       `json:"bts,omitempty"`
	BaseTimeEnd   int64       `json:"bte,omitempty"`
	Entries       []DataEntry `json:"e"`
}

type Index

type Index struct {
	Aggrs []Aggregation `json:"aggregations"`
}

type InfluxAggr

type InfluxAggr struct {
	// contains filtered or unexported fields
}

InfluxAggr implements a InfluxDB aggregation client for HDS Data API

func (*InfluxAggr) NtfCreated

func (a *InfluxAggr) NtfCreated(ds registry.DataSource, callback chan error)

NtfCreated handles the creation of a new data source

func (*InfluxAggr) NtfDeleted

func (a *InfluxAggr) NtfDeleted(ds registry.DataSource, callback chan error)

NtfDeleted handles deletion of a data source

func (*InfluxAggr) NtfUpdated

func (a *InfluxAggr) NtfUpdated(oldDS registry.DataSource, newDS registry.DataSource, callback chan error)

NtfUpdated handles updates of a data source

func (*InfluxAggr) Query

func (a *InfluxAggr) Query(aggr registry.Aggregation, q data.Query, page, perPage int, sources ...registry.DataSource) (DataSet, int, error)

Query retrieves aggregated data

type MongoAggr

type MongoAggr struct {
	// contains filtered or unexported fields
}

func (*MongoAggr) NtfCreated

func (a *MongoAggr) NtfCreated(ds registry.DataSource, callback chan error)

func (*MongoAggr) NtfDeleted

func (a *MongoAggr) NtfDeleted(ds registry.DataSource, callback chan error)

func (*MongoAggr) NtfUpdated

func (a *MongoAggr) NtfUpdated(oldDS registry.DataSource, newDS registry.DataSource, callback chan error)

func (*MongoAggr) Query

func (a *MongoAggr) Query(aggr registry.Aggregation, q data.Query, page, perPage int, sources ...registry.DataSource) (DataSet, int, error)

type RecordSet

type RecordSet struct {
	// URL is the URL of the returned recordset in the Data API
	URL string `json:"url"`
	// Data is a SenML object with data records, where
	// Name (bn and n) constitute the resource URL of the corresponding Data Source(s)
	Data DataSet `json:"data"`
	// Time is the time of query in milliseconds
	Time float64 `json:"time"`
	// Page is the current page in Data pagination
	Page int `json:"page"`
	// PerPage is the results per page in Data pagination
	PerPage int `json:"per_page"`
	// Total is the total #of pages in Data pagination
	Total int `json:"total"`
}

RecordSet describes the recordset returned on querying the Aggr API

type Storage

type Storage interface {
	// Queries data for specified data sources
	Query(aggr registry.Aggregation, q data.Query, page, perPage int, sources ...registry.DataSource) (DataSet, int, error)

	// Methods for handling notifications
	NtfCreated(ds registry.DataSource, callback chan error)
	NtfUpdated(old registry.DataSource, new registry.DataSource, callback chan error)
	NtfDeleted(ds registry.DataSource, callback chan error)
}

func NewInfluxAggr

func NewInfluxAggr(influxStorage *data.InfluxStorage) (Storage, chan<- common.Notification, error)

NewInfluxAggr returns a new InfluxAggr

func NewMongoAggr

func NewMongoAggr(mongoStorage *data.MongoStorage) (Storage, chan<- common.Notification, error)