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

fix issue

Browse Source
This commit is contained in:
joy.zhou
2019-09-30 11:04:46 +08:00
parent e74b9facd1
commit a3fc611615
7 changed files with 31 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Dockerfile
View File

@@ -1,7 +1,6 @@
FROM golang:1.12 as builder FROM golang:1.12 as builder
WORKDIR /go/src/github.com/fhmq/hmq WORKDIR /go/src/github.com/fhmq/hmq
COPY . . COPY . .
COPY ./vendor .
RUN CGO_ENABLED=0 go build -o hmq -a -ldflags '-extldflags "-static"' . RUN CGO_ENABLED=0 go build -o hmq -a -ldflags '-extldflags "-static"' .

3
broker/auth.go
View File

@@ -34,9 +34,6 @@ func (b *Broker) CheckTopicAuth(action, username, topic string) bool {
func (b *Broker) CheckConnectAuth(clientID, username, password string) bool { func (b *Broker) CheckConnectAuth(clientID, username, password string) bool {
if b.auth != nil { if b.auth != nil {
if clientID == "" || username == "" {
return false
}
return b.auth.CheckConnect(clientID, username, password) return b.auth.CheckConnect(clientID, username, password)
} }

4
broker/broker.go
View File

@@ -123,7 +123,9 @@ func (b *Broker) Start() {
return return
} }
go InitHTTPMoniter(b) if b.config.HTTPPort != "" {
go InitHTTPMoniter(b)
}
//listen clinet over tcp //listen clinet over tcp
if b.config.Port != "" { if b.config.Port != "" {

29
broker/config.go
View File

@@ -17,19 +17,20 @@ import (
) )
type Config struct { type Config struct {
Worker int `json:"workerNum"` Worker int `json:"workerNum"`
Host string `json:"host"` HTTPPort string `json:"httpPort"`
Port string `json:"port"` Host string `json:"host"`
Cluster RouteInfo `json:"cluster"` Port string `json:"port"`
Router string `json:"router"` Cluster RouteInfo `json:"cluster"`
TlsHost string `json:"tlsHost"` Router string `json:"router"`
TlsPort string `json:"tlsPort"` TlsHost string `json:"tlsHost"`
WsPath string `json:"wsPath"` TlsPort string `json:"tlsPort"`
WsPort string `json:"wsPort"` WsPath string `json:"wsPath"`
WsTLS bool `json:"wsTLS"` WsPort string `json:"wsPort"`
TlsInfo TLSInfo `json:"tlsInfo"` WsTLS bool `json:"wsTLS"`
Debug bool `json:"debug"` TlsInfo TLSInfo `json:"tlsInfo"`
Plugin Plugins `json:"plugins"` Debug bool `json:"debug"`
Plugin Plugins `json:"plugins"`
} }
type Plugins struct { type Plugins struct {
@@ -77,6 +78,8 @@ func ConfigureConfig(args []string) (*Config, error) {
fs.BoolVar(&help, "help", false, "Show this message.") fs.BoolVar(&help, "help", false, "Show this message.")
fs.IntVar(&config.Worker, "w", 1024, "worker num to process message, perfer (client num)/10.") fs.IntVar(&config.Worker, "w", 1024, "worker num to process message, perfer (client num)/10.")
fs.IntVar(&config.Worker, "worker", 1024, "worker num to process message, perfer (client num)/10.") fs.IntVar(&config.Worker, "worker", 1024, "worker num to process message, perfer (client num)/10.")
fs.StringVar(&config.HTTPPort, "httpport", "8080", "Port to listen on.")
fs.StringVar(&config.HTTPPort, "hp", "8080", "Port to listen on.")
fs.StringVar(&config.Port, "port", "1883", "Port to listen on.") fs.StringVar(&config.Port, "port", "1883", "Port to listen on.")
fs.StringVar(&config.Port, "p", "1883", "Port to listen on.") fs.StringVar(&config.Port, "p", "1883", "Port to listen on.")
fs.StringVar(&config.Host, "host", "0.0.0.0", "Network host to listen on") fs.StringVar(&config.Host, "host", "0.0.0.0", "Network host to listen on")

3
broker/lib/topics/memtopics.go
View File

@@ -244,6 +244,9 @@ func (this *snode) smatch(topic []byte, qos byte, subs *[]interface{}, qoss *[]b
// let's find the subscribers that match the qos and append them to the list. // let's find the subscribers that match the qos and append them to the list.
if len(topic) == 0 { if len(topic) == 0 {
this.matchQos(qos, subs, qoss) this.matchQos(qos, subs, qoss)
if mwcn, _ := this.snodes[MWC]; mwcn != nil {
mwcn.matchQos(qos, subs, qoss)
}
return nil return nil
} }

1
conf/hmq.config
View File

@@ -7,6 +7,7 @@
"host": "0.0.0.0", "host": "0.0.0.0",
"port": "1993" "port": "1993"
}, },
"httpPort": "8080",
"router": "127.0.0.1:9888", "router": "127.0.0.1:9888",
"tlsPort": "8883", "tlsPort": "8883",
"tlsHost": "0.0.0.0", "tlsHost": "0.0.0.0",

9
plugins/bridge/kafka.go
View File

@@ -5,6 +5,7 @@ import (
"errors" "errors"
"io/ioutil" "io/ioutil"
"strings" "strings"
"time"
"github.com/Shopify/sarama" "github.com/Shopify/sarama"
"go.uber.org/zap" "go.uber.org/zap"
@@ -109,11 +110,17 @@ func (k *kafka) publish(topics map[string]bool, key string, msg *Elements) error
} }
for topic, _ := range topics { for topic, _ := range topics {
k.kafkaClient.Input() <- &sarama.ProducerMessage{ select {
case k.kafkaClient.Input() <- &sarama.ProducerMessage{
Topic: topic, Topic: topic,
Key: sarama.ByteEncoder(key), Key: sarama.ByteEncoder(key),
Value: sarama.ByteEncoder(payload), Value: sarama.ByteEncoder(payload),
}:
continue
case <-time.After(5 * time.Second):
return errors.New("write kafka timeout")
} }
} }
return nil return nil