cli

Documentation

Index

• Constants
• func HandleAWSSNSFlags(_, writeCmd, _ *kingpin.CmdClause, opts *Options)
• func HandleAWSSQSFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)
• func HandleActiveMqFlags(readCmd, writeCmd *kingpin.CmdClause, opts *Options)
• func HandleAzureFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)
• func HandleGCPPubSubFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)
• func HandleGlobalFlags(cmd *kingpin.CmdClause, opts *Options)
• func HandleGlobalReadFlags(cmd *kingpin.CmdClause, opts *Options)
• func HandleGlobalWriteFlags(cmd *kingpin.CmdClause, opts *Options)
• func HandleKafkaFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)
• func HandleMQTTFlags(readCmd, writeCmd *kingpin.CmdClause, opts *Options)
• func HandleNatsFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)
• func HandleRabbitFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)
• func HandleRedisFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)
• func HandleRelayFlags(relayCmd *kingpin.CmdClause, opts *Options)
• func ValidateProtobufOptions(dirs []string, rootMessage string) error
• type AWSSNSOptions
• type AWSSQSOptions
• type ActiveMqOptions
• type AzureServiceBusOptions
• type GCPPubSubOptions
• type KafkaOptions
• type MQTTOptions
• type NatsOptions
• type Options
  • func Handle(cliArgs []string) (string, *Options, error)
• type RabbitOptions
• type RedisOptions

Constants

const (
	DefaultGRPCAddress         = "grpc-collector.batch.sh:9000"
	DefaultHTTPListenAddress   = ":8080"
	DefaultGRPCTimeout         = "10s"
	DefaultNumWorkers          = "10"
	DefaultStatsReportInterval = "5s"
	DefaultBatchSize           = "10"
)

const (
	DefaultKafkaConnectTimeout = "10s"
	DefaultKafkaGroupId        = "plumber"

	// Optimize for immediate output
	DefaultKafkaMaxWait          = "1s"
	DefaultKafkaMinBytes         = "1"
	DefaultKafkaMaxBytes         = "1"
	DefaultKafkaQueueCapacity    = "1"
	DefaultKafkaRebalanceTimeout = "0"

	// Optimize for throughput + reliability
	DefaultKafkaRelayMaxWait          = "5s"
	DefaultKafkaRelayMinBytes         = "1048576" // 1MB
	DefaultKafkaRelayMaxBytes         = "1048576" // 1MB
	DefaultKafkaRelayQueueCapacity    = "1000"
	DefaultKafkaRelayRebalanceTimeout = "5s"
	DefaultKafkaRelayCommitInterval   = "5s"
)

const (
	MQTTDefaultConnectTimeout = "5s"
	MQTTDefaultClientId       = "plumber"
)

Functions

func HandleAWSSNSFlags

func HandleAWSSNSFlags(_, writeCmd, _ *kingpin.CmdClause, opts *Options)

func HandleAWSSQSFlags

func HandleAWSSQSFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)

func HandleActiveMqFlags

func HandleActiveMqFlags(readCmd, writeCmd *kingpin.CmdClause, opts *Options)

func HandleAzureFlags

func HandleAzureFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)

func HandleGCPPubSubFlags

func HandleGCPPubSubFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)

func HandleGlobalFlags

func HandleGlobalFlags(cmd *kingpin.CmdClause, opts *Options)

func HandleGlobalReadFlags

func HandleGlobalReadFlags(cmd *kingpin.CmdClause, opts *Options)

func HandleGlobalWriteFlags

func HandleGlobalWriteFlags(cmd *kingpin.CmdClause, opts *Options)

func HandleKafkaFlags

func HandleKafkaFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)

func HandleMQTTFlags

func HandleMQTTFlags(readCmd, writeCmd *kingpin.CmdClause, opts *Options)

func HandleNatsFlags

func HandleNatsFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)

func HandleRabbitFlags

func HandleRabbitFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)

func HandleRedisFlags

func HandleRedisFlags(readCmd, writeCmd, relayCmd *kingpin.CmdClause, opts *Options)

func HandleRelayFlags

func HandleRelayFlags(relayCmd *kingpin.CmdClause, opts *Options)

func ValidateProtobufOptions

func ValidateProtobufOptions(dirs []string, rootMessage string) error

Types

type AWSSNSOptions

type AWSSNSOptions struct {
	TopicArn string
}

type AWSSQSOptions

type AWSSQSOptions struct {
	// Shared
	QueueName       string
	RemoteAccountID string

	// Read
	ReadMaxNumMessages          int64
	ReadAutoDelete              bool
	ReadReceiveRequestAttemptId string
	ReadWaitTimeSeconds         int64

	// Write
	WriteDelaySeconds int64
	WriteAttributes   map[string]string

	// Relay
	RelayMaxNumMessages          int64
	RelayReceiveRequestAttemptId string
	RelayAutoDelete              bool
	RelayWaitTimeSeconds         int64
}

type ActiveMqOptions

type ActiveMqOptions struct {
	// Shared
	Address  string
	Topic    string
	Queue    string
	ClientId string

	// Read
	ReadFollow bool
}

type AzureServiceBusOptions

type AzureServiceBusOptions struct {
	// Shared
	Queue            string
	ConnectionString string
	Topic            string

	// Read
	Subscription string
}

type GCPPubSubOptions

type GCPPubSubOptions struct {
	// Shared
	ProjectId string

	// Read
	ReadSubscriptionId string
	ReadAck            bool

	// Write
	WriteTopicId string
}

type KafkaOptions

type KafkaOptions struct {
	// Shared
	Address            string
	Topic              string
	Timeout            time.Duration
	InsecureTLS        bool
	Username           string
	Password           string
	AuthenticationType string

	// Read
	ReadGroupId      string
	MaxWait          time.Duration
	MinBytes         int
	MaxBytes         int
	QueueCapacity    int
	RebalanceTimeout time.Duration
	CommitInterval   time.Duration

	// Write
	WriteKey string
}

type MQTTOptions

type MQTTOptions struct {
	// Shared
	Address  string
	Topic    string
	Timeout  time.Duration
	ClientId string
	QoSLevel int

	// TLS-related pieces
	TLSCAFile         string
	TLSClientCertFile string
	TLSClientKeyFile  string
	InsecureTLS       bool

	// Read
	ReadTimeout time.Duration

	// Write
	WriteTimeout time.Duration
}

type NatsOptions

type NatsOptions struct {
	// Shared
	Subject string
	Address string

	// TLS-related pieces
	TLSCAFile         string
	TLSClientCertFile string
	TLSClientKeyFile  string
	InsecureTLS       bool

	// Authentication
	CredsFile string
}

type Options

type Options struct {
	// Global
	Debug               bool
	Quiet               bool
	Stats               bool
	StatsReportInterval time.Duration
	Action              string
	Version             string
	Backend             string

	// Serializers
	AvroSchemaFile string

	// Relay
	RelayToken             string
	RelayGRPCAddress       string
	RelayType              string
	RelayHTTPListenAddress string
	RelayNumWorkers        int
	RelayGRPCTimeout       time.Duration
	RelayGRPCDisableTLS    bool
	RelayBatchSize         int

	// Shared read flags
	ReadProtobufRootMessage string
	ReadProtobufDirs        []string
	ReadOutputType          string
	ReadFollow              bool
	ReadLineNumbers         bool
	ReadConvert             string

	// Shared write flags
	WriteInputData           string
	WriteInputFile           string
	WriteInputType           string
	WriteOutputType          string
	WriteProtobufDirs        []string
	WriteProtobufRootMessage string

	Kafka     *KafkaOptions
	Rabbit    *RabbitOptions
	GCPPubSub *GCPPubSubOptions
	MQTT      *MQTTOptions
	AWSSQS    *AWSSQSOptions
	AWSSNS    *AWSSNSOptions
	ActiveMq  *ActiveMqOptions
	Redis     *RedisOptions
	Azure     *AzureServiceBusOptions
	Nats      *NatsOptions
}

func Handle

func Handle(cliArgs []string) (string, *Options, error)

type RabbitOptions

type RabbitOptions struct {
	// Shared
	Address       string
	Exchange      string
	RoutingKey    string
	UseTLS        bool
	SkipVerifyTLS bool

	// Read
	ReadQueue           string
	ReadQueueDurable    bool
	ReadQueueAutoDelete bool
	ReadQueueExclusive  bool
	ReadAutoAck         bool
	ReadQueueDeclare    bool
	ReadConsumerTag     string

	// Write
	WriteAppID string
}

type RedisOptions

type RedisOptions struct {
	// Shared
	Address string
	Channel string
}