kstream

type BuilderOption func(*DefaultBuilders)

type DefaultBuilders struct {
	Producer producer.Builder

	Consumer          consumer.Builder
	PartitionConsumer consumer.PartitionConsumerBuilder
	Store             store.Builder
	IndexedStore      store.IndexedStoreBuilder
	Backend           backend.Builder
	StateStore        store.StateStoreBuilder
	OffsetManager     offsets.Manager
	KafkaAdmin        admin.KafkaAdmin
	// contains filtered or unexported fields
}

type GlobalTable interface {
	Stream
}

type GlobalTableOffset int64

type GlobalTableOption func(options *globalTableOptions)

type GlobalTableStreamConfig struct {
	ConsumerBuilder consumer.PartitionConsumerBuilder
	BackendBuilder  backend.Builder
	OffsetManager   offsets.Manager
	KafkaAdmin      admin.KafkaAdmin
	Metrics         metrics.Reporter
	Logger          log.Logger
}

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

type InstancesOptions func(config *instancesOptions)

type KSink struct {
	Id              int32
	KeyEncoder      encoding.Encoder
	ValEncoder      encoding.Encoder
	Producer        producer.Producer
	ProducerBuilder producer.Builder

	TopicPrefix string

	Repartitioned bool

	KeyEncoderBuilder encoding.Builder
	ValEncoderBuilder encoding.Builder
	// contains filtered or unexported fields
}

type Option func(*kStreamOptions)

type Repartition struct {
	Enable       bool
	StreamSide   Side
	KeyEncoder   encoding.Builder
	ValueEncoder encoding.Builder
	Topic        RepartitionTopic
}

type RepartitionOption func(sink *RepartitionOptions)

type RepartitionOptions struct {
	LeftTopic        topic
	RightTopic       topic
	LeftRepartition  Repartition
	RightRepartition Repartition
}

type RepartitionTopic struct {
	Name              string
	Suffix            string
	ReplicationFactor int
	NumOfPartitions   int
	MinInSycReplicas  int
}

type Side int

type SinkOption func(sink *KSink)

type SinkRecord struct {
	Key, Value interface{}
	Timestamp  time.Time          // only set if kafka is version 0.10+, inner message timestamp
	Headers    data.RecordHeaders // only set if kafka is version 0.11+
}

type SourceNode struct {
	Id int32
	// contains filtered or unexported fields
}

type StoreWriter func(r *data.Record, store store.Store) error

type Stream interface {
	Branch(branches []branch.Details, opts ...Option) []Stream
	SelectKey(selectKeyFunc processors.SelectKeyFunc) Stream
	TransformValue(valueTransformFunc processors.ValueTransformFunc) Stream
	Transform(transformer processors.TransFunc) Stream
	Filter(filter processors.FilterFunc) Stream
	Process(processor processors.ProcessFunc) Stream
	JoinGlobalTable(table Stream, keyMapper join.KeyMapper, valMapper join.ValueMapper, typ join.Type) Stream
	JoinKTable(stream Stream, keyMapper join.KeyMapper, valMapper join.ValueMapper) Stream
	JoinStream(stream Stream, valMapper join.ValueMapper, opts ...RepartitionOption) Stream
	//LeftJoin(stream Stream, keyMapper join.KeyMapper, valMapper join.ValueMapper) Stream
	Through(topic string, keyEncoder encoding.Builder, valEncoder encoding.Builder, options ...SinkOption) Stream
	To(topic string, keyEncoder encoding.Builder, valEncoder encoding.Builder, options ...SinkOption)
}

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

type StreamBuilderConfig struct {
	ApplicationId    string
	AsyncProcessing  bool
	BootstrapServers []string // kafka Brokers
	WorkerPool       *worker_pool.PoolConfig
	Store            struct {
		BackendBuilder backend.Builder
		ChangeLog      struct {
			MinInSycReplicas  int // min number of insync replications in other nodes
			ReplicationFactor int
			Suffix            string
			Buffered          bool
			BufferedSize      int
		}
		Http struct {
			Enabled bool
			Host    string
		}
	}
	DLQ struct {
		Enabled          bool
		BootstrapServers []string
		TopicFormat      string
		//Type             dlq.DqlType // G, T
		Topic string // if global
	}
	Host      string
	ChangeLog struct {
		Enabled           bool
		Replicated        bool
		MinInSycReplicas  int // min number of insync replications in other nodes
		ReplicationFactor int
		Suffix            string
		Buffer            struct {
			Enabled       bool
			Size          int
			FlushInterval time.Duration
		}
	}
	Consumer      *consumer.Config
	ConsumerCount int
	*sarama.Config
	Producer        *producer.Config
	MetricsReporter metrics.Reporter
	Logger          log.Logger
	DefaultBuilders *DefaultBuilders
}

type StreamConfigs map[string]interface{}

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