signalflow

Documentation

Index

• type AuthRequest
• type AuthType
  • func (at AuthType) MarshalJSON() ([]byte, error)
• type Channel
  • func (c *Channel) AcceptMessage(msg messages.Message)
  • func (c *Channel) Close()
  • func (c *Channel) Messages() <-chan messages.Message
• type Client
  • func NewClient(options ...ClientParam) (*Client, error)
  • func (c *Client) Close()
  • func (c *Client) Execute(req *ExecuteRequest) (*Computation, error)
  • func (c *Client) Stop(req *StopRequest) error
• type ClientParam
  • func AccessToken(token string) ClientParam
  • func MetadataTimeout(timeout time.Duration) ClientParam
  • func ReadTimeout(timeout time.Duration) ClientParam
  • func StreamURL(streamEndpoint string) ClientParam
  • func StreamURLForRealm(realm string) ClientParam
  • func UserAgent(userAgent string) ClientParam
  • func WriteTimeout(timeout time.Duration) ClientParam
• type Computation
  • func (c *Computation) Channel() *Channel
  • func (c *Computation) Data() <-chan *messages.DataMessage
  • func (c *Computation) Done() <-chan struct{}
  • func (c *Computation) Err() error
  • func (c *Computation) Handle() string
  • func (c *Computation) IsFinished() bool
  • func (c *Computation) Lag() time.Duration
  • func (c *Computation) MaxDelay() time.Duration
  • func (c *Computation) Resolution() time.Duration
  • func (c *Computation) Stop() error
  • func (c *Computation) StopWithReason(reason string) error
  • func (c *Computation) TSIDMetadata(tsid idtool.ID) *messages.MetadataProperties
• type DetachRequest
• type DetachType
  • func (DetachType) MarshalJSON() ([]byte, error)
• type ExecuteRequest
  • func (er ExecuteRequest) MarshalJSON() ([]byte, error)
• type ExecuteType
  • func (ExecuteType) MarshalJSON() ([]byte, error)
• type StopRequest
• type StopType
  • func (StopType) MarshalJSON() ([]byte, error)

Types

type AuthRequest

type AuthRequest struct {
	Type AuthType `json:"type"`
	// The Auth token for the org
	Token     string `json:"token"`
	UserAgent string `json:"userAgent,omitempty"`
}

type AuthType

type AuthType string

func (AuthType) MarshalJSON

func (at AuthType) MarshalJSON() ([]byte, error)

type Channel

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

Channel is a queue of messages that all pertain to the same computation.

func (*Channel) AcceptMessage

func (c *Channel) AcceptMessage(msg messages.Message)

AcceptMessage from a websocket. This might block if nothing is reading from the channel but generally a computation should always be doing so.

func (*Channel) Close

func (c *Channel) Close()

Close the channel. This does not actually stop a job in SignalFlow, for that use Computation.Stop().

func (*Channel) Messages

func (c *Channel) Messages() <-chan messages.Message

Messages returns a Go chan that will be pushed all of the deserialized SignalFlow messages from the websocket.

type Client

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

Client for SignalFlow via websockets (SSE is not currently supported).

func NewClient

func NewClient(options ...ClientParam) (*Client, error)

NewClient makes a new, but uninitialized, SignalFlow client.

func (*Client) Close

func (c *Client) Close()

Close the client and shutdown any ongoing connections and goroutines.

func (*Client) Execute

func (c *Client) Execute(req *ExecuteRequest) (*Computation, error)

Execute a SignalFlow job and return a channel upon which informational messages and data will flow.

func (*Client) Stop

func (c *Client) Stop(req *StopRequest) error

Stop sends a job stop request message to the backend. It does not wait for jobs to actually be stopped.

type ClientParam

type ClientParam func(*Client) error

ClientParam is the common type of configuration functions for the SignalFlow client

func AccessToken

func AccessToken(token string) ClientParam

AccessToken can be used to provide a SignalFx organization access token or user access token to the SignalFlow client.

func MetadataTimeout

func MetadataTimeout(timeout time.Duration) ClientParam

MetadataTimeout is the default amount of time that calls to metadata accessors on a SignalFlow Computation instance will wait to receive the metadata from the backend before failing and returning a zero value. Usually metadata comes in very quickly from the stream after the job start.

func ReadTimeout

func ReadTimeout(timeout time.Duration) ClientParam

ReadTimeout sets the duration to wait between messages that come on the websocket. If the resolution of the job is very low, this should be increased.

func StreamURL

func StreamURL(streamEndpoint string) ClientParam

StreamURL lets you set the full URL to the stream endpoint, including the path.

func StreamURLForRealm

func StreamURLForRealm(realm string) ClientParam

StreamURLForRealm can be used to configure the websocket url for a specific SignalFx realm.

func UserAgent

func UserAgent(userAgent string) ClientParam

UserAgent allows setting the `userAgent` field when authenticating to SignalFlow. This can be useful for accounting how many jobs are started from each client.

func WriteTimeout

func WriteTimeout(timeout time.Duration) ClientParam

WriteTimeout sets the maximum duration to wait to send a single message when writing messages to the SignalFlow server over the WebSocket connection.

type Computation

type Computation struct {

	// The timeout to wait for metadata when a metadata access function is
	// called.  This will default to what is set on the client, but can be
	// overridden by changing this field directly.
	MetadataTimeout time.Duration
	// contains filtered or unexported fields
}

Computation is a single running SignalFlow job

func (*Computation) Channel

func (c *Computation) Channel() *Channel

Channel returns the underlying Channel instance used by this computation.

func (*Computation) Data

func (c *Computation) Data() <-chan *messages.DataMessage

Data returns the channel on which data messages come.

func (*Computation) Done

func (c *Computation) Done() <-chan struct{}

Done passes through the computation context's Done channel for use in select statements to know when the computation is finished or an error occurred.

func (*Computation) Err

func (c *Computation) Err() error

Err returns the last fatal error that caused the computation to stop, if any. Will be nil if the computation stopped in an expected manner.

func (*Computation) Handle

func (c *Computation) Handle() string

Handle of the computation

func (*Computation) IsFinished

func (c *Computation) IsFinished() bool

IsFinished returns true if the computation is done and no more data should be expected from it.

func (*Computation) Lag

func (c *Computation) Lag() time.Duration

Lag detected for the job. This will wait for a short while for the lag message to come on the websocket, but will return 0 after a timeout if it does not come.

func (*Computation) MaxDelay

func (c *Computation) MaxDelay() time.Duration

MaxDelay detected of the job. This will wait for a short while for the max delay message to come on the websocket, but will return 0 after a timeout if it does not come.

func (*Computation) Resolution

func (c *Computation) Resolution() time.Duration

Resolution of the job. This will wait for a short while for the resolution message to come on the websocket, but will return 0 after a timeout if it does not come.

func (*Computation) Stop

func (c *Computation) Stop() error

Stop the computation on the backend.

func (*Computation) StopWithReason

func (c *Computation) StopWithReason(reason string) error

StopWithReason stops the computation with a given reason. This reason will be reflected in the control message that signals the end of the job/channel.

func (*Computation) TSIDMetadata

func (c *Computation) TSIDMetadata(tsid idtool.ID) *messages.MetadataProperties

TSIDMetadata for a particular tsid. This will wait for a short while for the tsid metadata message to come on the websocket, but will return nil after a timeout if it does not come.

type DetachRequest

type DetachRequest struct {
	Type    DetachType `json:"type"`
	Channel string     `json:"channel"`
	Reason  string     `json:"reason"`
}

type DetachType

type DetachType string

func (DetachType) MarshalJSON

func (DetachType) MarshalJSON() ([]byte, error)

type ExecuteRequest

type ExecuteRequest struct {
	Type         ExecuteType   `json:"type"`
	Program      string        `json:"program"`
	Channel      string        `json:"channel"`
	Start        time.Time     `json:"-"`
	Stop         time.Time     `json:"-"`
	Resolution   time.Duration `json:"-"`
	MaxDelay     time.Duration `json:"-"`
	StartMs      int64         `json:"start"`
	StopMs       int64         `json:"stop"`
	ResolutionMs int64         `json:"resolution"`
	MaxDelayMs   int64         `json:"maxDelay"`
	Immediate    bool          `json:"immediate"`
	Timezone     string        `json:"timezone"`
}

func (ExecuteRequest) MarshalJSON

func (er ExecuteRequest) MarshalJSON() ([]byte, error)

MarshalJSON does some assignments to allow using more native Go types for time/duration.

type ExecuteType

type ExecuteType string

func (ExecuteType) MarshalJSON

func (ExecuteType) MarshalJSON() ([]byte, error)

type StopRequest

type StopRequest struct {
	Type   StopType `json:"type"`
	Handle string   `json:"handle"`
	Reason string   `json:"reason"`
}

type StopType

type StopType string

func (StopType) MarshalJSON

func (StopType) MarshalJSON() ([]byte, error)