Feature/sink kafka click

config.go

@@ -6,6 +6,9 @@ package pipelaner
 
 import (
 	"errors"
+	"fmt"
+	"reflect"
+	"time"
 
 	"github.com/BurntSushi/toml"
 	"github.com/mitchellh/mapstructure"
@@ -58,6 +61,41 @@ type BaseLaneConfig struct {
 	Internal
 }
 
+type KafkaConfig struct {
+	KafkaBrokers           string   `pipelane:"brokers"`
+	KafkaVersion           string   `pipelane:"version"`
+	KafkaOffsetNewest      bool     `pipelane:"offset_newest"`
+	KafkaSASLEnabled       bool     `pipelane:"sasl_enabled"`
+	KafkaSASLMechanism     string   `pipelane:"sasl_mechanism"`
+	KafkaSASLUsername      string   `pipelane:"sasl_username"`
+	KafkaSASLPassword      string   `pipelane:"sasl_password"`
+	KafkaAutoCommitEnabled bool     `pipelane:"auto_commit_enabled"`
+	KafkaConsumerGroupId   string   `pipelane:"consumer_group_id"`
+	KafkaTopics            []string `pipelane:"topics"`
+	KafkaAutoOffsetReset   string   `pipelane:"auto_offset_reset"`
+	KafkaBatchSize         int      `pipelane:"batch_size"`
+	KafkaSchemaRegistry    string   `pipelane:"schema_registry"`
+	Internal
+}
+
+type ClickHouseConfig struct {
+	Address                  string        `pipelane:"address"`
+	User                     string        `pipelane:"user"`
+	Password                 string        `pipelane:"password"`
+	Database                 string        `pipelane:"database"`
+	MigrationEngine          string        `pipelane:"migration_engine"`
+	MigrationsPathClickhouse string        `pipelane:"migrations_path_clickhouse"`
+	MaxExecutionTime         time.Duration `pipelane:"max_execution_time"`
+	ConnMaxLifetime          time.Duration `pipelane:"conn_max_lifetime"`
+	DialTimeout              time.Duration `pipelane:"dial_timeout"`
+	MaxOpenConns             int           `pipelane:"max_open_conns"`
+	MaxIdleConns             int           `pipelane:"max_idle_conns"`
+	BlockBufferSize          uint8         `pipelane:"block_buffer_size"`
+	MaxCompressionBuffer     string        `pipelane:"max_compression_buffer"`
+	EnableDebug              bool          `pipelane:"enable_debug"`
+	Internal
+}
+
 func newConfig(c map[string]any) (*config, error) {
 	cfg := &config{}
 	dC := &mapstructure.DecoderConfig{
@@ -130,6 +168,24 @@ func NewBaseConfig(val map[string]any) (*BaseLaneConfig, error) {
 	return &cfg, nil
 }
 
+func NewKafkaConfig(val map[string]any) (*KafkaConfig, error) {
+	var cfg KafkaConfig
+	err := decode(val, &cfg)
+	if err != nil {
+		return nil, err
+	}
+	return &cfg, nil
+}
+
+func NewClickHouseConfig(val map[string]any) (*ClickHouseConfig, error) {
+	var cfg ClickHouseConfig
+	err := decode(val, &cfg)
+	if err != nil {
+		return nil, err
+	}
+	return &cfg, nil
+}
+
 func (c *BaseLaneConfig) ParseExtended(v any) error {
 	err := decode(c._extended, v)
 	if err != nil {
@@ -154,3 +210,41 @@ func decode(input map[string]any, output any) error {
 	}
 	return nil
 }
+
+func CastConfig[K, V any](config K) *V {
+	if !(reflect.TypeOf(config).Kind() == reflect.Ptr && reflect.TypeOf(config).Elem().Kind() == reflect.Struct) {
+		panic("Config is not struct")
+	}
+
+	res := new(V)
+	if reflect.ValueOf(res).Kind() == reflect.Struct {
+		panic("V is not struct")
+	}
+
+	val := reflect.ValueOf(config).Elem()
+	for i := 0; i < val.NumField(); i++ {
+		fieldName := val.Type().Field(i).Name
+
+		valueField := val.FieldByName(fieldName)
+		if !valueField.IsValid() {
+			panic(fmt.Sprintf("No such field: %s in valueField", fieldName))
+		}
+
+		setField := reflect.ValueOf(res).Elem().FieldByName(fieldName)
+		if !setField.IsValid() {
+			panic(fmt.Sprintf("No such field: %s in setField", fieldName))
+		}
+
+		if valueField.Type() != setField.Type() {
+			panic(fmt.Sprintf("Cannot cast %s to %s", valueField.Type(), setField.Type()))
+		}
+
+		if !setField.CanSet() {
+			panic(fmt.Sprintf("Cannot set %s field value", fieldName))
+		}
+
+		setField.Set(valueField)
+	}
+
+	return res
+}

example/main.go

@@ -6,6 +6,7 @@ package main
 
 import (
 	"github.com/pipelane/pipelaner/example/tests"
+	"github.com/pipelane/pipelaner/source/sink/console"
 	"time"
 
 	"github.com/pipelane/pipelaner"
@@ -17,6 +18,7 @@ func main() {
 	agent, err := pipelaner.NewAgent(
 		"example/pipeline.toml",
 	)
+	pipelaner.RegisterSink("int", &console.Console{})
 	if err != nil {
 		panic(err)
 	}

go.mod

@@ -1,28 +1,63 @@
 module github.com/pipelane/pipelaner
 
-go 1.22
+go 1.22.3
+
+toolchain go1.22.4
 
 require (
 	github.com/BurntSushi/toml v1.4.0
+	github.com/confluentinc/confluent-kafka-go/v2 v2.5.0
 	github.com/expr-lang/expr v1.16.9
 	github.com/goccy/go-json v0.10.3
+	github.com/huandu/go-sqlbuilder v1.27.3
 	github.com/mitchellh/mapstructure v1.5.0
+	github.com/pipelane/go-kit v0.0.0-20240716103119-fe9cc6ce7913
 	github.com/rs/zerolog v1.33.0
 	github.com/stretchr/testify v1.9.0
 	google.golang.org/grpc v1.65.0
 	google.golang.org/protobuf v1.34.2
 )
 
 require (
+	github.com/ClickHouse/ch-go v0.61.5 // indirect
+	github.com/ClickHouse/clickhouse-go/v2 v2.25.0 // indirect
 	github.com/LastPossum/kamino v0.0.1 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/andybalholm/brotli v1.1.0 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/go-faster/city v1.0.1 // indirect
+	github.com/go-faster/errors v0.7.1 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/hcl v1.0.0 // indirect
+	github.com/huandu/xstrings v1.4.0 // indirect
+	github.com/klauspost/compress v1.17.8 // indirect
+	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
-	github.com/mattn/go-isatty v0.0.19 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/paulmach/orb v0.11.1 // indirect
+	github.com/pelletier/go-toml/v2 v2.2.2 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/sagikazarmark/locafero v0.4.0 // indirect
+	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
+	github.com/segmentio/asm v1.2.0 // indirect
+	github.com/shopspring/decimal v1.4.0 // indirect
+	github.com/sourcegraph/conc v0.3.0 // indirect
+	github.com/spf13/afero v1.11.0 // indirect
+	github.com/spf13/cast v1.6.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/spf13/viper v1.19.0 // indirect
+	github.com/subosito/gotenv v1.6.0 // indirect
+	go.opentelemetry.io/otel v1.26.0 // indirect
+	go.opentelemetry.io/otel/trace v1.26.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect
 	golang.org/x/net v0.25.0 // indirect
 	golang.org/x/sys v0.20.0 // indirect
 	golang.org/x/text v0.15.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect
+	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )