replace plugin

plugins/bridge/bridge.go

@@ -0,0 +1,49 @@
+package bridge
+
+import "github.com/fhmq/rhmq/logger"
+
+const (
+	//Connect mqtt connect
+	Connect = "connect"
+	//Publish mqtt publish
+	Publish = "publish"
+	//Subscribe mqtt sub
+	Subscribe = "subscribe"
+	//Unsubscribe mqtt sub
+	Unsubscribe = "unsubscribe"
+	//Disconnect mqtt disconenct
+	Disconnect = "disconnect"
+)
+
+var (
+	log = logger.Get().Named("bridge")
+)
+
+//Elements kafka publish elements
+type Elements struct {
+	ClientID  string `json:"clientid"`
+	Username  string `json:"username"`
+	Topic     string `json:"topic"`
+	Payload   string `json:"payload"`
+	Timestamp int64  `json:"ts"`
+	Size      int32  `json:"size"`
+	Action    string `json:"action"`
+}
+
+const (
+	//Kafka plugin name
+	Kafka = "kafka"
+)
+
+type BridgeMQ interface {
+	Publish(e *Elements) error
+}
+
+func NewBridgeMQ(name string) BridgeMQ {
+	switch name {
+	case Kafka:
+		return InitKafka()
+	default:
+		return &mockMQ{}
+	}
+}

plugins/bridge/kafka.go

@@ -0,0 +1,120 @@
+package bridge
+
+import (
+	"encoding/json"
+	"errors"
+	"io/ioutil"
+	"regexp"
+
+	"github.com/Shopify/sarama"
+	"go.uber.org/zap"
+)
+
+type kafakConfig struct {
+	Addr             []string          `json:"addr"`
+	ConnectTopic     string            `json:"onConnect"`
+	SubscribeTopic   string            `json:"onSubscribe"`
+	PublishTopic     string            `json:"onPublish"`
+	UnsubscribeTopic string            `json:"onUnsubscribe"`
+	DisconnectTopic  string            `json:"onDisconnect"`
+	RegexpMap        map[string]string `json:"regexpMap"`
+}
+
+type kafka struct {
+	kafakConfig kafakConfig
+	kafkaClient sarama.AsyncProducer
+}
+
+//Init init kafak client
+func InitKafka() *kafka {
+	log.Info("start connect kafka....")
+	content, err := ioutil.ReadFile("./plugins/mq/kafka/kafka.json")
+	if err != nil {
+		log.Fatal("Read config file error: ", zap.Error(err))
+	}
+	// log.Info(string(content))
+	var config kafakConfig
+	err = json.Unmarshal(content, &config)
+	if err != nil {
+		log.Fatal("Unmarshal config file error: ", zap.Error(err))
+	}
+	c := &kafka{kafakConfig: config}
+	c.connect()
+	return c
+}
+
+//connect
+func (k *kafka) connect() {
+	conf := sarama.NewConfig()
+	conf.Version = sarama.V1_1_1_0
+	kafkaClient, err := sarama.NewAsyncProducer(k.kafakConfig.Addr, conf)
+	if err != nil {
+		log.Fatal("create kafka async producer failed: ", zap.Error(err))
+	}
+
+	go func() {
+		for err := range kafkaClient.Errors() {
+			log.Error("send msg to kafka failed: ", zap.Error(err))
+		}
+	}()
+
+	k.kafkaClient = kafkaClient
+}
+
+//Publish publish to kafka
+func (k *kafka) Publish(e *Elements) error {
+	config := k.kafakConfig
+	key := e.ClientID
+	var topics []string
+	switch e.Action {
+	case Connect:
+		if config.ConnectTopic != "" {
+			topics = append(topics, config.ConnectTopic)
+		}
+	case Publish:
+		if config.PublishTopic != "" {
+			topics = append(topics, config.PublishTopic)
+		}
+		// foreach regexp map config
+		for reg, topic := range config.RegexpMap {
+			match, _ := regexp.MatchString(reg, e.Topic)
+			if match {
+				topics = append(topics, topic)
+			}
+		}
+	case Subscribe:
+		if config.SubscribeTopic != "" {
+			topics = append(topics, config.SubscribeTopic)
+		}
+	case Unsubscribe:
+		if config.UnsubscribeTopic != "" {
+			topics = append(topics, config.UnsubscribeTopic)
+		}
+	case Disconnect:
+		if config.DisconnectTopic != "" {
+			topics = append(topics, config.DisconnectTopic)
+		}
+	default:
+		return errors.New("error action: " + e.Action)
+	}
+
+	return k.publish(topics, key, e)
+
+}
+
+func (k *kafka) publish(topics []string, key string, msg *Elements) error {
+	payload, err := json.Marshal(msg)
+	if err != nil {
+		return err
+	}
+
+	for _, topic := range topics {
+		k.kafkaClient.Input() <- &sarama.ProducerMessage{
+			Topic: topic,
+			Key:   sarama.ByteEncoder(key),
+			Value: sarama.ByteEncoder(payload),
+		}
+	}
+
+	return nil
+}

plugins/bridge/kafka/kafka.json

@@ -0,0 +1,15 @@
+{
+    "addr": [
+        "127.0.0.1:9090"
+    ],
+    "onConnect": "onConnect",
+    "onPublish": "onPublish",
+    "onSubscribe": "onSubscribe",
+    "onDisconnect": "onDisconnect",
+    "onUnsubscribe": "onUnsubscribe",
+    "regexpMap": [
+        {
+            "^/(.+)/(.+)/upload/(.*)$": "upload"
+        }
+    ]
+}

plugins/bridge/mock.go

@@ -0,0 +1,7 @@
+package bridge
+
+type mockMQ struct{}
+
+func (m *mockMQ) Publish(e *Elements) error {
+	return nil
+}