collab/hmq
3
0
Fork 0
mirror of https://github.com/fhmq/hmq.git synced 2026-05-02 14:28:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
joy.zhou
2019-07-15 19:32:59 +08:00
parent a3edbcc2ce
commit 67e39710e9
1 changed files with 1 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
plugins/kafka/kafka.go
View File

@@ -3,7 +3,6 @@ package kafka
import ( import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt"
"io/ioutil" "io/ioutil"
"regexp" "regexp"
"time" "time"
@@ -55,6 +54,7 @@ func Init() {
func connect() { func connect() {
var err error var err error
conf := sarama.NewConfig() conf := sarama.NewConfig()
conf.Version = sarama.V1_1_1_0
kafkaClient, err = sarama.NewAsyncProducer(config.Addr, conf) kafkaClient, err = sarama.NewAsyncProducer(config.Addr, conf)
if err != nil { if err != nil {
log.Fatal("create kafka async producer failed: ", zap.Error(err)) log.Fatal("create kafka async producer failed: ", zap.Error(err))
@@ -94,12 +94,9 @@ func Publish(e *plugins.Elements) {
if err != nil { if err != nil {
log.Error("publish kafka error: ", zap.Error(err)) log.Error("publish kafka error: ", zap.Error(err))
} }
fmt.Println("------", e.Topic)
match, _ := regexp.MatchString(_ThingModelTopicRegexp, e.Topic) match, _ := regexp.MatchString(_ThingModelTopicRegexp, e.Topic)
if match && e.Action == plugins.Publish { if match && e.Action == plugins.Publish {
fmt.Println("------ match", e.Topic)
topic := "tmodel.msg.upstream" topic := "tmodel.msg.upstream"
err := publish(topic, key, e) err := publish(topic, key, e)
if err != nil { if err != nil {