app

Documentation

Index

• Constants
• Variables
• func NewAWSSession(creds Credentials) *session.Session
• type App
  • func (app *App) DecodeSQSMessages(messages []*sqs.Message) ([]SQSMessage, error)
  • func (app *App) DeleteSQSMessage(messageID, QueueUrl string, receiptHandle *string) error
  • func (app *App) Init(creds Credentials)
  • func (app *App) PollSQS(params *Params) error
  • func (app *App) ProcessSQSEvent(params *Params, messages []*sqs.Message) error
  • func (app *App) ReadSQSMessage(QueueUrl string, batchSize, timeout int64) (*sqs.ReceiveMessageOutput, error)
  • func (app *App) ValidateSQSMessages(messages []SQSMessage) error
  • func (app *App) WriteMessagesToKinesisStream(stream string, sqsMessage []SQSMessage) error
• type Credentials
  • func (c Credentials) String() string
• type Event
• type KinesisClient
  • func NewKinesisClient(session *session.Session) (*KinesisClient, error)
  • func (kc *KinesisClient) BatchedDelete() error
  • func (kc *KinesisClient) BatchedWrite(records *kinesis.PutRecordsInput) (*kinesis.PutRecordsOutput, error)
  • func (kc *KinesisClient) Delete() error
  • func (kc *KinesisClient) Read() error
  • func (kc *KinesisClient) Write(record *kinesis.PutRecordInput) (*kinesis.PutRecordOutput, error)
• type KinesisRecord
• type Message
• type NetworkConnection
• type Params
• type Process
• type SQSClient
  • func NewSQSClient(session *session.Session) (*SQSClient, error)
  • func (sc *SQSClient) BatchedDelete(message *sqs.DeleteMessageBatchInput) (*sqs.DeleteMessageBatchOutput, error)
  • func (sc *SQSClient) BatchedWrite() error
  • func (sc *SQSClient) Delete(message *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error)
  • func (sc *SQSClient) Read(message *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error)
  • func (sc *SQSClient) Write(message *sqs.SendMessageInput) (*sqs.SendMessageOutput, error)
• type SQSMessage

Constants

const (
	UTC = "2006-01-02T15:04:05.999999Z"
)

Variables

var (
	ErrMissingEvents                      = errors.New("SQS event is missing events")
	ErrMissingProcessAndNetworkConnection = errors.New("SQS event is missing new_process and network_connection")
	ErrMissingDeviceID                    = errors.New("SQS event is missing device_id")
)

Functions

func NewAWSSession

func NewAWSSession(creds Credentials) *session.Session

Types

type App

type App struct {
	SQSClient     *SQSClient
	KinesisClient *KinesisClient
	Params        *Params
}

func (*App) DecodeSQSMessages

func (app *App) DecodeSQSMessages(messages []*sqs.Message) ([]SQSMessage, error)

func (*App) DeleteSQSMessage

func (app *App) DeleteSQSMessage(messageID, QueueUrl string, receiptHandle *string) error

func (*App) Init

func (app *App) Init(creds Credentials)

func (*App) PollSQS

func (app *App) PollSQS(params *Params) error

func (*App) ProcessSQSEvent

func (app *App) ProcessSQSEvent(params *Params, messages []*sqs.Message) error

func (*App) ReadSQSMessage

func (app *App) ReadSQSMessage(QueueUrl string, batchSize, timeout int64) (*sqs.ReceiveMessageOutput, error)

func (*App) ValidateSQSMessages

func (app *App) ValidateSQSMessages(messages []SQSMessage) error

func (*App) WriteMessagesToKinesisStream

func (app *App) WriteMessagesToKinesisStream(stream string, sqsMessage []SQSMessage) error

type Credentials

type Credentials struct {
	AccessKey       string  `json:"access_key"`
	SecretAccessKey string  `json:"secret_access_key"`
	SessionToken    string  `json:"session_token"`
	Region          string  `json:"region"`
	EndPoint        *string `json:"end_point"`
}

func (Credentials) String

func (c Credentials) String() string

type Event

type Event struct {
	Processes          []Process           `json:"new_process"`
	NetworkConnections []NetworkConnection `json:"network_connection"`
}

type KinesisClient

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

func NewKinesisClient

func NewKinesisClient(session *session.Session) (*KinesisClient, error)

func (*KinesisClient) BatchedDelete

func (kc *KinesisClient) BatchedDelete() error

func (*KinesisClient) BatchedWrite

func (kc *KinesisClient) BatchedWrite(records *kinesis.PutRecordsInput) (*kinesis.PutRecordsOutput, error)

func (*KinesisClient) Delete

func (kc *KinesisClient) Delete() error

func (*KinesisClient) Read

func (kc *KinesisClient) Read() error

func (*KinesisClient) Write

func (kc *KinesisClient) Write(record *kinesis.PutRecordInput) (*kinesis.PutRecordOutput, error)

type KinesisRecord

type KinesisRecord struct {
	RecordID string       `json:"id"`
	Data     []SQSMessage `json:"data"`
	Created  string       `json:"created"`
}

type Message

type Message interface {
	Read()
	Write()
	BatchedWrite()
	Delete()
	BatchedDelete()
}

type NetworkConnection

type NetworkConnection struct {
	SourceIP        string `json:"source_ip"`
	DestinationIP   string `json:"destination_ip"`
	DestinationPort int    `json:"destination_port"`
}

type Params

type Params struct {
	BatchSize    int
	PollInterval int
	Queue        string
	Stream       string
	Timeout      int
}

type Process

type Process struct {
	Cmdl string `json:"cmdl"`
	User string `json:"user"`
}

type SQSClient

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

func NewSQSClient

func NewSQSClient(session *session.Session) (*SQSClient, error)

func (*SQSClient) BatchedDelete

func (sc *SQSClient) BatchedDelete(message *sqs.DeleteMessageBatchInput) (*sqs.DeleteMessageBatchOutput, error)

func (*SQSClient) BatchedWrite

func (sc *SQSClient) BatchedWrite() error

func (*SQSClient) Delete

func (sc *SQSClient) Delete(message *sqs.DeleteMessageInput) (*sqs.DeleteMessageOutput, error)

func (*SQSClient) Read

func (sc *SQSClient) Read(message *sqs.ReceiveMessageInput) (*sqs.ReceiveMessageOutput, error)

func (*SQSClient) Write

func (sc *SQSClient) Write(message *sqs.SendMessageInput) (*sqs.SendMessageOutput, error)

type SQSMessage

type SQSMessage struct {
	SubmissionID string `json:"submission_id"`
	DeviceID     string `json:"device_id"`
	TimeCreated  string `json:"time_created"`
	Events       Event  `json:"events"`
}