feat(source): Implement new source event

adapter/rabbitmq/adapter.go

@@ -0,0 +1,10 @@
+package rabbitmq
+
+type Config struct {
+	UserName        string `koanf:"username"`
+	Password        string `koanf:"password"`
+	Host            string `koanf:"host"`
+	Port            int    `koanf:"port"`
+	Vhost           string `koanf:"vhost"`
+	ReconnectSecond int    `koanf:"reconnect_second"`
+}

cmd/source/faker/consumer/main.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"sync"
+
+	"github.com/labstack/gommon/log"
+	"github.com/ormushq/ormus/adapter/otela"
+	"github.com/ormushq/ormus/config"
+	"github.com/ormushq/ormus/destination/dconfig"
+	"github.com/ormushq/ormus/pkg/channel"
+	rbbitmqchannel "github.com/ormushq/ormus/pkg/channel/adapter/rabbitmq"
+	"github.com/ormushq/ormus/pkg/encoder"
+)
+
+func main() {
+	cfg := config.C()
+	done := make(chan bool)
+	wg := &sync.WaitGroup{}
+	dbConfig := dconfig.RabbitMQConsumerConnection{
+		User:            cfg.RabbitMq.UserName,
+		Password:        cfg.RabbitMq.Password,
+		Host:            cfg.RabbitMq.Host,
+		Port:            cfg.RabbitMq.Port,
+		Vhost:           cfg.RabbitMq.Vhost,
+		ReconnectSecond: cfg.RabbitMq.ReconnectSecond,
+	}
+	bufferSize := cfg.Source.BufferSize
+	numberInstants := cfg.Source.NumberInstants
+	maxRetryPolicy := cfg.Source.MaxRetry
+	eventName := cfg.Source.NewSourceEventName
+	err := otela.Configure(wg, done, otela.Config{Exporter: otela.ExporterConsole})
+	if err != nil {
+		panic(err.Error())
+	}
+
+	outputAdapter := rbbitmqchannel.New(done, wg, dbConfig)
+	err = outputAdapter.NewChannel(eventName, channel.OutputOnly, bufferSize, numberInstants, maxRetryPolicy)
+	if err != nil {
+		panic(err)
+	}
+	outputChannel, err := outputAdapter.GetOutputChannel(eventName)
+	if err != nil {
+		panic(err)
+	}
+
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+		for msg := range outputChannel {
+			m := encoder.DecodeNewSourceEvent(string(msg.Body))
+			log.Info(m.WriteKey)
+
+			if err := msg.Ack(); err != nil {
+				panic(err)
+			}
+		}
+	}()
+	wg.Wait()
+}

cmd/source/faker/publisher/main.go

@@ -0,0 +1,61 @@
+package main
+
+import (
+	"sync"
+
+	"github.com/google/uuid"
+	"github.com/ormushq/ormus/adapter/otela"
+	"github.com/ormushq/ormus/config"
+	"github.com/ormushq/ormus/contract/go/source"
+	"github.com/ormushq/ormus/destination/dconfig"
+	"github.com/ormushq/ormus/pkg/channel"
+	rbbitmqchannel "github.com/ormushq/ormus/pkg/channel/adapter/rabbitmq"
+	"github.com/ormushq/ormus/pkg/encoder"
+)
+
+func main() {
+	cfg := config.C()
+	done := make(chan bool)
+	wg := &sync.WaitGroup{}
+	dbConfig := dconfig.RabbitMQConsumerConnection{
+		User:            cfg.RabbitMq.UserName,
+		Password:        cfg.RabbitMq.Password,
+		Host:            cfg.RabbitMq.Host,
+		Port:            cfg.RabbitMq.Port,
+		Vhost:           cfg.RabbitMq.Vhost,
+		ReconnectSecond: cfg.RabbitMq.ReconnectSecond,
+	}
+	bufferSize := cfg.Source.BufferSize
+	numberInstants := cfg.Source.NumberInstants
+	maxRetryPolicy := cfg.Source.MaxRetry
+	testCount := 100
+
+	err := otela.Configure(wg, done, otela.Config{Exporter: otela.ExporterConsole})
+	if err != nil {
+		panic(err.Error())
+	}
+	inputAdapter := rbbitmqchannel.New(done, wg, dbConfig)
+	err = inputAdapter.NewChannel(cfg.Source.NewSourceEventName, channel.InputOnlyMode, bufferSize, numberInstants, maxRetryPolicy)
+	if err != nil {
+		panic(err.Error())
+	}
+	inputChannel, err := inputAdapter.GetInputChannel(cfg.Source.NewSourceEventName)
+	if err != nil {
+		panic(err.Error())
+	}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for messageID := 0; messageID < testCount; messageID++ {
+			msg := encoder.EncodeNewSourceEvent(&source.NewSourceEvent{
+				ProjectId: uuid.New().String(),
+				OwnerId:   uuid.New().String(),
+				WriteKey:  uuid.New().String(),
+			})
+			inputChannel <- []byte(msg)
+
+		}
+	}()
+	wg.Wait()
+}

cmd/source/main.go

@@ -1,16 +1,24 @@
 package main
 
 import (
+	"context"
 	"log/slog"
 	"os"
 	"os/signal"
 	"sync"
 
 	"github.com/ormushq/ormus/adapter/otela"
+	"github.com/ormushq/ormus/adapter/redis"
 	"github.com/ormushq/ormus/config"
+	"github.com/ormushq/ormus/destination/dconfig"
 	"github.com/ormushq/ormus/logger"
+	"github.com/ormushq/ormus/pkg/channel"
+	rbbitmqchannel "github.com/ormushq/ormus/pkg/channel/adapter/rabbitmq"
 	"github.com/ormushq/ormus/source/delivery/httpserver"
 	"github.com/ormushq/ormus/source/delivery/httpserver/statushandler"
+	sourceevent "github.com/ormushq/ormus/source/eventhandler"
+	writekeyrepo "github.com/ormushq/ormus/source/repository/redis/rediswritekey"
+	"github.com/ormushq/ormus/source/service/writekey"
 )
 
 //	@termsOfService	http://swagger.io/terms/
@@ -58,6 +66,14 @@ func main() {
 	handlers := []httpserver.Handler{
 		statushandler.New(),
 	}
+	err := otela.Configure(wg, done, otela.Config{Exporter: otela.ExporterConsole})
+	if err != nil {
+		panic(err.Error())
+	}
+
+	cfg := config.C()
+	_, Consumer := SetupSourceServices(cfg)
+	Consumer.Consume(context.Background(), cfg.Source.NewSourceEventName, done, wg, Consumer.ProcessNewSourceEvent)
 
 	//----------------- Setup Tracer -----------------//
 	otelcfg := otela.Config{
@@ -68,7 +84,7 @@ func main() {
 		MetricExposePort:   config.C().Source.Otel.MetricExposePort,
 		Exporter:           otela.ExporterGrpc,
 	}
-	err := otela.Configure(wg, done, otelcfg)
+	err = otela.Configure(wg, done, otelcfg)
 	if err != nil {
 		l.Error(err.Error())
 	}
@@ -86,3 +102,32 @@ func main() {
 	close(done)
 	wg.Wait()
 }
+
+func SetupSourceServices(cfg config.Config) (writekey.Service, sourceevent.Consumer) {
+	done := make(chan bool)
+	wg := &sync.WaitGroup{}
+	dbConfig := dconfig.RabbitMQConsumerConnection{
+		User:            cfg.RabbitMq.UserName,
+		Password:        cfg.RabbitMq.Password,
+		Host:            cfg.RabbitMq.Host,
+		Port:            cfg.RabbitMq.Port,
+		Vhost:           cfg.RabbitMq.Vhost,
+		ReconnectSecond: cfg.RabbitMq.ReconnectSecond,
+	}
+	outputAdapter := rbbitmqchannel.New(done, wg, dbConfig)
+	err := outputAdapter.NewChannel(cfg.Source.NewSourceEventName, channel.OutputOnly, cfg.Source.BufferSize, cfg.Source.NumberInstants, cfg.Source.MaxRetry)
+	if err != nil {
+		panic(err)
+	}
+
+	adapter, err := redis.New(cfg.Redis)
+	if err != nil {
+		panic(err)
+	}
+
+	writeKeyRepo := writekeyrepo.New(adapter)
+	writeKeySvc := writekey.New(&writeKeyRepo, cfg.Source)
+	eventHandler := sourceevent.New(outputAdapter, writeKeySvc)
+
+	return writeKeySvc, *eventHandler
+}

config.yml

@@ -4,6 +4,11 @@ source:
   http_server:
     port: 8080
     network: "tcp"
+  write_key_expiration: 120
+  new_source_event_name: "new-source-event"
+  buffersize: 100
+  number_instants: 10
+  maxretry: 5
   otel:
     endpoint: "otel_collector:4317"
     service_name: "source"
@@ -19,6 +24,13 @@ redis:
   host: 127.0.0.1
   db: 0
   password: ""
+rabbitmq:
+  host: "localhost"
+  port: 5672
+  password: "guest"
+  username: "guest"
+  reconnect_second: 1
+  vhost: "/"
 etcd:
   port: 2379
   host: 127.0.0.1

config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"github.com/ormushq/ormus/adapter/etcd"
+	"github.com/ormushq/ormus/adapter/rabbitmq"
 	"github.com/ormushq/ormus/adapter/redis"
 	"github.com/ormushq/ormus/adapter/scylladb"
 	"github.com/ormushq/ormus/destination/dconfig"
@@ -12,6 +13,7 @@ import (
 type Config struct {
 	Redis       redis.Config    `koanf:"redis"`
 	Etcd        etcd.Config     `koanf:"etcd"`
+	RabbitMq    rabbitmq.Config `koanf:"rabbitmq"`
 	Manager     manager.Config  `koanf:"manager"`
 	Source      source.Config   `koanf:"source"`
 	Destination dconfig.Config  `koanf:"destination"`