129 lines
3.3 KiB
Go
129 lines
3.3 KiB
Go
package broker
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.asio.asia/nixevol/NixMsg/internal/app/port"
|
|
"github.com/mochi-mqtt/server/v2/packets"
|
|
)
|
|
|
|
func TestPublishDownRequiresDownSubscription(t *testing.T) {
|
|
b, w, done := startTCPClient(t, "ep-nosub")
|
|
defer func() { _ = b.Close() }()
|
|
defer func() {
|
|
_ = w.Close()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
}
|
|
}()
|
|
writeConnect(t, w, "ep-nosub", 30, 0)
|
|
readExactPacket(t, w, packets.Connack, 3*time.Second)
|
|
waitSession(t, b, "ep-nosub")
|
|
err := b.PublishDown(context.Background(), "ep-nosub", "", []byte(`{"v":1}`), port.PublishOpts{QoS: 0})
|
|
if !errors.Is(err, ErrNotSubscribed) {
|
|
t.Fatalf("err=%v want ErrNotSubscribed", err)
|
|
}
|
|
}
|
|
|
|
func TestPublishDownRejectsStaleConnID(t *testing.T) {
|
|
b, w, done := startTCPClient(t, "ep-stale")
|
|
defer func() { _ = b.Close() }()
|
|
defer func() {
|
|
_ = w.Close()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
}
|
|
}()
|
|
writeConnect(t, w, "ep-stale", 30, 0)
|
|
readExactPacket(t, w, packets.Connack, 3*time.Second)
|
|
writeSubscribe(t, w, downTopic("ep-stale"))
|
|
readExactPacket(t, w, packets.Suback, 3*time.Second)
|
|
waitSession(t, b, "ep-stale")
|
|
err := b.PublishDown(context.Background(), "ep-stale", "dead-conn", []byte(`{"v":1}`), port.PublishOpts{QoS: 0})
|
|
if !errors.Is(err, ErrNoConnection) {
|
|
t.Fatalf("err=%v want ErrNoConnection", err)
|
|
}
|
|
}
|
|
|
|
func TestPublishThenDisconnectWritesThenCloses(t *testing.T) {
|
|
b, w, done := startTCPClient(t, "ep-ptd")
|
|
defer func() { _ = b.Close() }()
|
|
defer func() {
|
|
_ = w.Close()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
}
|
|
}()
|
|
writeConnect(t, w, "ep-ptd", 30, 0)
|
|
readExactPacket(t, w, packets.Connack, 3*time.Second)
|
|
writeSubscribe(t, w, downTopic("ep-ptd"))
|
|
readExactPacket(t, w, packets.Suback, 3*time.Second)
|
|
waitSession(t, b, "ep-ptd")
|
|
|
|
payload := []byte(`{"v":1,"type":"fatal","reason":"disabled"}`)
|
|
if err := b.PublishThenDisconnect(context.Background(), "ep-ptd", "", payload, 1, port.DisconnectFatal); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got := readDownPayload(t, w, 3*time.Second)
|
|
if !bytes.Equal(got, payload) {
|
|
t.Fatalf("got %s", got)
|
|
}
|
|
_ = w.SetReadDeadline(time.Now().Add(3 * time.Second))
|
|
buf := make([]byte, 64)
|
|
n, err := io.ReadAtLeast(w, buf, 2)
|
|
if err != nil && n == 0 {
|
|
return // 连接已关
|
|
}
|
|
if n > 0 && buf[0]>>4 == packets.Disconnect {
|
|
return
|
|
}
|
|
}
|
|
|
|
func TestShutdownUsesServerShuttingDown(t *testing.T) {
|
|
b, w, done := startTCPClient(t, "ep-shut")
|
|
defer func() {
|
|
_ = w.Close()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(3 * time.Second):
|
|
}
|
|
}()
|
|
writeConnect(t, w, "ep-shut", 30, 0)
|
|
readExactPacket(t, w, packets.Connack, 3*time.Second)
|
|
waitSession(t, b, "ep-shut")
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
if err := b.Shutdown(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = w.SetReadDeadline(time.Now().Add(2 * time.Second))
|
|
buf := make([]byte, 32)
|
|
n, err := io.ReadAtLeast(w, buf, 2)
|
|
if err != nil && n == 0 {
|
|
return
|
|
}
|
|
if n > 0 && buf[0]>>4 != packets.Disconnect {
|
|
t.Fatalf("want disconnect got %x", buf[:n])
|
|
}
|
|
}
|
|
|
|
func TestEffectivePayloadLimitSubtractsOverhead(t *testing.T) {
|
|
got := EffectivePayloadLimit(200, 0)
|
|
if got != 200-packetOverheadBudget {
|
|
t.Fatalf("got %d", got)
|
|
}
|
|
got = EffectivePayloadLimit(200, 50)
|
|
if got != 50 {
|
|
t.Fatalf("got %d want 50", got)
|
|
}
|
|
}
|