input

package
v0.6.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 2, 2018 License: MIT Imports: 39 Imported by: 0

Documentation

Overview

Package input defines consumers for aggregating data from a variety of sources. All consumer types must implement interface input.Type.

If the source of a consumer is given assurances of message receipt then the consumer must ensure that all messages read from a source are propagated to a reader before shutting down gracefully. For example, a ZMQ subscriber model based consumer would not need to provide such ensurance, a HTTP POST based consumer, however, would be expected to propagate any requests where a 200 OK message has been returned.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrFanInNoInputs is returned when creating a FanIn type with zero inputs.
	ErrFanInNoInputs = errors.New("attempting to create fan_in input type with no inputs")
)

Functions

func Descriptions 

func Descriptions() string

Descriptions returns a formatted string of descriptions for each type.

Types

type AMQP 

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

AMQP is an input type that reads messages via the AMQP 0.91 protocol.

func (*AMQP) CloseAsync 

func (a *AMQP) CloseAsync()

CloseAsync shuts down the AMQP input and stops processing requests.

func (*AMQP) MessageChan 

func (a *AMQP) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*AMQP) StartListening 

func (a *AMQP) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*AMQP) WaitForClose 

func (a *AMQP) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the AMQP input has closed down.

type AMQPConfig 

type AMQPConfig struct {
	URL           string `json:"url" yaml:"url"`
	Exchange      string `json:"exchange" yaml:"exchange"`
	ExchangeType  string `json:"exchange_type" yaml:"exchange_type"`
	Queue         string `json:"queue" yaml:"queue"`
	BindingKey    string `json:"key" yaml:"key"`
	ConsumerTag   string `json:"consumer_tag" yaml:"consumer_tag"`
	PrefetchCount int    `json:"prefetch_count" yaml:"prefetch_count"`
	PrefetchSize  int    `json:"prefetch_size" yaml:"prefetch_size"`
}

AMQPConfig is configuration for the AMQP input type.

func NewAMQPConfig 

func NewAMQPConfig() AMQPConfig

NewAMQPConfig creates a new AMQPConfig with default values.

type Config 

type Config struct {
	Type          string              `json:"type" yaml:"type"`
	HTTPServer    HTTPServerConfig    `json:"http_server" yaml:"http_server"`
	ScaleProto    ScaleProtoConfig    `json:"scalability_protocols" yaml:"scalability_protocols"`
	ZMQ4          *ZMQ4Config         `json:"zmq4,omitempty" yaml:"zmq4,omitempty"`
	Kafka         KafkaConfig         `json:"kafka" yaml:"kafka"`
	KafkaBalanced KafkaBalancedConfig `json:"kafka_balanced" yaml:"kafka_balanced"`
	AMQP          AMQPConfig          `json:"amqp" yaml:"amqp"`
	NSQ           NSQConfig           `json:"nsq" yaml:"nsq"`
	NATS          NATSConfig          `json:"nats" yaml:"nats"`
	NATSStream    NATSStreamConfig    `json:"nats_stream" yaml:"nats_stream"`
	RedisPubSub   RedisPubSubConfig   `json:"redis_pubsub" yaml:"redis_pubsub"`
	File          FileConfig          `json:"file" yaml:"file"`
	STDIN         STDINConfig         `json:"stdin" yaml:"stdin"`
	FanIn         FanInConfig         `json:"fan_in" yaml:"fan_in"`
	Processors    []processor.Config  `json:"processors" yaml:"processors"`
}

Config is the all encompassing configuration struct for all input types. Note that some configs are empty structs, as the type has no optional values but we want to list it as an option.

func NewConfig 

func NewConfig() Config

NewConfig returns a configuration struct fully populated with default values.

type FanInConfig 

type FanInConfig struct {
	Inputs []interface{} `json:"inputs" yaml:"inputs"`
}

FanInConfig is configuration for the FanIn input type.

func NewFanInConfig 

func NewFanInConfig() FanInConfig

NewFanInConfig creates a new FanInConfig with default values.

type FileConfig 

type FileConfig struct {
	Path        string `json:"path" yaml:"path"`
	Multipart   bool   `json:"multipart" yaml:"multipart"`
	MaxBuffer   int    `json:"max_buffer" yaml:"max_buffer"`
	CustomDelim string `json:"custom_delimiter" yaml:"custom_delimiter"`
}

FileConfig is configuration values for the File input type.

func NewFileConfig 

func NewFileConfig() FileConfig

NewFileConfig creates a new FileConfig with default values.

type HTTPServer 

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

HTTPServer is an input type that serves HTTPServer POST requests.

func (*HTTPServer) CloseAsync 

func (h *HTTPServer) CloseAsync()

CloseAsync shuts down the HTTPServer input and stops processing requests.

func (*HTTPServer) MessageChan 

func (h *HTTPServer) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*HTTPServer) StartListening 

func (h *HTTPServer) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*HTTPServer) WaitForClose 

func (h *HTTPServer) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the HTTPServer input has closed down.

type HTTPServerConfig 

type HTTPServerConfig struct {
	Address   string `json:"address" yaml:"address"`
	Path      string `json:"path" yaml:"path"`
	TimeoutMS int64  `json:"timeout_ms" yaml:"timeout_ms"`
	CertFile  string `json:"cert_file" yaml:"cert_file"`
	KeyFile   string `json:"key_file" yaml:"key_file"`
}

HTTPServerConfig is configuration for the HTTPServer input type.

func NewHTTPServerConfig 

func NewHTTPServerConfig() HTTPServerConfig

NewHTTPServerConfig creates a new HTTPServerConfig with default values.

type Kafka 

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

Kafka is an input type that reads from a Kafka instance.

func (*Kafka) CloseAsync 

func (k *Kafka) CloseAsync()

CloseAsync shuts down the Kafka input and stops processing requests.

func (*Kafka) MessageChan 

func (k *Kafka) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*Kafka) StartListening 

func (k *Kafka) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*Kafka) WaitForClose 

func (k *Kafka) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the Kafka input has closed down.

type KafkaBalanced 

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

KafkaBalanced is an input type that reads from a KafkaBalanced instance.

func (*KafkaBalanced) CloseAsync 

func (k *KafkaBalanced) CloseAsync()

CloseAsync shuts down the KafkaBalanced input and stops processing requests.

func (*KafkaBalanced) MessageChan 

func (k *KafkaBalanced) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*KafkaBalanced) StartListening 

func (k *KafkaBalanced) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*KafkaBalanced) WaitForClose 

func (k *KafkaBalanced) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the KafkaBalanced input has closed down.

type KafkaBalancedConfig 

type KafkaBalancedConfig struct {
	Addresses       []string `json:"addresses" yaml:"addresses"`
	ClientID        string   `json:"client_id" yaml:"client_id"`
	ConsumerGroup   string   `json:"consumer_group" yaml:"consumer_group"`
	Topics          []string `json:"topics" yaml:"topics"`
	StartFromOldest bool     `json:"start_from_oldest" yaml:"start_from_oldest"`
}

KafkaBalancedConfig is configuration for the KafkaBalanced input type.

func NewKafkaBalancedConfig 

func NewKafkaBalancedConfig() KafkaBalancedConfig

NewKafkaBalancedConfig creates a new KafkaBalancedConfig with default values.

type KafkaConfig 

type KafkaConfig struct {
	Addresses       []string `json:"addresses" yaml:"addresses"`
	ClientID        string   `json:"client_id" yaml:"client_id"`
	ConsumerGroup   string   `json:"consumer_group" yaml:"consumer_group"`
	Topic           string   `json:"topic" yaml:"topic"`
	Partition       int32    `json:"partition" yam:"partition"`
	StartFromOldest bool     `json:"start_from_oldest" yaml:"start_from_oldest"`
}

KafkaConfig is configuration for the Kafka input type.

func NewKafkaConfig 

func NewKafkaConfig() KafkaConfig

NewKafkaConfig creates a new KafkaConfig with default values.

type NATS 

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

NATS is an input type that receives NATS messages.

func (*NATS) CloseAsync 

func (n *NATS) CloseAsync()

CloseAsync shuts down the NATS input and stops processing requests.

func (*NATS) MessageChan 

func (n *NATS) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*NATS) StartListening 

func (n *NATS) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*NATS) WaitForClose 

func (n *NATS) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NATS input has closed down.

type NATSConfig 

type NATSConfig struct {
	URLs    []string `json:"urls" yaml:"urls"`
	Subject string   `json:"subject" yaml:"subject"`
}

NATSConfig is configuration for the NATS input type.

func NewNATSConfig 

func NewNATSConfig() NATSConfig

NewNATSConfig creates a new NATSConfig with default values.

type NATSStream 

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

NATSStream is an input type that receives NATSStream messages.

func (*NATSStream) CloseAsync 

func (n *NATSStream) CloseAsync()

CloseAsync shuts down the NATSStream input and stops processing requests.

func (*NATSStream) MessageChan 

func (n *NATSStream) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*NATSStream) StartListening 

func (n *NATSStream) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*NATSStream) WaitForClose 

func (n *NATSStream) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NATSStream input has closed down.

type NATSStreamConfig 

type NATSStreamConfig struct {
	URLs        []string `json:"urls" yaml:"urls"`
	ClusterID   string   `json:"cluster_id" yaml:"cluster_id"`
	ClientID    string   `json:"client_id" yaml:"client_id"`
	QueueID     string   `json:"queue" yaml:"queue"`
	DurableName string   `json:"durable_name" yaml:"durable_name"`
	Subject     string   `json:"subject" yaml:"subject"`
}

NATSStreamConfig is configuration for the NATSStream input type.

func NewNATSStreamConfig 

func NewNATSStreamConfig() NATSStreamConfig

NewNATSStreamConfig creates a new NATSStreamConfig with default values.

type NSQ 

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

NSQ is an input type that receives NSQ messages.

func (*NSQ) CloseAsync 

func (n *NSQ) CloseAsync()

CloseAsync shuts down the NSQ input and stops processing requests.

func (*NSQ) HandleMessage 

func (n *NSQ) HandleMessage(message *nsq.Message) error

HandleMessage handles an NSQ message.

func (*NSQ) MessageChan 

func (n *NSQ) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*NSQ) StartListening 

func (n *NSQ) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*NSQ) WaitForClose 

func (n *NSQ) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the NSQ input has closed down.

type NSQConfig 

type NSQConfig struct {
	Addresses       []string `json:"nsqd_tcp_addresses" yaml:"nsqd_tcp_addresses"`
	LookupAddresses []string `json:"lookupd_http_addresses" yaml:"lookupd_http_addresses"`
	Topic           string   `json:"topic" yaml:"topic"`
	Channel         string   `json:"channel" yaml:"channel"`
	UserAgent       string   `json:"user_agent" yaml:"user_agent"`
	MaxInFlight     int      `json:"max_in_flight" yaml:"max_in_flight"`
}

NSQConfig is configuration for the NSQ input type.

func NewNSQConfig 

func NewNSQConfig() NSQConfig

NewNSQConfig creates a new NSQConfig with default values.

type RedisPubSub 

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

RedisPubSub is an input type that reads Redis Pub/Sub messages.

func (*RedisPubSub) CloseAsync 

func (r *RedisPubSub) CloseAsync()

CloseAsync shuts down the RedisPubSub input and stops processing requests.

func (*RedisPubSub) MessageChan 

func (r *RedisPubSub) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*RedisPubSub) StartListening 

func (r *RedisPubSub) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*RedisPubSub) WaitForClose 

func (r *RedisPubSub) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the RedisPubSub input has closed down.

type RedisPubSubConfig 

type RedisPubSubConfig struct {
	URL      string   `json:"url" yaml:"url"`
	Channels []string `json:"channels" yaml:"channels"`
}

RedisPubSubConfig is configuration for the RedisPubSub input type.

func NewRedisPubSubConfig 

func NewRedisPubSubConfig() RedisPubSubConfig

NewRedisPubSubConfig creates a new RedisPubSubConfig with default values.

type STDINConfig 

type STDINConfig struct {
	Multipart   bool   `json:"multipart" yaml:"multipart"`
	MaxBuffer   int    `json:"max_buffer" yaml:"max_buffer"`
	CustomDelim string `json:"custom_delimiter" yaml:"custom_delimiter"`
}

STDINConfig contains config fields for the STDIN input type.

func NewSTDINConfig 

func NewSTDINConfig() STDINConfig

NewSTDINConfig creates a STDINConfig populated with default values.

type ScaleProto 

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

ScaleProto is an input type that serves Scalability Protocols messages.

func (*ScaleProto) CloseAsync 

func (s *ScaleProto) CloseAsync()

CloseAsync shuts down the ScaleProto input and stops processing requests.

func (*ScaleProto) MessageChan 

func (s *ScaleProto) MessageChan() <-chan types.Message

MessageChan returns the messages channel.

func (*ScaleProto) StartListening 

func (s *ScaleProto) StartListening(responses <-chan types.Response) error

StartListening sets the channel used by the input to validate message receipt.

func (*ScaleProto) WaitForClose 

func (s *ScaleProto) WaitForClose(timeout time.Duration) error

WaitForClose blocks until the ScaleProto input has closed down.

type ScaleProtoConfig 

type ScaleProtoConfig struct {
	URLs          []string `json:"urls" yaml:"urls"`
	Bind          bool     `json:"bind" yaml:"bind"`
	SocketType    string   `json:"socket_type" yaml:"socket_type"`
	SuccessStr    string   `json:"reply_success" yaml:"reply_success"`
	ErrorStr      string   `json:"reply_error" yaml:"reply_error"`
	SubFilters    []string `json:"sub_filters" yaml:"sub_filters"`
	PollTimeoutMS int      `json:"poll_timeout_ms" yaml:"poll_timeout_ms"`
	RepTimeoutMS  int      `json:"reply_timeout_ms" yaml:"reply_timeout_ms"`
}

ScaleProtoConfig is configuration for the ScaleProto input type.

func NewScaleProtoConfig 

func NewScaleProtoConfig() ScaleProtoConfig

NewScaleProtoConfig creates a new ScaleProtoConfig with default values.

type Type 

type Type interface {
	types.Closable
	types.Producer
}

Type is the standard interface of an input type.

func New 

func New(
	conf Config,
	log log.Modular,
	stats metrics.Type,
	pipelines ...pipeline.ConstructorFunc,
) (Type, error)

New creates an input type based on an input configuration.

func NewAMQP 

func NewAMQP(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewAMQP creates a new AMQP input type.

func NewFanIn 

func NewFanIn(
	conf Config,
	log log.Modular,
	stats metrics.Type,
	pipelines ...pipeline.ConstructorFunc,
) (Type, error)

NewFanIn creates a new FanIn input type.

func NewFile 

func NewFile(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewFile creates a new File input type.

func NewHTTPServer 

func NewHTTPServer(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewHTTPServer creates a new HTTPServer input type.

func NewKafka 

func NewKafka(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewKafka creates a new Kafka input type.

func NewKafkaBalanced 

func NewKafkaBalanced(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewKafkaBalanced creates a new KafkaBalanced input type.

func NewNATS 

func NewNATS(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewNATS creates a new NATS input type.

func NewNATSStream 

func NewNATSStream(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewNATSStream creates a new NATSStream input type.

func NewNSQ 

func NewNSQ(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewNSQ create a new NSQ input type.

func NewRedisPubSub 

func NewRedisPubSub(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewRedisPubSub creates a new RedisPubSub input type.

func NewSTDIN 

func NewSTDIN(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewSTDIN creates a new STDIN input type.

func NewScaleProto 

func NewScaleProto(conf Config, log log.Modular, stats metrics.Type) (Type, error)

NewScaleProto creates a new ScaleProto input type.

func WrapWithPipelines 

func WrapWithPipelines(in Type, pipeConstructors ...pipeline.ConstructorFunc) (Type, error)

WrapWithPipelines wraps an input with a variadic number of pipelines.

type WithPipeline 

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

WithPipeline is a type that wraps both an input type and a pipeline type by routing the input through the pipeline, and implements the input.Type interface in order to act like an ordinary input.

func WrapWithPipeline 

func WrapWithPipeline(in Type, pipeConstructor pipeline.ConstructorFunc) (*WithPipeline, error)

WrapWithPipeline routes an input directly into a processing pipeline and returns a type that manages both and acts like an ordinary input.

func (*WithPipeline) CloseAsync 

func (i *WithPipeline) CloseAsync()

CloseAsync triggers a closure of this object but does not block.

func (*WithPipeline) MessageChan 

func (i *WithPipeline) MessageChan() <-chan types.Message

MessageChan returns the channel used for consuming messages from this input.

func (*WithPipeline) StartListening 

func (i *WithPipeline) StartListening(resChan <-chan types.Response) error

StartListening starts the type listening to a response channel from a consumer.

func (*WithPipeline) WaitForClose 

func (i *WithPipeline) WaitForClose(timeout time.Duration) error

WaitForClose is a blocking call to wait until the object has finished closing down and cleaning up resources.

type ZMQ4Config 

type ZMQ4Config struct{}

ZMQ4Config is an empty stub for when ZMQ4 is not compiled.

func NewZMQ4Config 

func NewZMQ4Config() *ZMQ4Config

NewZMQ4Config returns nil.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.