node

Documentation

Index

• func DeepCopyMap(m map[string]interface{}) map[string]interface{}
• func DeepCopySlice(s []interface{}) []interface{}
• type BaseNode
  • func NewBaseNode(command *command.CreateNode) *BaseNode
  • func (node *BaseNode) Activate(cmd command.Command)
  • func (node *BaseNode) AddChild(child Node)
  • func (node *BaseNode) AddPipeline(child Node, pipeline pipeline.Pipeline)
  • func (node *BaseNode) Deactivate(cmd command.Command)
  • func (node *BaseNode) GetActive() bool
  • func (node *BaseNode) GetChildren() map[Node]pipeline.Pipeline
  • func (node *BaseNode) GetID() string
  • func (node *BaseNode) GetName() string
  • func (node *BaseNode) GetType() string
  • func (node *BaseNode) HasChild(child Node) bool
  • func (node *BaseNode) Receive(cmd command.Command)
  • func (node *BaseNode) RemoveChild(child Node)
  • func (node *BaseNode) RemovePipeline(child Node)
  • func (node *BaseNode) Send(cmd command.Command)
  • func (node *BaseNode) SetActive(active bool)
  • func (node *BaseNode) ToJSON() ([]byte, error)
  • func (node *BaseNode) ToJSONStruct() map[string]interface{}
• type Mongo
  • func NewMongoNode(cmd *command.CreateNode) *Mongo
  • func (node *Mongo) Add(cmd *command.AddMongo)
  • func (node *Mongo) Connect(cmd *command.ConnectMongo)
  • func (node *Mongo) Delete() error
  • func (node *Mongo) QueryAll(cmd *command.QueryAllMongo)
  • func (node *Mongo) Remove(cmd *command.RemoveMongo)
  • func (node *Mongo) RemoveByID(cmd *command.RemoveByIDMongo)
  • func (node *Mongo) ToJSONStruct() map[string]interface{}
  • func (node *Mongo) Update(cmd *command.UpdateMongo)
  • func (node *Mongo) UpdateByID(cmd *command.UpdateByIDMongo)
• type Node
  • func NewNode(command *command.CreateNode) Node
• type Publisher
  • func NewPublisherNode(cmd *command.CreateNode) *Publisher
  • func (node *Publisher) AddSubscriber(cmd *command.AddSubscriber) error
  • func (node *Publisher) Receive(cmd command.Command)
  • func (node *Publisher) RemoveSubscriber(subscriber *websocket.Conn)
  • func (node *Publisher) ToJSONStruct() map[string]interface{}
• type Subscriber
  • func NewSubscriberNode(cmd *command.CreateNode) *Subscriber
  • func (node *Subscriber) ActivateWS(cmd *command.ActivateWS)
  • func (node *Subscriber) Close()
  • func (node *Subscriber) DeactivateWS(cmd *command.DeactivateWS)
  • func (node *Subscriber) Listen(cmd command.Command)
  • func (node *Subscriber) ToJSONStruct() map[string]interface{}
  • func (node *Subscriber) UpdateURL(cmd *command.UpdateURL)

Functions

func DeepCopyMap

func DeepCopyMap(m map[string]interface{}) map[string]interface{}

func DeepCopySlice

func DeepCopySlice(s []interface{}) []interface{}

Types

type BaseNode

type BaseNode struct {
	ID        string `json:"id"`
	Name      string `json:"name"`
	Active    bool   `json:"activate"`
	Type      string `json:"type"`
	Children  map[Node]pipeline.Pipeline
	SyncMutex sync.Mutex
}

func NewBaseNode

func NewBaseNode(command *command.CreateNode) *BaseNode

func (*BaseNode) Activate

func (node *BaseNode) Activate(cmd command.Command)

func (*BaseNode) AddChild

func (node *BaseNode) AddChild(child Node)

func (*BaseNode) AddPipeline

func (node *BaseNode) AddPipeline(child Node, pipeline pipeline.Pipeline)

Addpipeline adds a pipeline from the parent to the child node. This can be done concurrently.

func (*BaseNode) Deactivate

func (node *BaseNode) Deactivate(cmd command.Command)

func (*BaseNode) GetActive

func (node *BaseNode) GetActive() bool

func (*BaseNode) GetChildren

func (node *BaseNode) GetChildren() map[Node]pipeline.Pipeline

func (*BaseNode) GetID

func (node *BaseNode) GetID() string

func (*BaseNode) GetName

func (node *BaseNode) GetName() string

func (*BaseNode) GetType

func (node *BaseNode) GetType() string

func (*BaseNode) HasChild

func (node *BaseNode) HasChild(child Node) bool

func (*BaseNode) Receive

func (node *BaseNode) Receive(cmd command.Command)

func (*BaseNode) RemoveChild

func (node *BaseNode) RemoveChild(child Node)

func (*BaseNode) RemovePipeline

func (node *BaseNode) RemovePipeline(child Node)

RemovePipeline removes a pipeline from parent to the child node. This can be done concurrently.

func (*BaseNode) Send

func (node *BaseNode) Send(cmd command.Command)

func (*BaseNode) SetActive

func (node *BaseNode) SetActive(active bool)

func (*BaseNode) ToJSON

func (node *BaseNode) ToJSON() ([]byte, error)

func (*BaseNode) ToJSONStruct

func (node *BaseNode) ToJSONStruct() map[string]interface{}

type Mongo

type Mongo struct {
	BaseNode
	// contains filtered or unexported fields
}

func NewMongoNode

func NewMongoNode(cmd *command.CreateNode) *Mongo

func (*Mongo) Add

func (node *Mongo) Add(cmd *command.AddMongo)

func (*Mongo) Connect

func (node *Mongo) Connect(cmd *command.ConnectMongo)

func (*Mongo) Delete

func (node *Mongo) Delete() error

func (*Mongo) QueryAll

func (node *Mongo) QueryAll(cmd *command.QueryAllMongo)

func (*Mongo) Remove

func (node *Mongo) Remove(cmd *command.RemoveMongo)

func (*Mongo) RemoveByID

func (node *Mongo) RemoveByID(cmd *command.RemoveByIDMongo)

func (*Mongo) ToJSONStruct

func (node *Mongo) ToJSONStruct() map[string]interface{}

func (*Mongo) Update

func (node *Mongo) Update(cmd *command.UpdateMongo)

func (*Mongo) UpdateByID

func (node *Mongo) UpdateByID(cmd *command.UpdateByIDMongo)

type Node

type Node interface {
	GetID() string
	GetName() string
	GetActive() bool
	SetActive(bool)
	GetType() string

	Activate(command.Command)
	Deactivate(command.Command)

	Send(command.Command)
	Receive(command.Command)

	GetChildren() map[Node]pipeline.Pipeline
	AddChild(Node)
	RemoveChild(Node)
	HasChild(Node) bool

	AddPipeline(Node, pipeline.Pipeline)
	RemovePipeline(Node)

	ToJSON() ([]byte, error)
	ToJSONStruct() map[string]interface{}
}

Node is an interface for all nodes in the flow tree.

func NewNode

func NewNode(command *command.CreateNode) Node

type Publisher

type Publisher struct {
	BaseNode
	Subscribers map[*websocket.Conn]struct{}
	Mu          sync.Mutex
}

func NewPublisherNode

func NewPublisherNode(cmd *command.CreateNode) *Publisher

func (*Publisher) AddSubscriber

func (node *Publisher) AddSubscriber(cmd *command.AddSubscriber) error

func (*Publisher) Receive

func (node *Publisher) Receive(cmd command.Command)

func (*Publisher) RemoveSubscriber

func (node *Publisher) RemoveSubscriber(subscriber *websocket.Conn)

func (*Publisher) ToJSONStruct

func (node *Publisher) ToJSONStruct() map[string]interface{}

type Subscriber

type Subscriber struct {
	BaseNode
	URL      string          `json:"url"`
	WSActive bool            `json:"wsactive"`
	Signal   chan bool       `json:"-"`
	Client   *websocket.Conn `json:"-"`
}

func NewSubscriberNode

func NewSubscriberNode(cmd *command.CreateNode) *Subscriber

func (*Subscriber) ActivateWS

func (node *Subscriber) ActivateWS(cmd *command.ActivateWS)

func (*Subscriber) Close

func (node *Subscriber) Close()

func (*Subscriber) DeactivateWS

func (node *Subscriber) DeactivateWS(cmd *command.DeactivateWS)

func (*Subscriber) Listen

func (node *Subscriber) Listen(cmd command.Command)

func (*Subscriber) ToJSONStruct

func (node *Subscriber) ToJSONStruct() map[string]interface{}

func (*Subscriber) UpdateURL

func (node *Subscriber) UpdateURL(cmd *command.UpdateURL)