谢谢大佬，我还没测出跨集群效果，单集群效果ok的，请教一个问题，你连接redis有没有出现Error: Connection reset by peer？因为跨集群只用一个redis服务，我从节点连主redis服务一直报 Connection reset by peer这个错误。。。搞了好久。感觉就差这一步了。

Originally posted by @skygp in #9 (comment)

目前看配置好像需要用到redis，没有使用redis，程序启动会失败，请问可以不用redis也能正常运行吗？那该如何修改呢？

Using some code based on cmd/single/main.go with MySQL configuration in .yml, every time a client connected to the broker resulted in "panic: runtime error: invalid memory address or nil pointer dereference"

I eventually traced this to a typo in the field names I had put in my .yml. The result was that the code:
a.authStmt, _ = sqlxDB.Preparex(authSql)
.. was failing as the SQL was invalid.

However there's no code to catch that error and so when the client connects,
err := a.authStmt.QueryRowx(key).Scan(&password, &allow)
.. in OnConnectAuthenticate was crashing.

This is the first time I've worked in Go so I'm not going to attempt to suggest a suitable fix - for now I'll just be careful about config file typos!

Firstly - thank you for this amazing library.

To save people the trouble if you're looking to use a DB other than Redis or PostgreSQL, here's a quick guide to doing authentication. In my case, i needed to access users in Mongo which had been stored by Laravel using Bcrypt. It wasn't amazingly clear when parsing the examples; particularly the use of packets.

This is obviously debug code (printing out user/password etc), and there's more involved. But it's a rough overview.

package main

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"os/signal"
	"strconv"
	"syscall"
	"time"

	"github.com/wind-c/comqtt/v2/mqtt"
	"github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
	"github.com/wind-c/comqtt/v2/mqtt/hooks/debug"
	"github.com/wind-c/comqtt/v2/mqtt/listeners"
	"github.com/wind-c/comqtt/v2/mqtt/packets"
	"golang.org/x/crypto/bcrypt"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

type Auth struct {
    mqtt.HookBase 
    mongo_client   *mongo.Client
    database       string
    collection string
}

func auth_hook(mongo_client *mongo.Client) *Auth {
    return &Auth{
		mongo_client: mongo_client,
		database:   env("DB_DATABASE", "mycooldb"),
		collection: env("DB_AUTH_COLLECTION", "users"),
	}
}

func (hook *Auth) ID() string {
	return "mongodb-auth"
}

func (hook *Auth) Provides(b byte) bool {

	return b == mqtt.OnConnectAuthenticate
}

func (hook *Auth) OnConnectAuthenticate(cl *mqtt.Client, pk packets.Packet) bool {

	log.Printf("Client trying to connect with username: %s, password: %v", string(pk.Connect.Username), string(pk.Connect.Password));

	var User bson.M

	err := users.FindOne(ctx, bson.M{env("DB_AUTH_USERNAME_FIELD", "username"): string(pk.Connect.Username)}).Decode(&User)
	
	if err != nil {
		log.Printf("Authentication failed for user %s: %v", string(pk.Connect.Username), err)

		return false // Authentication failed
	}

	password, ok := User[env("DB_AUTH_PASSWORD_FIELD", "password")].(string)

	if !ok {

		log.Printf("Invalid password format for user %s", string(pk.Connect.Username))

		return false
	}

	// Compare the provided password with the stored hashed password
	err = bcrypt.CompareHashAndPassword([]byte(password), []byte(pk.Connect.Password))

	if err != nil {
		
		log.Printf("Bcrypt hash comparison failed. Invalid password for user %s", string(pk.Connect.Username))

		return false // Authentication failed
	}

	log.Printf("User %s authenticated successfully.", string(pk.Connect.Username))
	return true
}

func env(key, def string) string {
	value := os.Getenv(key)

	if len(value) == 0 {
		return def
	}

	return value
}

func main() {
	done := make(chan bool, 1)
	sigs := make(chan os.Signal, 1)

	// Register the signals to catch
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		sig := <-sigs
		log.Printf("Received signal: %s", sig)
		done <- true
	}()

       // Other stuff 
	
	mongo_uri_flag := flag.String("mongo_uri", env("MONGO_URI", "mongodb://user:pass@localhost:27017/?authSource=admin"), "MongoDB URI")

        mongo_client := mongo_client(*mongo_uri_flag)
        
        // vars declared elsewhere
        mqtt_server := mqtt.New(&mqtt.Options{
		Capabilities: &mqtt.Capabilities{
			MaximumSessionExpiryInterval: uint32(max_session_expiry_interval),
			Compatibilities: mqtt.Compatibilities{
				ObscureNotAuthorized: obscure_not_authorized,
			},
		},
		ClientNetWriteBufferSize: client_net_write_buffer_size,
		ClientNetReadBufferSize:  client_net_read_buffer_size,
		SysTopicResendInterval:   int64(sys_topic_resend_interval),
	})

	debug_hook := new(debug.Hook)

	err := mqtt_server.AddHook(debug_hook, &debug.Options{})

        if err != nil {
            log.Fatalf("Failed to add debug hook: %v", err)
        }

        err = mqtt_server.AddHook(auth_hook(mongo_client), nil)

        if err != nil {
	        log.Fatalf("Failed to add MongoDB authentication hook: %v", err)
        }
        
	// Start the MQTT server
          go func() {
              if err := mqtt_server.Serve(); err != nil {
                  log.Fatalf("Buzz MQTT server failed to start: %v", err)
              }
          }()
          
         // loads of other stuff
          
        <-done
        log.Println("Shutting down the server...")
    
        // Clean up resources here if necessary
        if err := mqtt_server.Close(); err != nil {
            log.Printf("Error shutting down server: %v", err)
        } else {
            log.Println("Server shutdown successfully.")
        }
}

With the broker embedded into my app I get the following panic during testing, both when publishing directly (server.Publish) or using the paho client.

The i.ks list gets filled with *InflightMessage (inflight_map.go:34) but after reading tries to cast to uint16 (inflight_map.go:32)

panic: interface conversion: interface {} is *clients.InflightMessage, not uint16

goroutine 50 [running]:
github.com/wind-c/comqtt/server/internal/clients.(*InflightMap).Set(0xc0007ca060, 0x21c, 0xc0006b1ce0)
	/home/james/projects/web-mon/server/vendor/github.com/wind-c/comqtt/server/internal/clients/inflight_map.go:32 +0x2fe
github.com/wind-c/comqtt/server.(*Server).publishToSubscribers(_, {{0x2b4, 0x3, 0x1, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, ...}, ...})
	/home/james/projects/web-mon/server/vendor/github.com/wind-c/comqtt/server/server.go:654 +0x498
github.com/wind-c/comqtt/server.(*Server).processPublish(_, _, {{0x2b4, 0x3, 0x1, 0x0, 0x0}, {0x0, 0x0, 0x0}, ...})
	/home/james/projects/web-mon/server/vendor/github.com/wind-c/comqtt/server/process.go:149 +0x678
github.com/wind-c/comqtt/server.(*Server).processPacket(_, _, {{0x2b4, 0x3, 0x1, 0x0, 0x0}, {0x0, 0x0, 0x0}, ...})
	/home/james/projects/web-mon/server/vendor/github.com/wind-c/comqtt/server/process.go:28 +0x138
github.com/wind-c/comqtt/server/internal/clients.(*Client).Read(0xc0007cc000, 0xc0007f7940)
	/home/james/projects/web-mon/server/vendor/github.com/wind-c/comqtt/server/internal/clients/clients.go:401 +0x19d
github.com/wind-c/comqtt/server.(*Server).EstablishConnection(0xc000140370, {0xa1d4a4, 0x4}, {0xae60f8, 0xc00033e008}, {0xae2260?, 0xe8b340?})
	/home/james/projects/web-mon/server/vendor/github.com/wind-c/comqtt/server/server.go:362 +0x10f2
github.com/wind-c/comqtt/server/listeners.(*TCP).Serve.func1()
	/home/james/projects/web-mon/server/vendor/github.com/wind-c/comqtt/server/listeners/tcp.go:113 +0x3d
created by github.com/wind-c/comqtt/server/listeners.(*TCP).Serve
	/home/james/projects/web-mon/server/vendor/github.com/wind-c/comqtt/server/listeners/tcp.go:112 +0xea

The broker is started with

func startMqttBroker() *mqtt.Server {
	server := mqtt.NewServer(nil)
	tcp := listeners.NewTCP("tcp1", ":1883")
	err := server.AddListener(tcp, nil)
	if err != nil {
		log.Fatal(err)
	}
	go func() {
		err := server.Serve()
		if err != nil {
			log.Fatal(err)
		}
	}()
	return server
}

Seems to be the same issue as #54 and likely has a similar fix.

The auth config in single.yml doesn't align to the config structure.

Not really sure what the cause is. But it fails. :-)

Hello,
Can the default zerolog logger be changed?
or maybe at least disable it?

Thanks.

This seems to test if the retained flag is set. Not sure if the implementation or the test is failing.

Hey,

Hope you are doing well!
Could you please help me to find the best way to create postgres database/tables.
If it possible may you share a postgres schema?!

Thank you in advance,
Alex

Hi,
is there way to queue/retry messages sent to kafka if the broker isn't available?

Benchmark command:
mqtt-stresser -broker tcps://localhost:18883 -skip-tls-verification -num-clients 10 -num-messages 80000 -rampup-delay 1s -rampup-size 10 -global-timeout 180s -timeout 20s

It will panic when it got the "An existing connection was forcibly closed by the remote host" error message.

2022/11/07 18:56:43 error writing to buffer io.Writer; write tcp 127.0.0.1:18883->127.0.0.1:60634: wsasend: An existing connection was forcibly closed by the remote host.
2022/11/07 18:56:43 error writing to buffer io.Writer; write tcp 127.0.0.1:18883->127.0.0.1:60633: wsasend: An existing connection was forcibly closed by the remote host.
panic: runtime error: invalid memory address or nil pointer dereference
[signal 0xc0000005 code=0x0 addr=0x0 pc=0x46d467]

goroutine 39 [running]:
github.com/wind-c/comqtt/server/internal/circ.(*Writer).Write(0x0, {0xc001557b00, 0x4f, 0xc0000b7d28?})
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/internal/circ/writer.go:80 +0x27
github.com/wind-c/comqtt/server/internal/clients.(*Client).WritePacket(, {{0x4d, 0x3, 0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, ...}, ...})
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/internal/clients/clients.go:525 +0x17a
github.com/wind-c/comqtt/server.(*Server).writeClient(, , {{0x4d, 0x3, 0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, ...})
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/server.go:542 +0x5d
github.com/wind-c/comqtt/server.(*Server).publishToSubscribers(, {{0x4d, 0x3, 0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, ...}, ...})
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/server.go:665 +0x794
github.com/wind-c/comqtt/server.(*Server).processPublish(_, , {{0x4d, 0x3, 0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, ...})
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/process.go:149 +0x678
github.com/wind-c/comqtt/server.(*Server).processPacket(, _, {{0x4d, 0x3, 0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, ...})
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/process.go:28 +0x138
github.com/wind-c/comqtt/server/internal/clients.(*Client).Read(0xc0000b0280, 0xc0000b9940)
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/internal/clients/clients.go:401 +0x19d
github.com/wind-c/comqtt/server.(*Server).EstablishConnection(0xc000120370, {0x59cd6b, 0x2}, {0x62af78, 0xc0000a6700}, {0x628a10?, 0x83f3e0?})
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/server.go:362 +0x10f2
github.com/wind-c/comqtt/server/listeners.(*TCP).Serve.func1()
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/listeners/tcp.go:113 +0x3d
created by github.com/wind-c/comqtt/server/listeners.(*TCP).Serve
C:/Users/test/go/pkg/mod/github.com/wind-c/[email protected]/server/listeners/tcp.go:112 +0xea

在main.go里加了server.Events.OnSubscribe
// Add OnSubscribe Event Hook
server.Events.OnSubscribe = func(filter string, cl events.Client, qos byte, isFirst bool) {
fmt.Printf("<< OnSubscribe client subscribed %s: %s %d %t \n", cl.ID, filter, qos, isFirst)
}
客户端订阅主题的时候没有打印出信息?

有例子参考是如何使用subscribe吗? 客户端接收不到订阅的主题.

The tests seems to expect to get a blank packet back, which sounds weird, but I don't know enough about the wire protocol to figure it out.

How to use redis or mysql for plugin authentication? Can you provide code reference?

Hi!

Can we either add the topic "hacktoberfest" or add the PR label "hacktoberfest-accepted" to attract contributions during the global "hacktoberfest" movement?

Seems that at runtime debug hook results in panic due to lack of OnPublishedWithSharedFilters implementation. This happens only when some client publishes

Logs

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x98 pc=0x1a137e8]

goroutine 191 [running]:
github.com/wind-c/comqtt/v2/mqtt/hooks/debug.(*Hook).OnPublishedWithSharedFilters(_, {{{{0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, ...}, ...}, ...}, ...)
        <autogenerated>:1 +0x28
github.com/wind-c/comqtt/v2/mqtt.(*Hooks).OnPublishedWithSharedFilters(_, {{{{0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, ...}, ...}, ...}, ...)
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/hooks.go:558 +0x14c
github.com/wind-c/comqtt/v2/mqtt.(*Server).PublishToSubscribers(_, {{{{0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, ...}, ...}, ...}, ...)
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/server.go:961 +0x2ec
github.com/wind-c/comqtt/v2/mqtt.(*Server).publishToSubscribers(...)
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/server.go:925
github.com/wind-c/comqtt/v2/mqtt.(*Server).processPublish(_, _, {{{{0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, 0x0, ...}, ...}, ...}, ...})
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/server.go:875 +0x7e5
github.com/wind-c/comqtt/v2/mqtt.(*Server).processPacket(_, _, {{{{0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, 0x0, ...}, ...}, ...}, ...})
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/server.go:627 +0x128
github.com/wind-c/comqtt/v2/mqtt.(*Server).receivePacket(_, _, {{{{0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x0, 0x0, ...}, ...}, ...}, ...})
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/server.go:432 +0x58
github.com/wind-c/comqtt/v2/mqtt.(*Client).Read(0xc000ec8280, 0xc0012457c0)
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/clients.go:362 +0x15d
github.com/wind-c/comqtt/v2/mqtt.(*Server).attachClient(0xc000af5860, 0xc000ec8280, {0x1f055d9, 0x3})
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/server.go:387 +0x605
github.com/wind-c/comqtt/v2/mqtt.(*Server).EstablishConnection(0x0?, {0x1f055d9, 0x3}, {0x2298db8?, 0xc001132000?})
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/server.go:327 +0x5e
github.com/wind-c/comqtt/v2/mqtt/listeners.(*TCP).Serve.func1()
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/listeners/tcp.go:84 +0x43
created by github.com/wind-c/comqtt/v2/mqtt/listeners.(*TCP).Serve in goroutine 44
        /home/czort/go/pkg/mod/github.com/wind-c/comqtt/[email protected]/mqtt/listeners/tcp.go:83 +0x128

Hello.

We're considering adding support to persist all MQTT topic matching a certain filter, to Kafka. Not sure if this is relevant for the project. If it is relevant for the project, let me know and we'll work together as we implement it according to your guidelines.

If not, which is perfectly fine, we'll just do it in a private fork.

All the best,

Per.

设置了BufferSize，BufferBlockSize，消息过大时，在下图所在位置报错

客户端Subscribe的时候，报错：write: No valid packet available; 0

部分代码如下：
server := mqtt.New()
tcp := listeners.NewTCP("t1", ":1883")
err := server.AddListener(tcp, &listeners.Config{
Auth: new(auth.Allow),
})
if err != nil {
log.Fatal(err)
}

err = server.AddStore(bolt.New("mochi-test.db", &bbolt.Options{
	Timeout: 500 * time.Millisecond,
}))
if err != nil {
	log.Fatal(err)
}

server.Events.OnSubscribe = func(filter string, cl events.Client, qos byte, isFirst bool) {
	log.Printf("<< OnSubscribe %s: %s %v  %t\n", cl.ID, filter, qos, isFirst)
}

server.Events.OnError = func(cl events.Client, err error) {
	log.Printf("OnError %s:%s\n", cl.ID, err.Error())
}

输出：
2022/09/27 07:16:39 << OnSubscribe mqttx_1: testtopic/26 0 true
2022/09/27 07:16:39 OnError mqttx_1:write: No valid packet available; 0

你好。
想咨询下，
通过 auth 后将一部分用户信息进行存储，保存在当前连接的session下，需要怎么做，其他hook也可以获取到该session

这个是部署3台连成的集群，我只是加了ssl服务，就co-001无端的报错，我是上午部署，都没有连接请求，下午就报这错误。
希望大佬研究下稳定性如何。

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x8c4b6b]

goroutine 1846 [running]:
github.com/wind-c/comqtt/server/internal/clients.(*Client).ReadConnectPacket(_, )
/Users/w/Documents/comqtt-main/server/internal/clients/clients.go:425 +0x6b
github.com/wind-c/comqtt/server.(*Server).readConnectionPacket(, _)
/Users/w/Documents/comqtt-main/server/server.go:382 +0xc8
github.com/wind-c/comqtt/server.(*Server).EstablishConnection(0xc000184840, {0xc60db1, 0x2}, {0xd759a0, 0xc0004be000}, {0xd6ec08?, 0x121b3d8?})
/Users/w/Documents/comqtt-main/server/server.go:305 +0x728
github.com/wind-c/comqtt/server/listeners.(*TCP).Serve.func1()
/Users/w/Documents/comqtt-main/server/listeners/tcp.go:113 +0x3d
created by github.com/wind-c/comqtt/server/listeners.(*TCP).Serve
/Users/w/Documents/comqtt-main/server/listeners/tcp.go:112 +0xea
2022/11/18 18:15:45 [INFO] memberlist: Suspect co-001 has failed, no acks received

server.Events.OnStorage undefined (type events.Events has no field or method OnStorage)
????

1. 项目会长期维护吗
2. 项目和 https://github.com/mochi-co/mqtt readme内数据高度相似是为啥

Mochi | Mosquitto | EMQX | VerneMQ | Mosca
-- | -- | -- | -- | --
SEND Max | 36505 | 30597 | 27202 | 32782 | 30125

3. comqtt 和 mochi-co/mqtt 有什么关联吗

I have been trying to figure out if there is a way to drop messages in the OnPublish hook. Both to drop them silently (ie. let the client believe that the message was published normally) and to drop them with some error state sent back to the client.

windows netpoll

windows net

linux netpoll

windows net

最近在找支持规则引擎的mqtt项目
目前有的是emqx旗下的3款，最近在考虑ekuiper
不过都是基于sql的.逻辑不太清晰
不知道这个项目的规则引擎准备怎么做。
建议用 https://github.com/hyperjumptech/grule-rule-engine 这类 DSL
如果开始规则引擎代码之后。我计划也添砖加瓦一下，贡献一点想法和代码
PS。我还没学GO。等你们开工了，我跟着一边学GO

请问什么时候能支持MQTT Bridge？

Hi,

I'm using your software since a few months and its genius! Thank you for your work.

I experenced some issues, I have two gateways and one is missing from "comqtt:cl", the funny thing is that I still get updates from the missing gateway and the last messages from comqtt are for both gateways:
{
"action":"connect"
"timestamp":1711650090
}
{
"action":"connect"
"timestamp":1711650090
}

We are using this software for a new project and I wanted to ask if there is a way to fund your work?

Best Regards,
René

Hi.
I've been spending some more time in the code base and I have a couple of suggestions. I'm offering to help here if we want to do this:

1. Protect the main branch and have all development go through PRs. The main reason for this is:
2. Have github refuse to merge PRs that have failing tests.

Having failing tests in main for more than a few hours is not a great look and also makes it really hard to work on the codebase.

Before implementing this the tests need to be all green, though. I'll see what I can do, I'm new to the codebase and I'm not always sure what the issue is.

All the best,

Per.

我这里目前部署的是3个节点，单节点消息pub、sub没有任何问题，但是跨节点就会出现pub之后丢消息，消息丢失之后导致多借点之间数据同步阻塞后续消息无法pub同步到其他节点。
节点配置如下
节点1：

节点2：

节点3：

我这里测试使用的是mqtt-stresser作为测试客户端。
测试命令为./build/mqtt-stresser-linux-amd64 -log-level=0 -broker=tcp://127.0.0.1:1887 -broker-sub=tcp://127.0.0.1:1885 -num-clients=1 -num-messages=25
其中-broker-sub 为我自己加的一个配置，专门用于测试夸节点消息同步是否可能会丢消息。
目前测试发现一般情况下-broker-sub 小于20大概率不会有问题，超过20消息必丢。
pub节点日志

sub节点日志

麻烦大佬帮忙看看是否为我的配置存在问题导致的消息丢失

If bridge-way is set to 1, disconnecting an MQTT client causes the following error:

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x18 pc=0xbbee38]

goroutine 51 [running]:
github.com/wind-c/comqtt/v2/plugin/bridge/kafka.(*Bridge).OnDisconnect(0xc000382960, 0xc00011e780, {0x0, 0x0}, 0x68?)
     /app/plugin/bridge/kafka/kafka.go:279 +0x98
github.com/wind-c/comqtt/v2/mqtt.(*Hooks).OnDisconnect(0xc00030c2a0?, 0xd628c1?, {0x0, 0x0}, 0x3?)
     /app/mqtt/hooks.go:265 +0x13c
github.com/wind-c/comqtt/v2/mqtt.(*Server).attachClient(0xc000382190, 0xc00011e780, {0xd54ca3, 0x3})
    /app/mqtt/server.go:381 +0x7c5
github.com/wind-c/comqtt/v2/mqtt.(*Server).EstablishConnection(0x8747ae?, {0xd54ca3, 0x3}, {0xe712a8?, 0xc000012008?})
     /app/mqtt/server.go:311 +0x65
github.com/wind-c/comqtt/v2/mqtt/listeners.(*TCP).Serve.func1()
     /app/mqtt/listeners/tcp.go:84 +0x47
created by github.com/wind-c/comqtt/v2/mqtt/listeners.(*TCP).Serve
     /app/mqtt/listeners/tcp.go:83 +0x145

Not sure if it's because the client has already disconnected or what.

It looks like the test is broken. It seems to expect the ack to contain some very specific data. The test will pass when
replacing properties with:

	properties := packets.Properties{}

which I suspect is correct, but I'm not 100% sure.

https://github.com/mochi-co/mqtt

One last thing. Goleak seems to be in use. This sounds like a good idea, but it doesn't seem to be implemented correctly yet.

I've got no experience with it so I'm not sure if it serves a purpose. From the somewhat sparse documentation of Goleak, it seem the idea is to run some tests and then it'll go over the runtime and check if there are goroutines left.

If this is correct, I suggest we just delete main_test.go here.

Alternatively, I suspect we might need to refactor main() so it'll hand over control after setting up a context to something like main2() which would get the done-channel (or a ctx). Then write a test that takes up the process, send a couple of messages, and after that close the done-channel. Goleak, should than be able to check if we are able to clean up correctly.

Per.

`type Options struct {
// Capabilities defines the server features and behaviour. If you only wish to modify
// several of these values, set them explicitly - e.g.
// server.Options.Capabilities.MaximumClientWritesPending = 16 * 1024
Capabilities *Capabilities

// ClientNetWriteBufferSize specifies the size of the client *bufio.Writer write buffer.
ClientNetWriteBufferSize int

// ClientNetReadBufferSize specifies the size of the client *bufio.Reader read buffer.
ClientNetReadBufferSize int

// Logger specifies a custom configured implementation of zerolog to override
// the servers default logger configuration. If you wish to change the log level,
// of the default logger, you can do so by setting
// 	server := mqtt.New(nil)
// level := new(slog.LevelVar)
// server.Slog = slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
// 	Level: level,
// }))
// level.Set(slog.LevelDebug)
Logger *slog.Logger

// SysTopicResendInterval specifies the interval between $SYS topic updates in seconds.
SysTopicResendInterval int64

// Enable Inline client to allow direct subscribing and publishing from the parent codebase,
// with negligible performance difference (disabled by default to prevent confusion in statistics).
InlineClient bool

}`

配置的结构体没有加上yaml的配置，导致conf.yaml的mqtt.options下一级配置没有正确加载。

你好，我在测试集群功能的时候发现集群的PUBLISH消息无法正常转发。
通过Debug，发现是因为接受端的Broker在处理转发过来的字节流时，无法成功将字节流解析成PUBLISH消息。
问题代码如下

func (a *Agent) processRelayMsg(msg *message.Message) {
···
	case packets.Publish:
		pk := packets.Packet{FixedHeader: packets.FixedHeader{Type: packets.Publish}}
		pk.ProtocolVersion = msg.ProtocolVersion
		pk.Origin = msg.ClientID
		pk.FixedHeader.Decode(msg.Payload[0])                     // Unpack fixedheader.
		if err := pk.PublishDecode(msg.Payload[2:]); err == nil { // Unpack skips fixedheader
			a.mqttServer.PublishToSubscribers(pk)
			OnPublishPacketLog(DirectionInbound, msg.NodeID, msg.ClientID, pk.TopicName, pk.PacketID)
		}
···
}

出现解析失败的原因是因为剩余长度是一个变长字节整数，长度是不固定的。因此不能使用msg.Payload[2:]来获取可变报头和有效载荷。

解决方法就是计算出剩余长度remaining，用len(msg)-remaining求出offset，使用msg.Payload[offset:]来获取可变报头和有效载荷。

以下是我修改后的代码，已自测通过。

func (a *Agent) processRelayMsg(msg *message.Message) {
	switch msg.Type {
	case message.RaftJoin:
		addr := string(msg.Payload)
		err := a.raftPeer.Join(msg.NodeID, addr)
		OnJoinLog(msg.NodeID, addr, "raft join", err)
	case packets.Subscribe, packets.Unsubscribe:
		a.raftPropose(msg)
	case packets.Publish:
		pk := packets.Packet{FixedHeader: packets.FixedHeader{Type: packets.Publish}}
		pk.ProtocolVersion = msg.ProtocolVersion
		pk.Origin = msg.ClientID
		if err := a.ReadFixedHeader(msg.Payload, &pk.FixedHeader); err != nil {
			return
		}
		offset := len(msg.Payload) - pk.FixedHeader.Remaining
		if err := pk.PublishDecode(msg.Payload[offset:]); err == nil { // Unpack skips fixedheader
			a.mqttServer.PublishToSubscribers(pk)
			OnPublishPacketLog(DirectionInbound, msg.NodeID, msg.ClientID, pk.TopicName, pk.PacketID)
		}
	case packets.Connect:
		//If a client is connected to another node, the client's data cached on the node needs to be cleared
		if existing, ok := a.mqttServer.Clients.Get(msg.ClientID); ok {
			// connection notify from other node
			existing.Stop(packets.ErrSessionTakenOver)
			// clean local session and subscriptions
			a.mqttServer.UnsubscribeClient(existing)
			a.mqttServer.Clients.Delete(msg.ClientID)
		}
		OnConnectPacketLog(DirectionInbound, msg.NodeID, msg.ClientID)
	}
}

func (a *Agent) ReadFixedHeader(b []byte, fh *packets.FixedHeader) error {
	err := fh.Decode(b[0])
	if err != nil {
		return err
	}
	fh.Remaining, err = DecodeLength(b[1:])
	if err != nil {
		return err
	}
	return nil
}

func DecodeLength(b []byte) (n int, err error) {
	// see 1.5.5 Variable Byte Integer decode non-normative
	// https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901027
	var multiplier uint32
	var value uint32
	index := 0
	for {
		eb := b[index]
		if err != nil {
			return 0, err
		}

		value |= uint32(eb&127) << multiplier
		if value > 268435455 {
			return 0, nil
		}

		if (eb & 128) == 0 {
			break
		}

		multiplier += 7
		index++
	}

	return int(value), nil
}

In both the single and cluster main() there are web servers started in init() on port 6060.

This complicates thing when testing as there no easy way to shut the server down again. The server responds with a 404 and disabling it doesn't seem to affect the server.

Are you ok with removing this?

Hi there Wind.

Excellent work on the tests. 🥇 I'm very happy about this. We're 99% of the way.

I struggle a bit with that leak test I did. So what I do is run a cluster node with the following arguments:
-node-name testnode -members localhost:7946 -raft-bootstrap true -storage-way memory

This runs cleanly, but if I shut it down (Ctrl-C), the following happens:

12:39PM INF comqtt server started
^C12:39PM WRN caught signal, stopping...
12:39PM INF stopping raft...
12:39PM ??? 2023-08-01T10:39:42.648+0200 [ERROR] raft: failed to take snapshot: error="nothing new to snapshot"
12:39PM WRN failed to create snapshot!
12:39PM INF raft stopped
12:39PM INF stopping node...
12:39PM ??? 2023/08/01 10:39:42 [INFO] serf: EventMemberLeave: testnode 127.0.0.1
12:39PM ERR not graceful stop
12:39PM DBG system event loop halted

I suspect this causes the leaktest, which basically does this, does not to complete correctly, leaving a couple of goroutines hanging. Is the error "nothing new to snapshot" - really an error?

All the best,
Per.

Again, not quite sure what is causing this, but this test fails.

目前就两台外网机器，进行跨机器建立集群

其中一台 conf.yml:

cluster:
node-name: co-009 #The name of this node. This must be unique in the cluster.If nodename is not set, use the local hostname.
bind-port: 1886 #The port is used for both UDP and TCP gossip.Used for member discovery and communication.
members: localhost:1886,172.17.224.50:1886,172.17.224.51:1886 #seeds member list, format such as 192.168.0.103:7946,192.168.0.104:7946
queue-depth: 1024000 #size of Memberlist's internal channel which handles UDP messages.
raft-port: 1887
raft-dir: ./raft/node1

另中一台的conf.yml:

cluster:
node-name: co-001 #The name of this node. This must be unique in the cluster.If nodename is not set, use the local hostname.
bind-port: 1886 #The port is used for both UDP and TCP gossip.Used for member discovery and communication.
members: localhost:1886,172.17.224.50:1886,172.17.224.51:1886 #seeds member list, format such as 192.168.0.103:7946,192.168.0.104:7946
queue-depth: 1024000 #size of Memberlist's internal channel which handles UDP messages.
raft-port: 1887
raft-dir: ./raft/node1

就是我两台机器配置的members，都填一样的ip

报错信息:

A node has joined: co-009
A node has joined: co-001
Local member 172.17.224.50:1886
gogo BootstrapRaft
2022-10-11T10:20:51.812+0800 [INFO] raft: initial configuration: index=1 servers="[{Suffrage:Voter ID:co-009 Address:172.17.224.50:1887}]"
2022-10-11T10:20:51.812+0800 [INFO] raft: entering follower state: follower="Node at 172.17.224.50:1887 [Follower]" leader-address= leader-id=
Cluster Node Created!
Mqtt Server Started!
2022-10-11T10:20:53.139+0800 [WARN] raft: heartbeat timeout reached, starting election: last-leader-addr= last-leader-id=
2022-10-11T10:20:53.140+0800 [INFO] raft: entering candidate state: node="Node at 172.17.224.50:1887 [Candidate]" term=7
2022-10-11T10:20:53.146+0800 [INFO] raft: election won: tally=1
2022-10-11T10:20:53.146+0800 [INFO] raft: entering leader state: leader="Node at 172.17.224.50:1887 [Leader]"
2022/10/11 10:20:53 [INFO] memberlist: Suspect co-001 has failed, no acks received
2022/10/11 10:20:56 [INFO] memberlist: Suspect co-001 has failed, no acks received
2022/10/11 10:20:57 [INFO] memberlist: Marking co-001 as failed, suspect timeout reached (0 peer confirmations)
A node has left: co-001
2022/10/11 10:21:00 [INFO] memberlist: Suspect co-001 has failed, no acks received

有加入joined，就是没有回应，不知道是不是配置文件错了, 看了是内部代码报的错，如果是我的conf.yml配置错了，希望作者给个正确conf.yml配置方式，感谢

Hi.

I'm my quest to make the tests greener I'm currently trying to figure out how http auth plugins work. I'm looking at https://github.com/wind-c/comqtt/blob/main/plugin/auth/http/http.go#L138 and I see that comqtt no longer sends the topic to the auth server.

Is this by design? If so it would be hard to limit a user only to certain topics.

Cheers,

Per.

现在用haproxy做负载均衡，拿到的是内网IP，在haproxy里面配置send-proxy后发起连接会导致comqtt退出

Description

Should add an event interface which named OnConnected, Because that's only have OnConnect , But it seems only suit for todo something BEFORE connect，when we open Auth plugin，'OnConnected' maybe not trigger if auth failure。

Such bellow

type Events struct {
	OnProcessMessage // published message receieved before evaluation.
	OnMessage          // published message receieved.
	OnError                // server error.
	OnConnect           // client ready connect.
        OnConnected        // client connected.
	OnDisconnect       // client disconnected.
	OnSubscribe        // topic subscription created.
	OnUnsubscribe    // topic subscription removed.
}