reading

Documentation

Overview

Package reading provides the full application stack for sensor readings. Including transport, persistence and event handling.

Index

• Constants
• type EventHandler
  • func NewEventHandler(readings Repository, logger *log.Logger) *EventHandler
  • func (h *EventHandler) HandleEvent(ctx context.Context, message json.RawMessage) error
• type EventWriter
• type ForEachFunc
• type HTTP
  • func NewHTTP(events EventWriter, logger *log.Logger) *HTTP
  • func (h *HTTP) Ingest(w http.ResponseWriter, r *http.Request)
  • func (h *HTTP) Register(router *mux.Router)
• type IngestResponse
• type PostgresRepository
  • func NewPostgresRepository(db *sql.DB) *PostgresRepository
  • func (pr *PostgresRepository) ForEachOnDate(ctx context.Context, date time.Time, fn ForEachFunc) error
  • func (pr *PostgresRepository) Save(ctx context.Context, reading Reading) error
• type Reading
  • func (r Reading) String() string
  • func (r Reading) Valid() bool
• type Repository
• type SensorType
  • func (st SensorType) Valid() bool

Constants

const (
	SensorTypeSpeed             = SensorType("speed")
	SensorTypeFuel              = SensorType("fuel")
	SensorTypeRevolution        = SensorType("revolution")
	SensorTypeEngineTemperature = SensorType("engine_temperature")
	SensorTypeLocationLatitude  = SensorType("location_latitude")
	SensorTypeLocationLongitude = SensorType("location_longitude")
)

Constants for sensor types.

Types

type EventHandler

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

The EventHandler type is used to handle inbound readings and persist them to the Repository implementation.

func NewEventHandler

func NewEventHandler(readings Repository, logger *log.Logger) *EventHandler

NewEventHandler returns a new instance of the EventHandler type. Inbound events will be stored using the provided Repository implementation.

func (*EventHandler) HandleEvent

func (h *EventHandler) HandleEvent(ctx context.Context, message json.RawMessage) error

HandleEvent handles an inbound JSON payload. It expects the payload to be unmarshallable into a storage.Reading type. Once decoded, the reading is persisted via the Repository implementation.

type EventWriter

type EventWriter interface {
	Write(ctx context.Context, message json.RawMessage) error
}

The EventWriter interface describes types that can publish messages onto an event stream.

type ForEachFunc

type ForEachFunc func(ctx context.Context, reading Reading) error

The ForEachFunc type is a function invoked when calling PostgresRepository.ForEachOnDate.

type HTTP

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

The HTTP type is responsible for handling HTTP requests and publishing events onto an event sink.

func NewHTTP

func NewHTTP(events EventWriter, logger *log.Logger) *HTTP

NewHTTP returns a new instance of the HTTP type that will publish Reading events onto the provided EventWriter implementation to the desired subject. The HTTP.Register method should be used to register the handling methods onto an HTTP router.

func (*HTTP) Ingest

func (h *HTTP) Ingest(w http.ResponseWriter, r *http.Request)

Ingest readings from the request body, publishing each onto the configured EventWriter. This method expects the request body to contain a JSON stream of individual readings. Each reading is validated then published.

func (*HTTP) Register

func (h *HTTP) Register(router *mux.Router)

Register the HTTP's routes onto the HTTP router.

type IngestResponse

type IngestResponse struct {
	Invalid []Reading `json:"invalid,omitempty"`
}

The IngestResponse type is the response DTO when calling HTTP.Ingest. It contains an array of all readings that failed validation.

type PostgresRepository

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

The PostgresRepository type is used to persist Reading data into a PostgreSQL instance.

func NewPostgresRepository

func NewPostgresRepository(db *sql.DB) *PostgresRepository

NewPostgresRepository returns a new instance of the PostgresRepository type that will perform queries against the provided sql.DB instance.

func (*PostgresRepository) ForEachOnDate

func (pr *PostgresRepository) ForEachOnDate(ctx context.Context, date time.Time, fn ForEachFunc) error

ForEachOnDate iterates through all readings stored in the database on the date component of the given time. For each record, the ForEachFunc is invoked. Iteration will stop when there are no more records, the context is cancelled or the ForEachFunc returns an error. Readings are processed in batches of 100 at the time.

func (*PostgresRepository) Save

func (pr *PostgresRepository) Save(ctx context.Context, reading Reading) error

Save the reading to the database. If a reading already exists for the sensor at the given timestamp, do nothing.

type Reading

type Reading struct {
	Sensor    SensorType `json:"sensor"`
	Value     float64    `json:"value"`
	Timestamp time.Time  `json:"timestamp"`
}

The Reading type describes a single sensor reading as stored in the database.

func (Reading) String

func (r Reading) String() string

String returns a string representation of the reading.

func (Reading) Valid

func (r Reading) Valid() bool

Valid returns true if the Reading is deemed to be in a valid state. This means a valid sensor type and a non-zero timestamp.

type Repository

type Repository interface {
	Save(ctx context.Context, reading Reading) error
}

The Repository interface describes types that can store readings.

type SensorType

type SensorType string

The SensorType type describes the kind of sensor that the value relates to.

func (SensorType) Valid

func (st SensorType) Valid() bool

Valid returns true if the SensorType is one of the valid types of sensor.