Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

input

package

Go to main page

Versions in this module

v4

v4.0.0

Apr 20, 2022

Changes in this version

+ const TypeAMQP09

+ const TypeAWSKinesis

+ const TypeAWSS3

+ const TypeAWSSQS

+ const TypeAzureBlobStorage

+ const TypeAzureQueueStorage

+ const TypeBroker

+ const TypeCSVFile

+ const TypeDynamic

+ const TypeFile

+ const TypeGCPCloudStorage

+ const TypeGCPPubSub

+ const TypeGenerate

+ const TypeHDFS

+ const TypeHTTPClient

+ const TypeHTTPServer

+ const TypeInproc

+ const TypeKafka

+ const TypeKinesis

+ const TypeMQTT

+ const TypeNATS

+ const TypeNATSJetStream

+ const TypeNATSStream

+ const TypeNSQ

+ const TypeNanomsg

+ const TypeReadUntil

+ const TypeRedisList

+ const TypeRedisPubSub

+ const TypeRedisStreams

+ const TypeResource

+ const TypeSFTP

+ const TypeSTDIN

+ const TypeSequence

+ const TypeSocket

+ const TypeSocketServer

+ const TypeSubprocess

+ const TypeWebsocket

+ var Constructors = map[string]TypeSpec

+ func AppendProcessorsFromConfig(conf Config, mgr interop.Manager, ...) []iprocessor.PipelineConstructorFunc

+ func New(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type, ...) (input.Streamed, error)

+ func NewAsyncReader(typeStr string, allowSkipAcks bool, r reader.Async, log log.Modular, ...) (input.Streamed, error)

+ func NewBatcher(batcher *policy.Batcher, child input.Streamed, log log.Modular, ...) input.Streamed

+ func NewCSVFile(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewFile(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewGCPPubSub(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewHDFS(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewHTTPClient(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewHTTPServer(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewInproc(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewKafka(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewNATS(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewNATSStream(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewNSQ(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewNanomsg(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewReadUntil(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewRedisList(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewRedisPubSub(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewRedisStreams(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewResource(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewSTDIN(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewSequence(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewSocket(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewSocketServer(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func NewWebsocket(conf Config, mgr interop.Manager, log log.Modular, stats metrics.Type) (input.Streamed, error)

+ func WalkConstructors(fn func(ConstructorFunc, docs.ComponentSpec))

+ func WrapWithPipelines(in input.Streamed, pipeConstructors ...iprocessor.PipelineConstructorFunc) (input.Streamed, error)

+ type AMQP09BindingConfig struct

+ Exchange string

+ RoutingKey string

+ type AMQP09Config struct

+ AutoAck bool

+ BindingsDeclare []AMQP09BindingConfig

+ ConsumerTag string

+ NackRejectPatterns []string

+ PrefetchCount int

+ PrefetchSize int

+ Queue string

+ QueueDeclare AMQP09QueueDeclareConfig

+ TLS btls.Config

+ URLs []string

+ func NewAMQP09Config() AMQP09Config

+ type AMQP09QueueDeclareConfig struct

+ Durable bool

+ Enabled bool

+ type AMQP1Config struct

+ AzureRenewLock bool

+ SASL shared.SASLConfig

+ SourceAddress string

+ TLS btls.Config

+ URL string

+ func NewAMQP1Config() AMQP1Config

+ type AWSKinesisConfig struct

+ Batching policy.Config

+ CheckpointLimit int

+ CommitPeriod string

+ DynamoDB DynamoDBCheckpointConfig

+ LeasePeriod string

+ RebalancePeriod string

+ StartFromOldest bool

+ Streams []string

+ func NewAWSKinesisConfig() AWSKinesisConfig

+ type AWSS3Config struct

+ Bucket string

+ Codec string

+ DeleteObjects bool

+ ForcePathStyleURLs bool

+ Prefix string

+ SQS AWSS3SQSConfig

+ func NewAWSS3Config() AWSS3Config

+ type AWSS3SQSConfig struct

+ BucketPath string

+ DelayPeriod string

+ Endpoint string

+ EnvelopePath string

+ KeyPath string

+ MaxMessages int64

+ URL string

+ func NewAWSS3SQSConfig() AWSS3SQSConfig

+ type AWSSQSConfig struct

+ DeleteMessage bool

+ MaxNumberOfMessages int

+ ResetVisibility bool

+ URL string

+ func NewAWSSQSConfig() AWSSQSConfig

+ type AsyncReader struct

+ func (r *AsyncReader) CloseAsync()

+ func (r *AsyncReader) Connected() bool

+ func (r *AsyncReader) TransactionChan() <-chan message.Transaction

+ func (r *AsyncReader) WaitForClose(timeout time.Duration) error

+ type AzureBlobStorageConfig struct

+ Codec string

+ Container string

+ DeleteObjects bool

+ Prefix string

+ StorageAccessKey string

+ StorageAccount string

+ StorageConnectionString string

+ StorageSASToken string

+ func NewAzureBlobStorageConfig() AzureBlobStorageConfig

+ type AzureQueueStorageConfig struct

+ DequeueVisibilityTimeout string

+ MaxInFlight int32

+ QueueName string

+ StorageAccessKey string

+ StorageAccount string

+ StorageConnectionString string

+ StorageSASToken string

+ TrackProperties bool

+ func NewAzureQueueStorageConfig() AzureQueueStorageConfig

+ type Batcher struct

+ func (m *Batcher) CloseAsync()

+ func (m *Batcher) Connected() bool

+ func (m *Batcher) TransactionChan() <-chan message.Transaction

+ func (m *Batcher) WaitForClose(timeout time.Duration) error

+ type Bloblang struct

+ func (b *Bloblang) CloseAsync()

+ func (b *Bloblang) ConnectWithContext(ctx context.Context) error

+ func (b *Bloblang) ReadWithContext(ctx context.Context) (*message.Batch, reader.AsyncAckFn, error)

+ func (b *Bloblang) WaitForClose(timeout time.Duration) error

+ type BloblangConfig struct

+ Count int

+ Interval string

+ Mapping string

+ func NewBloblangConfig() BloblangConfig

+ type BrokerConfig struct

+ Batching policy.Config

+ Copies int

+ Inputs []Config

+ func NewBrokerConfig() BrokerConfig

+ type CSVFileConfig struct

+ BatchCount int

+ Delim string

+ ParseHeaderRow bool

+ Paths []string

+ func NewCSVFileConfig() CSVFileConfig

+ type Config struct

+ AMQP09 AMQP09Config

+ AMQP1 AMQP1Config

+ AWSKinesis AWSKinesisConfig

+ AWSS3 AWSS3Config

+ AWSSQS AWSSQSConfig

+ AzureBlobStorage AzureBlobStorageConfig

+ AzureQueueStorage AzureQueueStorageConfig

+ Broker BrokerConfig

+ CSVFile CSVFileConfig

+ Dynamic DynamicConfig

+ File FileConfig

+ GCPCloudStorage GCPCloudStorageConfig

+ GCPPubSub reader.GCPPubSubConfig

+ Generate BloblangConfig

+ HDFS reader.HDFSConfig

+ HTTPClient HTTPClientConfig

+ HTTPServer HTTPServerConfig

+ Inproc InprocConfig

+ Kafka KafkaConfig

+ Label string

+ MQTT MQTTConfig

+ NATS reader.NATSConfig

+ NATSStream reader.NATSStreamConfig

+ NSQ reader.NSQConfig

+ Nanomsg reader.ScaleProtoConfig

+ Plugin interface{}

+ Processors []processor.Config

+ ReadUntil ReadUntilConfig

+ RedisList reader.RedisListConfig

+ RedisPubSub reader.RedisPubSubConfig

+ RedisStreams reader.RedisStreamsConfig

+ Resource string

+ SFTP SFTPConfig

+ STDIN STDINConfig

+ Sequence SequenceConfig

+ Socket SocketConfig

+ SocketServer SocketServerConfig

+ Subprocess SubprocessConfig

+ Type string

+ Websocket reader.WebsocketConfig

+ func NewConfig() Config

+ func (conf *Config) UnmarshalYAML(value *yaml.Node) error

+ type ConstructorFunc func(Config, interop.Manager, log.Modular, metrics.Type, ...) (input.Streamed, error)

+ type DynamicConfig struct

+ Inputs map[string]Config

+ Prefix string

+ func NewDynamicConfig() DynamicConfig

+ type DynamoDBCheckpointConfig struct

+ BillingMode string

+ Create bool

+ ReadCapacityUnits int64

+ Table string

+ WriteCapacityUnits int64

+ func NewDynamoDBCheckpointConfig() DynamoDBCheckpointConfig

+ type FileConfig struct

+ Codec string

+ DeleteOnFinish bool

+ MaxBuffer int

+ Paths []string

+ func NewFileConfig() FileConfig

+ type GCPCloudStorageConfig struct

+ Bucket string

+ Codec string

+ DeleteObjects bool

+ Prefix string

+ func NewGCPCloudStorageConfig() GCPCloudStorageConfig

+ type HTTPClient struct

+ func (h *HTTPClient) CloseAsync()

+ func (h *HTTPClient) ConnectWithContext(ctx context.Context) (err error)

+ func (h *HTTPClient) ReadWithContext(ctx context.Context) (*message.Batch, reader.AsyncAckFn, error)

+ func (h *HTTPClient) WaitForClose(timeout time.Duration) error

+ type HTTPClientConfig struct

+ DropEmptyBodies bool

+ Payload string

+ Stream StreamConfig

+ func NewHTTPClientConfig() HTTPClientConfig

+ type HTTPServer struct

+ func (h *HTTPServer) CloseAsync()

+ func (h *HTTPServer) Connected() bool

+ func (h *HTTPServer) TransactionChan() <-chan message.Transaction

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

+ type HTTPServerConfig struct

+ Address string

+ AllowedVerbs []string

+ CORS httpdocs.ServerCORS

+ CertFile string

+ KeyFile string

+ Path string

+ RateLimit string

+ Response HTTPServerResponseConfig

+ Timeout string

+ WSPath string

+ WSRateLimitMessage string

+ WSWelcomeMessage string

+ func NewHTTPServerConfig() HTTPServerConfig

+ type HTTPServerResponseConfig struct

+ ExtractMetadata imetadata.IncludeFilterConfig

+ Headers map[string]string

+ Status string

+ func NewHTTPServerResponseConfig() HTTPServerResponseConfig

+ type Inproc struct

+ func (i *Inproc) CloseAsync()

+ func (i *Inproc) Connected() bool

+ func (i *Inproc) TransactionChan() <-chan message.Transaction

+ func (i *Inproc) WaitForClose(timeout time.Duration) error

+ type InprocConfig string

+ func NewInprocConfig() InprocConfig

+ type KafkaBalancedGroupConfig struct

+ HeartbeatInterval string

+ RebalanceTimeout string

+ SessionTimeout string

+ func NewKafkaBalancedGroupConfig() KafkaBalancedGroupConfig

+ type KafkaConfig struct

+ Addresses []string

+ Batching policy.Config

+ CheckpointLimit int

+ ClientID string

+ CommitPeriod string

+ ConsumerGroup string

+ ExtractTracingMap string

+ FetchBufferCap int

+ Group KafkaBalancedGroupConfig

+ MaxProcessingPeriod string

+ RackID string

+ SASL sasl.Config

+ StartFromOldest bool

+ TLS btls.Config

+ TargetVersion string

+ Topics []string

+ func NewKafkaConfig() KafkaConfig

+ type MQTTConfig struct

+ CleanSession bool

+ ClientID string

+ ConnectTimeout string

+ DynamicClientIDSuffix string

+ KeepAlive int64

+ Password string

+ QoS uint8

+ TLS tls.Config

+ Topics []string

+ URLs []string

+ User string

+ Will mqttconf.Will

+ func NewMQTTConfig() MQTTConfig

+ type ReadUntil struct

+ func (r *ReadUntil) CloseAsync()

+ func (r *ReadUntil) Connected() bool

+ func (r *ReadUntil) TransactionChan() <-chan message.Transaction

+ func (r *ReadUntil) WaitForClose(timeout time.Duration) error

+ type ReadUntilConfig struct

+ Check string

+ Input *Config

+ Restart bool

+ func NewReadUntilConfig() ReadUntilConfig

+ func (r ReadUntilConfig) MarshalJSON() ([]byte, error)

+ func (r ReadUntilConfig) MarshalYAML() (interface{}, error)

+ type Resource struct

+ func (r *Resource) CloseAsync()

+ func (r *Resource) Connected() (isConnected bool)

+ func (r *Resource) TransactionChan() (tChan <-chan message.Transaction)

+ func (r *Resource) WaitForClose(timeout time.Duration) error

+ type SFTPConfig struct

+ Address string

+ Codec string

+ Credentials sftpSetup.Credentials

+ DeleteOnFinish bool

+ MaxBuffer int

+ Paths []string

+ Watcher watcherConfig

+ func NewSFTPConfig() SFTPConfig

+ type STDINConfig struct

+ Codec string

+ MaxBuffer int

+ func NewSTDINConfig() STDINConfig

+ type Sequence struct

+ func (r *Sequence) CloseAsync()

+ func (r *Sequence) Connected() bool

+ func (r *Sequence) TransactionChan() <-chan message.Transaction

+ func (r *Sequence) WaitForClose(timeout time.Duration) error

+ type SequenceConfig struct

+ Inputs []Config

+ ShardedJoin SequenceShardedJoinConfig

+ func NewSequenceConfig() SequenceConfig

+ type SequenceShardedJoinConfig struct

+ IDPath string

+ Iterations int

+ MergeStrategy string

+ Type string

+ func NewSequenceShardedJoinConfig() SequenceShardedJoinConfig

+ type SocketConfig struct

+ Address string

+ Codec string

+ MaxBuffer int

+ Network string

+ func NewSocketConfig() SocketConfig

+ type SocketServer struct

+ func (t *SocketServer) Addr() net.Addr

+ func (t *SocketServer) CloseAsync()

+ func (t *SocketServer) Connected() bool

+ func (t *SocketServer) TransactionChan() <-chan message.Transaction

+ func (t *SocketServer) WaitForClose(timeout time.Duration) error

+ type SocketServerConfig struct

+ Address string

+ Codec string

+ MaxBuffer int

+ Network string

+ func NewSocketServerConfig() SocketServerConfig

+ type StreamConfig struct

+ Codec string

+ Enabled bool

+ MaxBuffer int

+ Reconnect bool

+ type Subprocess struct

+ func (s *Subprocess) CloseAsync()

+ func (s *Subprocess) ConnectWithContext(ctx context.Context) error

+ func (s *Subprocess) ReadWithContext(ctx context.Context) (*message.Batch, reader.AsyncAckFn, error)

+ func (s *Subprocess) WaitForClose(timeout time.Duration) error

+ type SubprocessConfig struct

+ Args []string

+ Codec string

+ MaxBuffer int

+ Name string

+ RestartOnExit bool

+ func NewSubprocessConfig() SubprocessConfig

+ type TypeSpec struct

+ Categories []string

+ Config docs.FieldSpec

+ Description string

+ Examples []docs.AnnotatedExample

+ Footnotes string

+ Status docs.Status

+ Summary string

+ Version string

+ type WithPipeline struct

+ func WrapWithPipeline(in input.Streamed, pipeConstructor iprocessor.PipelineConstructorFunc) (*WithPipeline, error)

+ func (i *WithPipeline) CloseAsync()

+ func (i *WithPipeline) Connected() bool

+ func (i *WithPipeline) TransactionChan() <-chan message.Transaction

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

v4.0.0-rc3

Mar 31, 2022

v4.0.0-rc2

Mar 31, 2022

v4.0.0-rc1

Mar 20, 2022