✨ feat: added kafka pre-service #50

queues/apache_kafka_model.go

@@ -55,6 +55,7 @@ type ClusterMultiTenantKafkaConfig struct {
 }
 
 type KafkaPublisherRequest struct {
-	Topic string      `json:"topic" binding:"required" yaml:"topic"`
-	Data  interface{} `json:"data" yaml:"data"`
+	TopicKey  string                 `json:"topic_key" binding:"required" yaml:"topic_key"`
+	TenantKey string                 `json:"tenant_key" binding:"required" yaml:"tenant_key"`
+	Payload   map[string]interface{} `json:"payload" yaml:"payload"`
 }

queues/apache_kafka_producer.go

@@ -1,6 +1,10 @@
 package queues
 
-import "github.com/sivaosorg/govm/utils"
+import (
+	"fmt"
+
+	"github.com/sivaosorg/govm/utils"
+)
 
 func NewKafkaProducerConfig() *KafkaProducerConfig {
 	return &KafkaProducerConfig{
@@ -52,3 +56,36 @@ func GetKafkaProducerConfigSample() *KafkaProducerConfig {
 		AppendProperty("retries", 3).AppendProperty(Bootstrap_Servers, "kafka-broker-3:9092")
 	return k
 }
+
+func NewKafkaPublisherRequest() *KafkaPublisherRequest {
+	return &KafkaPublisherRequest{}
+}
+
+func (k *KafkaPublisherRequest) SetTopicKey(value string) *KafkaPublisherRequest {
+	k.TopicKey = value
+	return k
+}
+
+func (k *KafkaPublisherRequest) SetTenantKey(value string) *KafkaPublisherRequest {
+	k.TenantKey = value
+	return k
+}
+
+func (k *KafkaPublisherRequest) SetPayload(value map[string]interface{}) *KafkaPublisherRequest {
+	k.Payload = value
+	return k
+}
+
+func (k *KafkaPublisherRequest) Json() string {
+	return utils.ToJson(k)
+}
+
+func KafkaPublisherRequestValidator(k KafkaPublisherRequest) error {
+	if utils.IsEmpty(k.TopicKey) {
+		return fmt.Errorf("Topic key is required")
+	}
+	if utils.IsEmpty(k.TenantKey) {
+		return fmt.Errorf("Tenant key is required")
+	}
+	return nil
+}

queues/apache_kafka_service.go

@@ -1 +1,53 @@
 package queues
+
+import (
+	"github.com/sivaosorg/govm/apix"
+)
+
+type KafkaService interface {
+}
+
+type kafkaServiceImpl struct {
+	conf []apix.ApiRequestConfig
+}
+
+func NewKafkaService(conf []apix.ApiRequestConfig) KafkaService {
+	return &kafkaServiceImpl{
+		conf: conf,
+	}
+}
+
+func NewKafkaServiceSlices(conf ...apix.ApiRequestConfig) KafkaService {
+	return &kafkaServiceImpl{
+		conf: conf,
+	}
+}
+
+// func (s *kafkaServiceImpl) ProduceCallback(request KafkaPublisherRequest) (*restify.Response, error) {
+// err := KafkaPublisherRequestValidator(request)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+// 	if len(s.conf) == 0 {
+// 		return nil, fmt.Errorf("API Conf is required")
+// 	}
+// 	configs, ok := apix.Get(s.conf, request.TenantKey)
+// 	if !ok {
+// 		return nil, fmt.Errorf("Tenant Key undefined: %v", request.TenantKey)
+// 	}
+// 	endpoint, err := configs.GetEndpoint(request.TopicKey)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+// 	if len(request.Payload) > 0 {
+// 		for k, v := range request.Payload {
+// 			endpoint.AppendBodyWith(k, v)
+// 		}
+// 	}
+// 	svc := apix.NewApiService(configs)
+// 	return svc.Do(nil, endpoint)
+// }
+
+// func (s *kafkaServiceImpl) ProduceCallbackNoneWait(request KafkaPublisherRequest) {
+// 	go s.ProduceCallback(request)
+// }