conf

Documentation

Index

• func LoadConfig(config *InternalProcessorConfig) error
• type ConsumerConfig
• type GlobalKafkaConfig
• type GlobalStoreConfig
• type InitialTopic
• type InternalKafkaConfig
• type InternalProcessorConfig
  • func Config() *InternalProcessorConfig
  • func NewInternalProcessorConfig(rootConfig *RootConfig, processorName string) *InternalProcessorConfig
  • func (ipc *InternalProcessorConfig) Validate() error
• type KafkaConfig
• type MemcachedConfig
  • func GetMemcachedConfigByName(name string) *MemcachedConfig
• type NamedKafkaConfig
• type OutputConfig
• type ProcessorConfig
• type RedisConfig
  • func GetRedisConfigByName(name string) *RedisConfig
• type RootConfig
  • func (rc *RootConfig) Validate() error

Functions

func LoadConfig

func LoadConfig(config *InternalProcessorConfig) error

Types

type ConsumerConfig

type ConsumerConfig struct {
	Address      string `json:"address"`
	Topic        string `json:"topic"`
	NumOfWorkers int    `json:"num_of_workers"`
	TopicIndex   int    `json:"topic_index"`
}

type GlobalKafkaConfig

type GlobalKafkaConfig struct {
	Address       string          `yaml:"address"`
	InitialTopics []*InitialTopic `yaml:"initialTopics"`
}

type GlobalStoreConfig

type GlobalStoreConfig struct {
	RedisConfigs     []*RedisConfig     `yaml:"redisConfigs"`
	MemcachedConfigs []*MemcachedConfig `yaml:"memcachedConfigs"`
}

type InitialTopic

type InitialTopic struct {
	Topic      string `yaml:"topic"`
	Partitions int    `yaml:"partitions"`
}

type InternalKafkaConfig

type InternalKafkaConfig struct {
	Address    string
	Topic      string
	Partitions int
}

type InternalProcessorConfig

type InternalProcessorConfig struct {
	Name                     string                  `json:"name"`
	Type                     string                  `json:"type"`
	GlobalKafkaConfig        *GlobalKafkaConfig      `json:"global_kafka_config"`
	ConsumerConfigs          []*ConsumerConfig       `json:"consumer_configs"`
	DefaultOutputKafkaConfig *KafkaConfig            `json:"default_output_kafka_config"`
	OutputKafkaConfigs       map[string]*KafkaConfig `json:"output_kafka_configs"`
	GlobalStoreConfig        *GlobalStoreConfig      `json:"global_store_config"`
	WindowSize               int                     `json:"window_size"`
}

func Config

func Config() *InternalProcessorConfig

func NewInternalProcessorConfig

func NewInternalProcessorConfig(rootConfig *RootConfig, processorName string) *InternalProcessorConfig

func (*InternalProcessorConfig) Validate

func (ipc *InternalProcessorConfig) Validate() error

type KafkaConfig

type KafkaConfig struct {
	Address string `yaml:"address"`
	Topic   string `yaml:"topic"`
}

type MemcachedConfig

type MemcachedConfig struct {
	Name      string   `yaml:"name"`
	Addresses []string `yaml:"addresses"`
}

func GetMemcachedConfigByName

func GetMemcachedConfigByName(name string) *MemcachedConfig

type NamedKafkaConfig

type NamedKafkaConfig struct {
	Name    string `yaml:"name"`
	Address string `yaml:"address"`
	Topic   string `yaml:"topic"`
}

type OutputConfig

type OutputConfig struct {
	// DefaultTopicPartitions defines the number of partitions of the default output topic of the processor.
	// If DefaultTopicPartitions is set to 0, the default topic of the processor will not be created.
	DefaultTopicPartitions int `yaml:"defaultTopicPartitions"`

	OutputProcessors []string `yaml:"outputProcessors"`

	// OutputKafkaConfigs defines non-processor destinations. The name of a self-defined OutputKafkaConfig
	// must not be the same as any of the processor names, and must be unique among all the OutputKafkaConfigs.
	OutputKafkaConfigs []*NamedKafkaConfig `yaml:"outputKafkaConfigs"`
}

type ProcessorConfig

type ProcessorConfig struct {
	// ParentDirectory is the parent directory that stores user-defined go files.
	ParentDirectory string `yaml:"parentDirectory"`

	// Files are the paths of user-defined go files, which includes fission handler, process function, init function.
	// If ParentDirectory is not empty, the file paths will be treate as paths relative to parent directory.
	Files []string `yaml:"files"`

	// EntryPoint is the entry point function of the processor.
	EntryPoint string `yaml:"entryPoint"`

	// Name is the name of the processor, which should be unique among all processors in a config.
	Name string `yaml:"name"`

	Type string `yaml:"type"`

	// NumOfWorker defines the number of workers in processor. Note that the number of parallel workers
	// in a processor is capped by the number of partitions in that processor's source Kafka topic.
	NumOfWorker int `yaml:"numOfWorker"`

	// InputProcessors and InputKafkaConfigs define the source of a processor's input. Either InputProcessor
	// or InputKafkaConfig must be not empty. If both are specified, InputKafkaConfig will be used.
	InputProcessors   []string       `yaml:"inputProcessors"`
	InputKafkaConfigs []*KafkaConfig `yaml:"inputKafkaConfigs"`

	// OutputConfig defines the output of a processor's input. OutputConfig must be specified in a config.
	OutputConfig *OutputConfig `yaml:"outputConfig"`

	WindowSize int `yaml:"windowSize"`
}

type RedisConfig

type RedisConfig struct {
	Name    string `yaml:"name"`
	Address string `yaml:"address"`
}

func GetRedisConfigByName

func GetRedisConfigByName(name string) *RedisConfig

type RootConfig

type RootConfig struct {
	// ProcessorConfigs defines all the processors in the stream processing pipeline. At least one processor config
	// must be provided.
	ProcessorConfigs []*ProcessorConfig `yaml:"processorConfigs"`

	// GlobalStoreConfig defines all global stores used in the stream processing pipeline.
	GlobalStoreConfig *GlobalStoreConfig `yaml:"globalStoreConfig"`

	// GlobalKafkaConfig specifies which Kafka cluster the interim topics should be created on.
	GlobalKafkaConfig *GlobalKafkaConfig `yaml:"globalKafkaConfig"`
}

func (*RootConfig) Validate

func (rc *RootConfig) Validate() error