eventstream

package

v0.0.0-...-b646b39 Latest Latest Go to latest Published: Jan 12, 2021 License: Apache-2.0 Imports: 14 Imported by: 0

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/kexxu-robotics/kex-stream-server

Documentation ¶

Index ¶

type EventMessage
- func ParseRows(rows pgx.Rows) ([]EventMessage, error)
type EventStream
type Handler
- func (h *Handler) AddEvent(w http.ResponseWriter, r *http.Request)
- func (h *Handler) GetOriginEvents(w http.ResponseWriter, r *http.Request)
type Origin
type Secure
- func (s *Secure) Check(id, pass string) (bool, error, string)
- func (s *Secure) ReloadOriginsChron()
type SecureOrigin
type Status

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type EventMessage ¶

type EventMessage struct {

	// message id
	Id                  int64
	EventId             string
	CreationTimeUnixSec int64

	// info about the origin
	OriginId           string
	OriginIter         int64
	OriginGroupId      string
	OriginBuildVersion string
	DestinationId      string

	// the time this event happened
	EventTimeUnixSec int64

	// message type and versioning
	EventType    string
	EventSubtype string
	EventVersion string

	// content of the message
	PayloadJson string
}

EventMessage is the format used to send Events around to save Events in the database

func ParseRows ¶

func ParseRows(rows pgx.Rows) ([]EventMessage, error)

type EventStream ¶

type EventStream struct {
	Conn       *pgxpool.Pool
	MqttClient *mqtt.Client // (optional) MQTT client to notify when a new event is added

	EventStreamId string
	// contains filtered or unexported fields
}

func (*EventStream) GenStreamEventId ¶

func (es *EventStream) GenStreamEventId() string

GenStreamEventId format: <unixtime nana>_<origin iterator uint64>_<origin id> this should usually happen already on the origin, to prevent duplicate inserts

func (*EventStream) GetByDestinationId ¶

func (es *EventStream) GetByDestinationId(destId string, newestId, limit int) ([]EventMessage, error)

GetByDestinationId use -1 for newestId if you start from zero

func (*EventStream) GetByDestinationIdAndEventType ¶

func (es *EventStream) GetByDestinationIdAndEventType(destId, eventType string, newestId, limit int) ([]EventMessage, error)

GetByDestinationIdAndEventType use -1 for newestId if you start from zero

func (*EventStream) GetByDestinationIdAndEventTypePage ¶

func (es *EventStream) GetByDestinationIdAndEventTypePage(destId, eventType string, newestId, lastId, limit int) ([]EventMessage, error)

GetByDestinationIdAndEventTypePage use -1 for newestId if you start from zero

func (*EventStream) GetByDestinationIdPage ¶

func (es *EventStream) GetByDestinationIdPage(destId string, newestId, lastId, limit int) ([]EventMessage, error)

GetByDestinationIdPage use -1 for newestId if you start from zero

func (*EventStream) SaveMessage ¶

func (es *EventStream) SaveMessage(em EventMessage) (EventMessage, error)

SaveMessage

type Handler ¶

type Handler struct {
	Debug       bool
	BaseUrl     string
	StaticPath  string
	Conn        *pgxpool.Pool
	Secure      *Secure
	EventStream *EventStream
}

func (*Handler) AddEvent ¶

func (h *Handler) AddEvent(w http.ResponseWriter, r *http.Request)

func (*Handler) GetOriginEvents ¶

func (h *Handler) GetOriginEvents(w http.ResponseWriter, r *http.Request)

GetOriginEvents gets the events from the provided originId <id> Events are returned paginated, from new to old To get the next page, provide the last (lowest) <lastId> from the previous page All events are returned until <newestId> is reached.

So in a typical situation:

The client would already have events with ids 0, 1, 2, 3, 4, 5, 6 where 6 here is the newestId. It would then try to get the the events it has not got yet, for instance 15, 14, 13, 12, 11, 10 (limit=6, newestId=6) Because the client has not reached its current newestId (6) it will get more (paginated) results: 9, 8, 7 (limit=6, newestId=6, lastId=10)

if no <newestId> is provided, paginated results until the very first events are returned

type Origin ¶

type Origin struct {
	Id       string
	PassHash string
}

type Secure ¶

type Secure struct {
	sync.Mutex

	Conn              *pgxpool.Pool
	LastRefreshed     int64
	MaxRequestsPerMin int64

	Origins map[string]*SecureOrigin
}

func (*Secure) Check ¶

func (s *Secure) Check(id, pass string) (bool, error, string)

func (*Secure) ReloadOriginsChron ¶

func (s *Secure) ReloadOriginsChron()

type SecureOrigin ¶

type SecureOrigin struct {
	Id          string
	PassHash    string
	ReqsLastMin int64
}

type Status ¶

type Status struct {
	Id         int64
	OriginIter int64
	UnixSec    int64
}

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL