data

package
v0.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 14, 2018 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Overview

Package data implements Data API

Index

Constants

View Source 
const (
	MaxPerPage = 1000
)

Variables

View Source 
var SupportedContentTypes = map[string]bool{
	"application/senml+json": true,
}

Supported content-types for data ingestion

Functions

func NewMQTTConnector 

func NewMQTTConnector(registryClient registry.Client, storage Storage) (chan<- common.Notification, error)

func NtfListener 

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

Handles an incoming notification

func NtfListenerMQTT 

func NtfListenerMQTT(c *MQTTConnector, ntChan <-chan common.Notification)

Handles an incoming notification

Types

type HTTPAPI 

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

HTTPAPI describes the RESTful HTTP data API

func NewHTTPAPI 

func NewHTTPAPI(registryClient registry.Client, storage Storage, autoRegistration bool) *HTTPAPI

NewHTTPAPI returns the configured Data API

func (*HTTPAPI) Query 

func (d *HTTPAPI) Query(w http.ResponseWriter, r *http.Request)

Query is a handler for querying data Expected parameters: id(s), optional: pagination, query string

func (*HTTPAPI) Submit 

func (d *HTTPAPI) Submit(w http.ResponseWriter, r *http.Request)

Submit is a handler for submitting a new data point Expected parameters: id(s) TODO: check SupportedContentTypes instead of hard-coding SenML

func (*HTTPAPI) SubmitWithoutID 

func (d *HTTPAPI) SubmitWithoutID(w http.ResponseWriter, r *http.Request)

SubmitWithoutID is a handler for submitting a new data point Expected parameters: none

type InfluxStorage 

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

InfluxStorage implements a InfluxDB storage client for HDS Data API

func NewInfluxStorage 

func NewInfluxStorage(conf common.DataConf, retentionPeriods []string) (*InfluxStorage, chan<- common.Notification, error)

NewInfluxStorage returns a new InfluxStorage

func (*InfluxStorage) ChangeRetentionPolicy added in v0.5.2 

func (s *InfluxStorage) ChangeRetentionPolicy(measurement, countField, oldRP, newRP string) error

func (*InfluxStorage) CountSprintf 

func (s *InfluxStorage) CountSprintf(format string, a ...interface{}) (int64, error)

CountSprintf constructs a counting query for influxdb

func (*InfluxStorage) Database 

func (s *InfluxStorage) Database() string

Database returns database name

func (*InfluxStorage) FieldForType 

func (s *InfluxStorage) FieldForType(t string) string

FieldForType returns the field-name for HDS data types

func (*InfluxStorage) MeasurementName added in v0.5.0 

func (s *InfluxStorage) MeasurementName(id string) string

MeasurementName returns formatted measurement name for a given data source

func (*InfluxStorage) MeasurementNameFQ added in v0.5.0 

func (s *InfluxStorage) MeasurementNameFQ(retention, measurementName string) string

MeasurementNameFQ returns formatted fully-qualified measurement name

func (*InfluxStorage) NtfCreated 

func (s *InfluxStorage) NtfCreated(ds registry.DataSource, callback chan error)

NtfCreated handles the creation of a new data source

func (*InfluxStorage) NtfDeleted 

func (s *InfluxStorage) NtfDeleted(ds registry.DataSource, callback chan error)

NtfDeleted handles deletion of a data source

func (*InfluxStorage) NtfUpdated 

func (s *InfluxStorage) NtfUpdated(oldDS registry.DataSource, newDS registry.DataSource, callback chan error)

NtfUpdated handles updates of a data source

func (*InfluxStorage) ParseDuration added in v0.5.2 

func (s *InfluxStorage) ParseDuration(durationStr string) (time.Duration, error)

func (*InfluxStorage) Query 

func (s *InfluxStorage) Query(q Query, page, perPage int, sources ...*registry.DataSource) (senml.SenML, int, error)

Query retrieves data for specified data sources

func (*InfluxStorage) QuerySprintf 

func (s *InfluxStorage) QuerySprintf(format string, a ...interface{}) (res []influx.Result, err error)

QuerySprintf constructs a query for influxdb

func (*InfluxStorage) Replication 

func (s *InfluxStorage) Replication() int

Replication returns Influxdb Replication factor

func (*InfluxStorage) RetentionPolicyName added in v0.5.0 

func (s *InfluxStorage) RetentionPolicyName(period string) string

RetentionPolicyName returns formatted retention policy name for a given period

func (*InfluxStorage) Submit 

func (s *InfluxStorage) Submit(data map[string][]senml.SenMLRecord, sources map[string]*registry.DataSource) error

Submit adds multiple data points for multiple data sources data is a map where keys are data source ids

type MQTTConnector 

type MQTTConnector struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func (*MQTTConnector) NtfCreated 

func (c *MQTTConnector) NtfCreated(ds registry.DataSource, callback chan error)

Handles the creation of a new data source

func (*MQTTConnector) NtfDeleted 

func (c *MQTTConnector) NtfDeleted(oldDS registry.DataSource, callback chan error)

Handles deletion of a data source

func (*MQTTConnector) NtfUpdated 

func (c *MQTTConnector) NtfUpdated(oldDS registry.DataSource, newDS registry.DataSource, callback chan error)

Handles updates of a data source

type Manager 

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

type Query 

type Query struct {
	Start time.Time
	End   time.Time
	Sort  string
	Limit int
}

func ParseQueryParameters 

func ParseQueryParameters(form url.Values) (Query, 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 []senml.SenMLRecord `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 Data API

type RemoteClient 

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

func NewRemoteClient 

func NewRemoteClient(serverEndpoint string, ticket *obtainer.Client) (*RemoteClient, error)

func (*RemoteClient) Query 

func (c *RemoteClient) Query(q Query, page, perPage int, id ...string) (*RecordSet, error)

func (*RemoteClient) Submit 

func (c *RemoteClient) Submit(data []byte, contentType string, id ...string) error

Submit data for ingestion, where: data - is a byte array with actual data contentType - mime-type of the data (will be set in the header) id... - ID (or array of IDs) of data sources for which the data is being submitted

type Storage 

type Storage interface {
	// Adds data points for multiple data sources
	// data is a map where keys are data source ids
	// sources is a map where keys are data source ids
	Submit(data map[string][]senml.SenMLRecord, sources map[string]*registry.DataSource) error

	// Queries data for specified data sources
	Query(q Query, page, perPage int, sources ...*registry.DataSource) (senml.SenML, 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)
}

Storage is an interface of a Data storage backend

type Subscription 

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

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.