lib

Documentation

Index

• Variables
• func CloseDB()
• func CreateServer()
• func DeletePipelineAdminEndpoint(w http.ResponseWriter, req *http.Request)
• func DeletePipelineEndpoint(w http.ResponseWriter, req *http.Request)
• func GetDB() *mongo.Client
• func GetEnv(key, fallback string) string
• func GetPipelineEndpoint(w http.ResponseWriter, req *http.Request)
• func GetPipelinesAdminEndpoint(w http.ResponseWriter, req *http.Request)
• func GetPipelinesEndpoint(w http.ResponseWriter, req *http.Request)
• func GetRootEndpoint(w http.ResponseWriter, req *http.Request)
• func InitDB()
• func IntInSlice(a int, list []int) bool
• func Mongo() *mongo.Collection
• func PostPipelineEndpoint(w http.ResponseWriter, req *http.Request)
• func PutPipelineEndpoint(w http.ResponseWriter, req *http.Request)
• func ToJson(resp string) map[string]interface{}
• type Claims
  • func (c Claims) Valid() error
• type DownstreamConfig
• type InputSelection
• type InputTopic
• type LoggerMiddleWare
  • func NewLogger(handler http.Handler, logLevel string) *LoggerMiddleWare
  • func (this *LoggerMiddleWare) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type Mapping
• type MockRepo
  • func NewMockRepo() *MockRepo
  • func (r *MockRepo) All(userId string, admin bool, args map[string][]string) (pipelines []Pipeline, err error)
  • func (r *MockRepo) DeletePipeline(id string, userId string, admin bool) (err error)
  • func (r *MockRepo) FindPipeline(id string, userId string) (pipeline Pipeline, err error)
  • func (r *MockRepo) InsertPipeline(pipeline Pipeline) (err error)
  • func (r *MockRepo) UpdatePipeline(pipeline Pipeline, userId string) (err error)
• type MongoRepo
  • func NewMongoRepo() *MongoRepo
  • func (r *MongoRepo) All(userId string, admin bool, args map[string][]string) (pipelines []Pipeline, err error)
  • func (r *MongoRepo) DeletePipeline(id string, userId string, admin bool) (err error)
  • func (r *MongoRepo) FindPipeline(id string, userId string) (pipeline Pipeline, err error)
  • func (r *MongoRepo) InsertPipeline(pipeline Pipeline) (err error)
  • func (r *MongoRepo) UpdatePipeline(pipeline Pipeline, userId string) (err error)
• type Operator
• type Pipeline
• type PipelineRepository
• type PipelineResponse
• type Registry
  • func NewRegistry(repository PipelineRepository) *Registry
  • func (r *Registry) DeletePipeline(id string, userId string) (Response, error)
  • func (r *Registry) DeletePipelineAdmin(id string, userId string) (Response, error)
  • func (r *Registry) GetPipeline(id string, userId string) (pipeline Pipeline, err error)
  • func (r *Registry) GetPipelines(userId string, args map[string][]string) (pipelines []Pipeline, err error)
  • func (r *Registry) GetPipelinesAdmin(userId string, args map[string][]string) (pipelines []Pipeline, err error)
  • func (r *Registry) SavePipeline(pipeline Pipeline, userId string) (id uuid.UUID, err error)
  • func (r *Registry) UpdatePipeline(pipeline Pipeline, userId string) (id uuid.UUID, err error)
• type Response
• type UpstreamConfig

Variables

var CTX mongo.SessionContext

var DB *mongo.Client

var REGISTRY = NewRegistry(NewMongoRepo())

Functions

func CloseDB

func CloseDB()

func CreateServer

func CreateServer()

func DeletePipelineAdminEndpoint

func DeletePipelineAdminEndpoint(w http.ResponseWriter, req *http.Request)

func DeletePipelineEndpoint

func DeletePipelineEndpoint(w http.ResponseWriter, req *http.Request)

func GetDB

func GetDB() *mongo.Client

func GetEnv

func GetEnv(key, fallback string) string

func GetPipelineEndpoint

func GetPipelineEndpoint(w http.ResponseWriter, req *http.Request)

func GetPipelinesAdminEndpoint

func GetPipelinesAdminEndpoint(w http.ResponseWriter, req *http.Request)

func GetPipelinesEndpoint

func GetPipelinesEndpoint(w http.ResponseWriter, req *http.Request)

func GetRootEndpoint

func GetRootEndpoint(w http.ResponseWriter, req *http.Request)

func InitDB

func InitDB()

func IntInSlice

func IntInSlice(a int, list []int) bool

func Mongo

func Mongo() *mongo.Collection

func PostPipelineEndpoint

func PostPipelineEndpoint(w http.ResponseWriter, req *http.Request)

func PutPipelineEndpoint

func PutPipelineEndpoint(w http.ResponseWriter, req *http.Request)

func ToJson

func ToJson(resp string) map[string]interface{}

Types

type Claims

type Claims struct {
	Sub         string              `json:"sub,omitempty"`
	RealmAccess map[string][]string `json:"realm_access,omitempty"`
}

func (Claims) Valid

func (c Claims) Valid() error

type DownstreamConfig

type DownstreamConfig struct {
	Enabled    bool
	InstanceID string
}

type InputSelection

type InputSelection struct {
	InputName         string   `json:"inputName,omitempty"` // references mapping name
	AspectId          string   `json:"aspectId,omitempty"`
	FunctionId        string   `json:"functionId,omitempty"`
	CharacteristicIds []string `json:"characteristicIds,omitempty"`
	SelectableId      string   `json:"selectableId,omitempty"` // either device or group. can be used for SNRGY-1172, needed to update devices in group
}

type InputTopic

type InputTopic struct {
	Name        string    `json:"name,omitempty"`
	FilterType  string    `json:"filterType,omitempty"`
	FilterValue string    `json:"filterValue,omitempty"`
	Mappings    []Mapping `json:"mappings,omitempty"`
}

type LoggerMiddleWare

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

func NewLogger

func NewLogger(handler http.Handler, logLevel string) *LoggerMiddleWare

func (*LoggerMiddleWare) ServeHTTP

func (this *LoggerMiddleWare) ServeHTTP(w http.ResponseWriter, r *http.Request)

type Mapping

type Mapping struct {
	Dest   string `json:"dest,omitempty"`
	Source string `json:"source,omitempty"`
}

type MockRepo

type MockRepo struct {
}

func NewMockRepo

func NewMockRepo() *MockRepo

func (*MockRepo) All

func (r *MockRepo) All(userId string, admin bool, args map[string][]string) (pipelines []Pipeline, err error)

func (*MockRepo) DeletePipeline

func (r *MockRepo) DeletePipeline(id string, userId string, admin bool) (err error)

func (*MockRepo) FindPipeline

func (r *MockRepo) FindPipeline(id string, userId string) (pipeline Pipeline, err error)

func (*MockRepo) InsertPipeline

func (r *MockRepo) InsertPipeline(pipeline Pipeline) (err error)

func (*MockRepo) UpdatePipeline

func (r *MockRepo) UpdatePipeline(pipeline Pipeline, userId string) (err error)

type MongoRepo

type MongoRepo struct {
}

func NewMongoRepo

func NewMongoRepo() *MongoRepo

func (*MongoRepo) All

func (r *MongoRepo) All(userId string, admin bool, args map[string][]string) (pipelines []Pipeline, err error)

func (*MongoRepo) DeletePipeline

func (r *MongoRepo) DeletePipeline(id string, userId string, admin bool) (err error)

func (*MongoRepo) FindPipeline

func (r *MongoRepo) FindPipeline(id string, userId string) (pipeline Pipeline, err error)

func (*MongoRepo) InsertPipeline

func (r *MongoRepo) InsertPipeline(pipeline Pipeline) (err error)

func (*MongoRepo) UpdatePipeline

func (r *MongoRepo) UpdatePipeline(pipeline Pipeline, userId string) (err error)

type Operator

type Operator struct {
	Id               string            `json:"id,omitempty"`
	Name             string            `json:"name,omitempty"`
	ApplicationId    uuid.UUID         `json:"applicationId,omitempty"`
	ImageId          string            `json:"imageId,omitempty"`
	DeploymentType   string            `json:"deploymentType,omitempty"`
	OperatorId       string            `json:"operatorId,omitempty"`
	Config           map[string]string `json:"config,omitempty"`
	OutputTopic      string            `json:"outputTopic,omitempty"`
	InputTopics      []InputTopic      `json:"inputTopics,omitempty"`
	InputSelections  []InputSelection  `json:"inputSelections,omitempty"`
	PersistData      bool              `json:"persistData,omitempty"`
	Cost             uint              `json:"cost"`
	UpstreamConfig   UpstreamConfig    `json:"upstream,omitempty"`
	DownstreamConfig DownstreamConfig  `json:"downstream,omitempty"`
}

type Pipeline

type Pipeline struct {
	Id                 string    `bson:"id" json:"id"`
	Name               string    `json:"name,omitempty"`
	Description        string    `json:"description,omitempty"`
	FlowId             string    `json:"flowId,omitempty"`
	Image              string    `json:"image,omitempty"`
	WindowTime         int       `json:"windowTime,omitempty"`
	MergeStrategy      string    `json:"mergeStrategy,omitempty"`
	ConsumeAllMessages bool      `json:"consumeAllMessages,omitempty"`
	Metrics            bool      `json:"metrics,omitempty"`
	CreatedAt          time.Time `json:"createdAt,omitempty"`
	UpdatedAt          time.Time `json:"updatedAt,omitempty"`
	UserId             string
	Operators          []Operator `json:"operators,omitempty"`
}

type PipelineRepository

type PipelineRepository interface {
	InsertPipeline(pipeline Pipeline) (err error)
	UpdatePipeline(pipeline Pipeline, userId string) (err error)
	All(userId string, admin bool, args map[string][]string) (pipelines []Pipeline, err error)
	FindPipeline(id string, userId string) (pipeline Pipeline, err error)
	DeletePipeline(id string, userId string, admin bool) (err error)
}

type PipelineResponse

type PipelineResponse struct {
	Id uuid.UUID `json:"id,omitempty"`
}

type Registry

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

func NewRegistry

func NewRegistry(repository PipelineRepository) *Registry

func (*Registry) DeletePipeline

func (r *Registry) DeletePipeline(id string, userId string) (Response, error)

func (*Registry) DeletePipelineAdmin

func (r *Registry) DeletePipelineAdmin(id string, userId string) (Response, error)

func (*Registry) GetPipeline

func (r *Registry) GetPipeline(id string, userId string) (pipeline Pipeline, err error)

func (*Registry) GetPipelines

func (r *Registry) GetPipelines(userId string, args map[string][]string) (pipelines []Pipeline, err error)

func (*Registry) GetPipelinesAdmin

func (r *Registry) GetPipelinesAdmin(userId string, args map[string][]string) (pipelines []Pipeline, err error)

func (*Registry) SavePipeline

func (r *Registry) SavePipeline(pipeline Pipeline, userId string) (id uuid.UUID, err error)

func (*Registry) UpdatePipeline

func (r *Registry) UpdatePipeline(pipeline Pipeline, userId string) (id uuid.UUID, err error)

type Response

type Response struct {
	Message string `json:"message,omitempty"`
}

type UpstreamConfig

type UpstreamConfig struct {
	Enabled bool
}