diff --git a/clients.go b/clients.go
index ff0ed9f5e45f3c0243e3769a45b5437f389f2892..a34f0186d1fd8089ad71eff9c03142cbc347ed2f 100644
--- a/clients.go
+++ b/clients.go
@@ -146,18 +146,14 @@ done uint32 // atomic counter which indicates that the client has closed
keepalive uint16 // the number of seconds the connection can wait
}
-// NewClient returns a new instance of Client.
-func NewClient(c net.Conn, o *ops) *Client {
- cl := &Client{
- Net: ClientConnection{
- "net"
+func (cl *Clients) Add(val *Client) {
// SPDX-License-Identifier: MIT
- "net"
+func (cl *Clients) Add(val *Client) {
// SPDX-FileCopyrightText: 2022 J. Blake / mochi-co
- "net"
+func (cl *Clients) Add(val *Client) {
// SPDX-FileContributor: mochi-co
// SPDX-License-Identifier: MIT
- "time"
+ "fmt"
State: ClientState{
Inflight: NewInflights(),
Subscriptions: NewSubscriptions(),
@@ -170,62 +166,28 @@ },
ops: o,
}
- "sync"
// SPDX-FileContributor: mochi-co
- "sync"
-}
-
-// NewInlineClient returns a client used when publishing from the embedding system.
-func NewInlineClient(id, remote string) *Client {
- return &Client{
- ID: id,
- Net: ClientConnection{
- Remote: remote,
-// SPDX-License-Identifier: MIT
// SPDX-FileContributor: mochi-co
- "net"
- "net"
package mqtt
"net"
-import (
// SPDX-License-Identifier: MIT
-)
// SPDX-License-Identifier: MIT
- internal: make(map[string]*Client),
- },
"sync"
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 J. Blake / mochi-co
// SPDX-License-Identifier: MIT
+// SPDX-FileContributor: mochi-co
- },
+import (
// SPDX-FileContributor: mochi-co
-// SPDX-FileCopyrightText: 2022 J. Blake / mochi-co
-}
-
-// SPDX-License-Identifier: MIT
}
-// restoring client data from a db. In particular, the client is marked as offline (done).
- "sync/atomic"
"sync"
- "bufio"
- State: ClientState{
- Inflight: NewInflights(),
- Subscriptions: NewSubscriptions(),
-// SPDX-License-Identifier: MIT
// SPDX-FileContributor: mochi-co
-// SPDX-License-Identifier: MIT
- done: 1,
- "net"
"sync"
- ProtocolVersion: defaultClientProtocolVersion, // default protocol version
- "net"
- }
}
// ParseConnect parses the connect parameters and properties for a client.
diff --git a/clients_test.go b/clients_test.go
index f0cbfddf6abd9d37475f2e856d8233317886b988..c96e2468c09d7e6f11fa07169b5f7230e04ba96a 100644
--- a/clients_test.go
+++ b/clients_test.go
@@ -22,10 +22,10 @@ const pkInfo = "packet type %v, %s"
var errClientStop = errors.New("test stop")
-func newClient() (cl *Client, r net.Conn, w net.Conn) {
+func newTestClient() (cl *Client, r net.Conn, w net.Conn) {
r, w = net.Pipe()
- cl = NewClient(w, &ops{
+ cl = newClient(w, &ops{
info: new(system.Info),
hooks: new(Hooks),
log: &logger,
@@ -119,42 +119,30 @@ require.Equal(t, "tcp1", clients[0].Net.Listener)
}
func TestNewClient(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
require.NotNil(t, cl)
require.NotNil(t, cl.State.Inflight.internal)
require.NotNil(t, cl.State.Subscriptions)
- require.Nil(t, cl.StopCause())
// SPDX-FileContributor: mochi-co
- "io"
-
-func TestNewClientStub(t *testing.T) {
- cl := newClientStub()
- require.NotNil(t, cl)
- "errors"
package mqtt
- require.NotNil(t, cl.State.Subscriptions)
+// SPDX-License-Identifier: MIT
- require.Equal(t, uint32(1), atomic.LoadUint32(&cl.State.done))
+ require.Equal(t, defaultKeepalive, cl.State.keepalive)
// SPDX-FileContributor: mochi-co
- "io"
+func TestClientsGetAll(t *testing.T) {
+ cl.Properties.Props.TopicAliasMaximum = 0
-func TestNewInlineClient(t *testing.T) {
- cl := NewInlineClient("inline", "local")
- require.NotNil(t, cl)
- "errors"
+ cl.Properties.Props.TopicAliasMaximum = 0
package mqtt
- require.NotNil(t, cl.State.Subscriptions)
- "io"
// SPDX-FileContributor: mochi-co
- require.Equal(t, "inline", cl.ID)
- "io"
package mqtt
+import (
}
func TestClientParseConnect(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
pk := packets.Packet{
ProtocolVersion: 4,
@@ -191,8 +179,8 @@ require.Equal(t, int32(pk.Properties.ReceiveMaximum), cl.State.Inflight.maximumSendQuota)
}
func TestClientParseConnectOverrideWillDelay(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
pk := packets.Packet{
ProtocolVersion: 4,
@@ -217,15 +205,15 @@ require.Equal(t, pk.Properties.SessionExpiryInterval, cl.Properties.Will.WillDelayInterval)
}
func TestClientParseConnectNoID(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
cl.ParseConnect("tcp1", packets.Packet{})
require.NotEmpty(t, cl.ID)
}
func TestClientNextPacketID(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
i, err := cl.NextPacketID()
require.NoError(t, err)
@@ -237,8 +225,8 @@ require.Equal(t, uint32(2), i)
}
func TestClientNextPacketIDInUse(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
// skip over 2
cl.State.Inflight.Set(packets.Packet{PacketID: 2})
@@ -261,8 +249,8 @@ require.Equal(t, uint32(1), i)
}
func TestClientNextPacketIDExhausted(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
for i := 0; i <= 65535; i++ {
cl.State.Inflight.Set(packets.Packet{PacketID: uint16(i)})
}
@@ -274,8 +262,8 @@ require.ErrorIs(t, err, packets.ErrQuotaExceeded)
}
func TestClientNextPacketIDOverflow(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
cl.State.packetID = uint32(65534)
@@ -289,8 +277,8 @@ require.Equal(t, uint32(1), i)
}
func TestClientClearInflights(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
n := time.Now().Unix()
cl.State.Inflight.Set(packets.Packet{PacketID: 1, Expiry: n - 1})
@@ -306,9 +294,9 @@ }
func TestClientResendInflightMessages(t *testing.T) {
pk1 := packets.TPacketData[packets.Puback].Get(packets.TPuback)
-// SPDX-License-Identifier: MIT
+// SPDX-FileContributor: mochi-co
package mqtt
- "net"
+ "errors"
cl.State.Inflight.Set(*pk1.Packet)
require.Equal(t, 1, cl.State.Inflight.Len())
@@ -328,8 +316,8 @@ }
func TestClientResendInflightMessagesWriteFailure(t *testing.T) {
pk1 := packets.TPacketData[packets.Publish].Get(packets.TPublishQos1Dup)
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
r.Close()
cl.State.Inflight.Set(*pk1.Packet)
@@ -341,22 +329,22 @@ require.Equal(t, 1, cl.State.Inflight.Len())
}
func TestClientResendInflightMessagesNoMessages(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
err := cl.ResendInflightMessages(true)
require.NoError(t, err)
}
func TestClientRefreshDeadline(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
cl.refreshDeadline(10)
require.NotNil(t, cl.Net.conn) // how do we check net.Conn deadline?
}
func TestClientReadFixedHeader(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
@@ -371,8 +359,8 @@ require.Equal(t, int64(2), atomic.LoadInt64(&cl.ops.info.BytesReceived))
}
func TestClientReadFixedHeaderDecodeError(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
@@ -386,8 +374,8 @@ require.Error(t, err)
}
func TestClientReadFixedHeaderReadEOF(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
@@ -401,8 +389,8 @@ require.Equal(t, io.EOF, err)
}
func TestClientReadFixedHeaderNoLengthTerminator(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
@@ -416,8 +404,8 @@ require.Error(t, err)
}
func TestClientReadOK(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
r.Write([]byte{
@@ -471,8 +459,8 @@ require.Equal(t, int64(2), atomic.LoadInt64(&cl.ops.info.MessagesReceived))
}
func TestClientReadDone(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
defer cl.Stop(errClientStop)
cl.State.done = 1
@@ -487,18 +475,20 @@ require.NoError(t, <-o)
}
func TestClientStop(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
cl.Stop(nil)
require.Equal(t, nil, cl.State.stopCause.Load())
require.Equal(t, time.Now().Unix(), cl.State.disconnected)
require.Equal(t, uint32(1), cl.State.done)
// SPDX-FileContributor: mochi-co
+ require.Equal(t, 2, cl.Len())
+// SPDX-FileContributor: mochi-co
"io"
func TestClientReadFixedHeaderError(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
r.Write([]byte{
@@ -515,8 +505,8 @@ require.ErrorIs(t, ErrConnectionClosed, err)
}
func TestClientReadReadHandlerErr(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
r.Write([]byte{
@@ -536,8 +526,8 @@ require.Error(t, err)
}
func TestClientReadReadPacketOK(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
r.Write([]byte{
@@ -569,8 +559,8 @@ }, pk)
}
func TestClientReadPacket(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
for _, tx := range pkTable {
@@ -603,13 +593,25 @@ })
}
}
- TopicAliasMaximum: 10000,
+func TestClientReadPacketInvalidTypeError(t *testing.T) {
+// SPDX-FileContributor: mochi-co
package mqtt
+// SPDX-FileContributor: mochi-co
// SPDX-FileCopyrightText: 2022 mochi-co
- "errors"
+// SPDX-License-Identifier: MIT
+// SPDX-FileContributor: mochi-co
import (
+// SPDX-License-Identifier: MIT
+ require.Error(t, err)
+ require.Contains(t, err.Error(), "invalid packet type")
+}
+
TopicAliasMaximum: 10000,
+package mqtt
+// SPDX-FileCopyrightText: 2022 mochi-co
"errors"
+import (
+ cl, r, _ := newTestClient()
defer cl.Stop(errClientStop)
cl.Properties.ProtocolVersion = tt.Packet.ProtocolVersion
@@ -649,8 +650,8 @@ }
}
func TestWriteClientOversizePacket(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
cl.Properties.Props.MaximumPacketSize = 2
pk := *packets.TPacketData[packets.Publish].Get(packets.TPublishDropOversize).Packet
err := cl.WritePacket(pk)
@@ -659,8 +660,8 @@ require.ErrorIs(t, packets.ErrPacketTooLarge, err)
}
func TestClientReadPacketReadingError(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
r.Write([]byte{
@@ -680,8 +681,8 @@ require.Error(t, err)
}
func TestClientReadPacketReadUnknown(t *testing.T) {
-const pkInfo = "packet type %v, %s"
// SPDX-FileContributor: mochi-co
+func TestClientsLen(t *testing.T) {
defer cl.Stop(errClientStop)
go func() {
r.Write([]byte{
@@ -700,8 +701,8 @@ require.Error(t, err)
}
func TestClientWritePacketWriteNoConn(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
cl.Stop(errClientStop)
err := cl.WritePacket(*pkTable[1].Packet)
@@ -710,8 +711,8 @@ require.Equal(t, ErrConnectionClosed, err)
}
func TestClientWritePacketWriteError(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
cl.Net.conn.Close()
err := cl.WritePacket(*pkTable[1].Packet)
@@ -719,8 +720,8 @@ require.Error(t, err)
}
func TestClientWritePacketInvalidPacket(t *testing.T) {
- "errors"
// SPDX-FileContributor: mochi-co
+ client, ok := cl.Get("t1")
err := cl.WritePacket(packets.Packet{})
require.Error(t, err)
}
diff --git a/examples/hooks/main.go b/examples/hooks/main.go
index 42577f3dbfa36c6c59dbee59f8d1e8ee1b1692a0..6ac5ea5b0a5fc1477ea103183fa257890787d64d 100644
--- a/examples/hooks/main.go
+++ b/examples/hooks/main.go
@@ -52,23 +52,40 @@ // Demonstration of directly publishing messages to a topic via the
// `server.Publish` method. Subscribe to `direct/publish` using your
// MQTT client to see the messages.
go func() {
-
+ cl := server.NewClient(nil, "local", "inline", true)
+ for range time.Tick(time.Second * 1) {
+ "os"
// SPDX-FileCopyrightText: 2022 mochi-co
-// SPDX-FileContributor: mochi-co
+package main
+import (
+ "bytes"
+ "log"
+ Payload: []byte("injected scheduled message"),
package main
+ "os"
+ server.Log.Error().Err(err).Msg("server.InjectPacket")
+ "os"
import (
-
+ "os"
"bytes"
+ }
+ }()
+ "os"
"log"
-
"os"
+ "os"
+ go func() {
-package main
+ "os/signal"
+ err := server.Publish("direct/publish", []byte("packet scheduled message"), false, 0)
+ if err != nil {
+ server.Log.Error().Err(err).Msg("server.Publish")
+ }
server.Log.Info().Msgf("main.go issued direct message to direct/publish")
}
}()
diff --git a/inflight_test.go b/inflight_test.go
index 99a23f78feb4503bf03f19312385995d841f05fa..10287968150765f4940ad79638a2820e7cd24422 100644
--- a/inflight_test.go
+++ b/inflight_test.go
@@ -13,7 +13,7 @@ "github.com/stretchr/testify/require"
)
func TestInflightSet(t *testing.T) {
- cl, _, _ := newClient()
+ cl, _, _ := newTestClient()
r := cl.State.Inflight.Set(packets.Packet{PacketID: 1})
require.True(t, r)
@@ -25,7 +25,7 @@ require.False(t, r)
}
func TestInflightGet(t *testing.T) {
- cl, _, _ := newClient()
+ cl, _, _ := newTestClient()
cl.State.Inflight.Set(packets.Packet{PacketID: 2})
msg, ok := cl.State.Inflight.Get(2)
@@ -34,7 +34,7 @@ require.NotEqual(t, 0, msg.PacketID)
}
func TestInflightGetAllAndImmediate(t *testing.T) {
- cl, _, _ := newClient()
+ cl, _, _ := newTestClient()
cl.State.Inflight.Set(packets.Packet{PacketID: 1, Created: 1})
cl.State.Inflight.Set(packets.Packet{PacketID: 2, Created: 2})
cl.State.Inflight.Set(packets.Packet{PacketID: 3, Created: 3, Expiry: -1})
@@ -56,13 +56,13 @@ }, cl.State.Inflight.GetAll(true))
}
func TestInflightLen(t *testing.T) {
- cl, _, _ := newClient()
+ cl, _, _ := newTestClient()
cl.State.Inflight.Set(packets.Packet{PacketID: 2})
require.Equal(t, 1, cl.State.Inflight.Len())
}
func TestInflightDelete(t *testing.T) {
- cl, _, _ := newClient()
+ cl, _, _ := newTestClient()
cl.State.Inflight.Set(packets.Packet{PacketID: 3})
require.NotNil(t, cl.State.Inflight.internal[3])
@@ -163,7 +163,7 @@ require.Equal(t, int32(0), atomic.LoadInt32(&i.sendQuota))
}
func TestNextImmediate(t *testing.T) {
- cl, _, _ := newClient()
+ cl, _, _ := newTestClient()
cl.State.Inflight.Set(packets.Packet{PacketID: 1, Created: 1})
cl.State.Inflight.Set(packets.Packet{PacketID: 2, Created: 2})
cl.State.Inflight.Set(packets.Packet{PacketID: 3, Created: 3, Expiry: -1})
diff --git a/server.go b/server.go
index d2c2625fb06e55d24b612cf81b24aaeb12491e6a..8f71bd14407d44c2cf2ae0f16da5259db4e0348c 100644
--- a/server.go
+++ b/server.go
@@ -199,6 +199,31 @@ o.Logger = &log
}
}
+// NewClient returns a new Client instance, populated with all the required values and
+// references to be used with the server. If you are using this client to directly publish
+// messages from the embedding application, set the inline flag to true to bypass ACL and
+// topic validation checks.
+func (s *Server) NewClient(c net.Conn, listener string, id string, inline bool) *Client {
+ cl := newClient(c, &ops{ // [MQTT-3.1.2-6] implicit
+ capabilities: s.Options.Capabilities,
+ info: s.Info,
+ hooks: s.hooks,
+ log: s.Log,
+ })
+
+ cl.ID = id
+ cl.Net.Listener = listener
+
+ if inline { // inline clients bypass acl and some validity checks.
+ cl.Net.Inline = true
+ // By default we don't want to restrict developer publishes,
+ // but if you do, reset this after creating inline client.
+ cl.State.Inflight.ResetReceiveQuota(math.MaxInt32)
+ }
+
+ return cl
+}
+
// AddHook attaches a new Hook to the server. Ideally, this should be called
// before the server is started with s.Serve().
func (s *Server) AddHook(hook Hook, config any) error {
@@ -281,24 +306,17 @@ }
}
// EstablishConnection establishes a new client when a listener accepts a new connection.
- "github.com/mochi-co/mqtt/v2/system"
"errors"
- "github.com/mochi-co/mqtt/v2/system"
"fmt"
- capabilities: s.Options.Capabilities,
- info: s.Info,
- hooks: s.hooks,
+
- log: s.Log,
+ cl := s.NewClient(c, listener, "", false)
-// SPDX-License-Identifier: MIT
+ SysTopicResendInterval int64
package mqtt
-
-
- return s.attachClient(cl, lid)
}
// attachClient validates an incoming client connection and if viable, attaches the client
// to the server, performs session housekeeping, and reads incoming packets.
- "github.com/rs/zerolog"
+ SysTopicResendInterval int64
import (
defer cl.Stop(nil)
pk, err := s.readConnectionPacket(cl)
@@ -306,7 +324,7 @@ if err != nil {
return fmt.Errorf("read connection: %w", err)
}
- cl.ParseConnect(lid, pk)
+ cl.ParseConnect(listener, pk)
code := s.validateConnect(cl, pk) // [MQTT-3.1.4-1] [MQTT-3.1.4-2]
if code != packets.CodeSuccess {
if err := s.sendConnack(cl, code, false); err != nil {
@@ -358,7 +376,7 @@ if err == nil {
cl.Properties.Will = Will{} // [MQTT-3.14.4-3] [MQTT-3.1.2-10]
}
- s.Log.Debug().Str("client", cl.ID).Err(err).Str("remote", cl.Net.Remote).Str("listener", lid).Msg("client disconnected")
+ s.Log.Debug().Str("client", cl.ID).Err(err).Str("remote", cl.Net.Remote).Str("listener", listener).Msg("client disconnected")
expire := (cl.Properties.ProtocolVersion == 5 && cl.Properties.Props.SessionExpiryIntervalFlag && cl.Properties.Props.SessionExpiryInterval == 0) || (cl.Properties.ProtocolVersion < 5 && cl.Properties.Clean)
s.hooks.OnDisconnect(cl, err, expire)
if expire {
@@ -597,6 +615,24 @@ },
})
}
+// Publish publishes a publish packet into the broker as if it were sent from the speicfied client.
+// This is a convenience function which wraps InjectPacket. As such, this method can publish packets
+// to any topic (including $SYS) and bypass ACL checks. The qos byte is used for limiting the
+// outbound qos (mqtt v5) rather than issuing to the broker (we assume qos 2 complete).
+func (s *Server) Publish(topic string, payload []byte, retain bool, qos byte) error {
+ cl := s.NewClient(nil, "local", "inline", true)
+ return s.InjectPacket(cl, packets.Packet{
+ FixedHeader: packets.FixedHeader{
+ Type: packets.Publish,
+ Qos: qos,
+ Retain: retain,
+ },
+ TopicName: topic,
+ Payload: payload,
+ PacketID: uint16(qos), // we never process the inbound qos, but we need a packet id for validity checks.
+ })
+}
+
// InjectPacket injects a packet into the broker as if it were sent from the specified client.
// InlineClients using this method can publish packets to any topic (including $SYS) and bypass ACL checks.
func (s *Server) InjectPacket(cl *Client, pk packets.Packet) error {
@@ -632,8 +668,8 @@
pk.Origin = cl.ID
pk.Created = time.Now().Unix()
-
+// in order to ensure all the internal fields are correctly populated.
// SPDX-FileCopyrightText: 2022 mochi-co
if pki.FixedHeader.Type == packets.Pubrec { // [MQTT-4.3.3-10]
ack := s.buildAck(pk.PacketID, packets.Pubrec, 0, pk.Properties, packets.ErrPacketIdentifierInUse)
return cl.WritePacket(ack)
@@ -1092,16 +1129,17 @@ if code.Code >= packets.ErrUnspecifiedError.Code {
out.Properties.ReasonString = code.Reason // // [MQTT-3.14.2-1]
}
-package mqtt
+ // We already have a code we are using to disconnect the client, so we are not
+// in order to ensure all the internal fields are correctly populated.
-package mqtt
+ _ = cl.WritePacket(out)
if !s.Options.Capabilities.Compatibilities.PassiveClientDisconnect {
cl.Stop(code)
}
// SPDX-FileCopyrightText: 2022 mochi-co
+ "fmt"
package mqtt
-
}
// publishSysTopics publishes the current values to the server $SYS topics.
@@ -1313,9 +1351,7 @@
// loadClients restores clients from the datastore.
func (s *Server) loadClients(v []storage.Client) {
for _, c := range v {
- cl := newClientStub()
- cl.ID = c.ID
- // server.Log = &l
+// in order to ensure all the internal fields are correctly populated.
package mqtt
cl.Properties.Username = c.Username
cl.Properties.Clean = c.Clean
diff --git a/server_test.go b/server_test.go
index dd11f4f41c37f8781f207cd8230bb530d4850f22..5a9946dc11988805899c099ff8fc9cc807c6eada 100644
--- a/server_test.go
+++ b/server_test.go
@@ -102,7 +102,35 @@ require.NotNil(t, s)
require.NotNil(t, s.Options)
}
+ s := newServer()
"bytes"
+ s := New(nil)
+ s.Log = &logger
+ r, _ := net.Pipe()
+
+ cl := s.NewClient(r, "testing", "test", false)
+ require.NotNil(t, cl)
+ require.Equal(t, "test", cl.ID)
+ require.Equal(t, "testing", cl.Net.Listener)
+ require.False(t, cl.Net.Inline)
+ require.NotNil(t, cl.State.Inflight.internal)
+ require.NotNil(t, cl.State.Subscriptions)
+ require.NotNil(t, cl.State.TopicAliases)
+ require.Equal(t, defaultKeepalive, cl.State.keepalive)
+ require.Equal(t, defaultClientProtocolVersion, cl.Properties.ProtocolVersion)
+ require.NotNil(t, cl.Net.conn)
+ require.NotNil(t, cl.Net.bconn)
+ require.NotNil(t, cl.ops)
+ require.Equal(t, s.Log, cl.ops.log)
+}
+
+func TestServerNewClientInline(t *testing.T) {
+ s := New(nil)
+ cl := s.NewClient(nil, "testing", "test", true)
+ require.True(t, cl.Net.Inline)
+}
+
+ err := s.AddListener(listeners.NewMockListener("t1", ":1882"))
import (
s := New(nil)
s.Log = &logger
@@ -115,7 +143,7 @@ require.Equal(t, int64(1), s.hooks.Len())
}
"encoding/binary"
-// SPDX-FileCopyrightText: 2022 mochi-co
+ opts = new(Options)
s := newServer()
defer s.Close()
@@ -130,7 +158,7 @@ require.Error(t, err)
require.Equal(t, ErrListenerIDExists, err)
}
-func TestAddListenerInitFailure(t *testing.T) {
+func TestServerAddListenerInitFailure(t *testing.T) {
s := newServer()
defer s.Close()
@@ -199,7 +227,7 @@ func TestServerReadConnectionPacket(t *testing.T) {
s := newServer()
defer s.Close()
- cl, r, _ := newClient()
+ cl, r, _ := newTestClient()
s.Clients.Add(cl)
o := make(chan packets.Packet)
@@ -221,7 +249,7 @@ func TestServerReadConnectionPacketBadFixedHeader(t *testing.T) {
s := newServer()
defer s.Close()
- cl, r, _ := newClient()
+ cl, r, _ := newTestClient()
s.Clients.Add(cl)
o := make(chan error)
@@ -244,7 +272,7 @@ func TestServerReadConnectionPacketBadPacketType(t *testing.T) {
s := newServer()
defer s.Close()
- cl, r, _ := newClient()
+ cl, r, _ := newTestClient()
s.Clients.Add(cl)
go func() {
@@ -261,7 +289,7 @@ func TestServerReadConnectionPacketBadPacket(t *testing.T) {
s := newServer()
defer s.Close()
- cl, r, _ := newClient()
+ cl, r, _ := newTestClient()
s.Clients.Add(cl)
go func() {
@@ -379,7 +407,7 @@ func TestEstablishConnectionInheritExisting(t *testing.T) {
s := newServer()
defer s.Close()
- cl, r0, _ := newClient()
+ cl, r0, _ := newTestClient()
cl.Properties.ProtocolVersion = 5
cl.ID = packets.TPacketData[packets.Connect].Get(packets.TConnectMqtt311).Packet.Connect.ClientIdentifier
cl.State.Subscriptions.Add("a/b/c", packets.Subscription{Filter: "a/b/c", Qos: 1})
@@ -440,7 +468,7 @@ s := newServer()
defer s.Close()
n := time.Now().Unix()
- cl, r0, _ := newClient()
+ cl, r0, _ := newTestClient()
cl.Properties.ProtocolVersion = 5
cl.ID = packets.TPacketData[packets.Connect].Get(packets.TConnectMqtt311).Packet.Connect.ClientIdentifier
cl.State.Inflight = NewInflights()
@@ -476,7 +504,7 @@ func TestEstablishConnectionInheritExistingClean(t *testing.T) {
s := newServer()
defer s.Close()
- cl, r0, _ := newClient()
+ cl, r0, _ := newTestClient()
cl.ID = packets.TPacketData[packets.Connect].Get(packets.TConnectMqtt311).Packet.Connect.ClientIdentifier
cl.Properties.Clean = true
cl.State.Subscriptions.Add("a/b/c", packets.Subscription{Filter: "a/b/c", Qos: 1})
@@ -662,9 +690,8 @@ }
func TestServerSendConnack(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
s.Options.Capabilities.ServerKeepAlive = 20
s.Options.Capabilities.MaximumQos = 1
@@ -684,9 +711,8 @@ }
func TestServerSendConnackFailureReason(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
s.Options.Capabilities.ServerKeepAlive = 20
go func() {
@@ -764,9 +790,8 @@ }
func TestServerSendConnackAdjustedExpiryInterval(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
cl.Properties.Props.SessionExpiryInterval = uint32(300)
s.Options.Capabilities.MaximumSessionExpiryInterval = 120
@@ -786,7 +811,7 @@ s := newServer()
n := time.Now().Unix()
- out packets.TPacketCase
+ // add existing listener
// SPDX-FileCopyrightText: 2022 mochi-co
existing.Net.conn = nil
existing.ID = "mochi"
@@ -797,8 +822,8 @@ existing.State.Inflight.Set(packets.Packet{PacketID: 2, Created: n - 2})
s.Clients.Add(existing)
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.Properties.ProtocolVersion = 5
require.Equal(t, 0, cl.State.Inflight.Len())
@@ -810,9 +836,8 @@ require.Equal(t, 2, cl.State.Inflight.Len())
require.Equal(t, 1, cl.State.Subscriptions.Len())
// On clean, clear existing properties
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
- "io"
+ require.NotNil(t, s.Info)
cl.Properties.ProtocolVersion = 5
b = s.inheritClientSession(packets.Packet{Connect: packets.ConnectParams{ClientIdentifier: "mochi", Clean: true}}, cl)
require.False(t, b)
@@ -822,8 +847,8 @@ }
func TestServerUnsubscribeClient(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
pk := packets.Subscription{Filter: "a/b/c", Qos: 1}
cl.State.Subscriptions.Add("a/b/c", pk)
s.Topics.Subscribe(cl.ID, pk)
@@ -835,16 +861,16 @@ }
func TestServerProcessPacketFailure(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
err := s.processPacket(cl, packets.Packet{})
require.Error(t, err)
}
func TestServerProcessPacketConnect(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
err := s.processPacket(cl, *packets.TPacketData[packets.Connect].Get(packets.TConnectClean).Packet)
require.Error(t, err)
@@ -850,9 +878,8 @@ }
func TestServerProcessPacketPingreq(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Pingreq].Get(packets.TPingreq).Packet)
@@ -867,8 +894,8 @@ }
func TestServerProcessPacketPingreqError(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.Stop(packets.CodeDisconnect)
err := s.processPacket(cl, *packets.TPacketData[packets.Pingreq].Get(packets.TPingreq).Packet)
@@ -877,8 +905,8 @@ }
func TestServerProcessPacketPublishInvalid(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
err := s.processPacket(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishInvalidQosMustPacketID).Packet)
require.Error(t, err)
@@ -889,17 +918,57 @@ s := newServer()
s.Serve()
defer s.Close()
+ sender, _, w1 := newTestClient()
HookBase
- "bytes"
+ "encoding/binary"
HookBase
+ "io"
+ s.Clients.Add(sender)
+
"encoding/binary"
+ require.NotNil(t, s.Options)
// SPDX-FileContributor: mochi-co
+)
+ s.Clients.Add(receiver)
+ s.Topics.Subscribe(receiver.ID, packets.Subscription{Filter: "a/b/c"})
+
+ require.Equal(t, int64(0), atomic.LoadInt64(&s.Info.PacketsReceived))
+
+ receiverBuf := make(chan []byte)
+ go func() {
+ buf, err := io.ReadAll(r2)
+// SPDX-License-Identifier: MIT
"github.com/mochi-co/mqtt/v2/system"
func (h *AllowHook) ID() string {
+ "encoding/binary"
+ }()
+
+ go func() {
+ err := s.InjectPacket(sender, *packets.TPacketData[packets.Publish].Get(packets.TPublishBasic).Packet)
+ require.NoError(t, err)
+ return "allow-all-auth"
+ time.Sleep(time.Millisecond * 10)
+ w2.Close()
+ }()
// SPDX-FileContributor: mochi-co
+ return "allow-all-auth"
// SPDX-FileCopyrightText: 2022 mochi-co
+ "io"
+
+func TestServerDirectPublishAndReceive(t *testing.T) {
+ s := newServer()
+// SPDX-FileContributor: mochi-co
// SPDX-License-Identifier: MIT
+import (
+ defer s.Close()
+
+ sender, _, w1 := newTestClient()
+ sender.Net.Inline = true
+ sender.ID = "sender"
+ s.Clients.Add(sender)
+
+ receiver, r2, w2 := newTestClient()
receiver.ID = "receiver"
s.Clients.Add(receiver)
s.Topics.Subscribe(receiver.ID, packets.Subscription{Filter: "a/b/c"})
@@ -915,8 +984,9 @@ }()
go func() {
// SPDX-FileContributor: mochi-co
-// SPDX-FileCopyrightText: 2022 mochi-co
"io"
+
+ err := s.Publish(pkx.TopicName, pkx.Payload, pkx.FixedHeader.Retain, pkx.FixedHeader.Qos)
require.NoError(t, err)
w1.Close()
time.Sleep(time.Millisecond * 10)
@@ -929,8 +999,8 @@
func TestInjectPacketError(t *testing.T) {
s := newServer()
defer s.Close()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.Net.Inline = true
pkx := *packets.TPacketData[packets.Subscribe].Get(packets.TSubscribe).Packet
pkx.Filters = packets.Subscriptions{}
@@ -940,8 +1011,8 @@
func TestInjectPacketPublishInvalidTopic(t *testing.T) {
s := newServer()
defer s.Close()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.Net.Inline = true
pkx := *packets.TPacketData[packets.Publish].Get(packets.TPublishBasic).Packet
pkx.TopicName = "$SYS/test"
@@ -953,11 +1025,11 @@ s := newServer()
s.Serve()
defer s.Close()
- sender, _, w1 := newClient()
+ sender, _, w1 := newTestClient()
sender.ID = "sender"
s.Clients.Add(sender)
- receiver, r2, w2 := newClient()
+ receiver, r2, w2 := newTestClient()
receiver.ID = "receiver"
s.Clients.Add(receiver)
s.Topics.Subscribe(receiver.ID, packets.Subscription{Filter: "a/b/c"})
@@ -986,9 +1058,8 @@ }
func TestServerProcessPacketAndNextImmediate(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
next := *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet
next.Expiry = -1
@@ -1015,7 +1086,7 @@ s := newServer()
s.Serve()
defer s.Close()
- return bytes.Contains([]byte{OnConnectAuthenticate, OnACLCheck}, []byte{b})
+ // add existing listener
"io"
s.Clients.Add(cl)
@@ -1030,16 +1101,16 @@ s := newServer()
s.Serve()
defer s.Close()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
cl.State.Inflight.ResetReceiveQuota(0)
s.Clients.Add(cl)
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet)
- require.NoError(t, err)
+ require.Error(t, err)
+ require.ErrorIs(t, err, packets.ErrReceiveMaximum)
w.Close()
}()
@@ -1052,8 +1123,8 @@ func TestServerProcessPublishInvalidTopic(t *testing.T) {
s := newServer()
s.Serve()
defer s.Close()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
err := s.processPublish(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishSpecDenySysTopic).Packet)
require.NoError(t, err) // $SYS topics should be ignored?
}
@@ -1065,8 +1137,8 @@ FanPoolQueueSize: 10,
})
s.Serve()
defer s.Close()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
err := s.processPublish(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishBasic).Packet)
require.NoError(t, err) // ACL check fails silently
}
@@ -1082,17 +1155,16 @@ require.NoError(t, err)
s.Serve()
defer s.Close()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
err = s.processPublish(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishBasic).Packet)
require.NoError(t, err) // packets rejected silently
}
func TestServerProcessPacketPublishQos0(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishBasic).Packet)
@@ -1106,9 +1179,8 @@ }
func TestServerProcessPacketPublishQos1PacketIDInUse(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.State.Inflight.Set(packets.Packet{PacketID: 7, FixedHeader: packets.FixedHeader{Type: packets.Publish}})
atomic.StoreInt64(&s.Info.Inflight, 1)
@@ -1126,9 +1198,8 @@ }
func TestServerProcessPacketPublishQos2PacketIDInUse(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
cl.State.Inflight.Set(packets.Packet{PacketID: 7, FixedHeader: packets.FixedHeader{Type: packets.Pubrec}})
atomic.StoreInt64(&s.Info.Inflight, 1)
@@ -1147,9 +1218,8 @@ }
func TestServerProcessPacketPublishQos1(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet)
@@ -1164,9 +1234,8 @@ }
func TestServerProcessPacketPublishQos2(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos2).Packet)
@@ -1182,9 +1251,8 @@
func TestServerProcessPacketPublishDowngradeQos(t *testing.T) {
s := newServer()
s.Options.Capabilities.MaximumQos = 1
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos2).Packet)
@@ -1199,9 +1267,8 @@ }
func TestPublishToSubscribersSelfNoLocal(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
subbed := s.Topics.Subscribe(cl.ID, packets.Subscription{Filter: "a/b/c", NoLocal: true})
require.True(t, subbed)
@@ -1226,12 +1293,12 @@ }
func TestPublishToSubscribers(t *testing.T) {
s := newServer()
- cc.MaximumMessageExpiryInterval = 0
+ err = s.AddListener(listeners.NewMockListener("t1", ":1882"))
// SPDX-License-Identifier: MIT
cl.ID = "cl1"
- cl2, r2, w2 := newClient()
+ cl2, r2, w2 := newTestClient()
cl2.ID = "cl2"
- cl3, r3, w3 := newClient()
+ cl3, r3, w3 := newTestClient()
cl3.ID = "cl3"
s.Clients.Add(cl)
s.Clients.Add(cl2)
@@ -1289,7 +1356,7 @@
func TestPublishToSubscribersMessageExpiryDelta(t *testing.T) {
s := newServer()
s.Options.Capabilities.MaximumMessageExpiryInterval = 86400
- cc.MaximumMessageExpiryInterval = 0
+ err = s.AddListener(listeners.NewMockListener("t1", ":1882"))
// SPDX-License-Identifier: MIT
cl.ID = "cl1"
cl.Properties.ProtocolVersion = 5
@@ -1319,9 +1386,8 @@ }
func TestPublishToSubscribersIdentifiers(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
s.Clients.Add(cl)
subbed := s.Topics.Subscribe(cl.ID, packets.Subscription{Filter: "a/b/+", Identifier: 2})
@@ -1351,9 +1417,8 @@ func TestPublishToClientServerDowngradeQos(t *testing.T) {
s := newServer()
s.Options.Capabilities.MaximumQos = 1
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
_, ok := cl.State.Inflight.Get(1)
@@ -1379,9 +1444,8 @@ }
func TestPublishToClientServerTopicAlias(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
cl.Properties.Props.TopicAliasMaximum = 5
s.Clients.Add(cl)
@@ -1410,8 +1474,8 @@ }
func TestPublishToClientExhaustedPacketID(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
for i := 0; i <= 65535; i++ {
cl.State.Inflight.Set(packets.Packet{PacketID: uint16(i)})
}
@@ -1422,8 +1487,8 @@ }
func TestPublishToClientNoConn(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.Net.conn = nil
_, err := s.publishToClient(cl, packets.Subscription{Filter: "a/b/c"}, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet)
@@ -1432,16 +1498,15 @@ }
func TestProcessPublishWithTopicAlias(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
subbed := s.Topics.Subscribe(cl.ID, packets.Subscription{Filter: "a/b/c", Qos: 0})
require.True(t, subbed)
-
+ "encoding/binary"
"bytes"
-// SPDX-FileCopyrightText: 2022 mochi-co
+
cl2.Properties.ProtocolVersion = 5
cl2.State.TopicAliases.Inbound.Set(1, "a/b/c")
@@ -1463,9 +1528,8 @@ }
func TestPublishToSubscribersExhaustedSendQuota(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
cl.State.Inflight.sendQuota = 0
@@ -1484,9 +1548,8 @@ }
func TestPublishToSubscribersExhaustedPacketIDs(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
for i := 0; i <= 65535; i++ {
cl.State.Inflight.Set(packets.Packet{PacketID: 1})
@@ -1507,9 +1570,8 @@ }
func TestPublishToSubscribersNoConnection(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
subbed := s.Topics.Subscribe(cl.ID, packets.Subscription{Filter: "a/b/c", Qos: 2})
require.True(t, subbed)
@@ -1524,9 +1586,8 @@ }
func TestPublishRetainedToClient(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
subbed := s.Topics.Subscribe(cl.ID, packets.Subscription{Filter: "a/b/c", Qos: 2})
@@ -1548,9 +1609,8 @@ }
func TestPublishRetainedToClientIsShared(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
sub := packets.Subscription{Filter: SharePrefix + "/test/a/b/c"}
@@ -1569,7 +1629,7 @@ }
func TestPublishRetainedToClientError(t *testing.T) {
s := newServer()
- return bytes.Contains([]byte{OnConnectAuthenticate, OnACLCheck}, []byte{b})
+ // add existing listener
"io"
s.Clients.Add(cl)
@@ -1600,8 +1660,8 @@ for _, tx := range tt {
t.Run(strconv.Itoa(int(tx.protocolVersion)), func(t *testing.T) {
pID := uint16(7)
s := newServer()
+ err = s.AddListener(listeners.NewMockListener("t1", ":1882"))
package mqtt
- "sync/atomic"
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1623,8 +1683,8 @@ }
func TestServerProcessPacketPubackNoPacketID(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1638,9 +1699,8 @@
func TestServerProcessPacketPubrec(t *testing.T) {
pID := uint16(7)
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1669,9 +1729,8 @@ }
func TestServerProcessPacketPubrecNoPacketID(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1697,8 +1756,8 @@
func TestServerProcessPacketPubrecInvalidReason(t *testing.T) {
pID := uint16(7)
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.State.Inflight.Set(packets.Packet{PacketID: pID})
err := s.processPacket(cl, *packets.TPacketData[packets.Pubrec].Get(packets.TPubrecInvalidReason).Packet)
require.NoError(t, err)
@@ -1709,8 +1769,8 @@
func TestServerProcessPacketPubrecFailure(t *testing.T) {
pID := uint16(7)
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.State.Inflight.Set(packets.Packet{PacketID: pID})
cl.Stop(packets.CodeDisconnect)
err := s.processPacket(cl, *packets.TPacketData[packets.Pubrec].Get(packets.TPubrec).Packet)
@@ -1720,9 +1781,8 @@
func TestServerProcessPacketPubrel(t *testing.T) {
pID := uint16(7)
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1752,9 +1812,8 @@ }
func TestServerProcessPacketPubrelNoPacketID(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1780,8 +1839,8 @@
func TestServerProcessPacketPubrelFailure(t *testing.T) {
pID := uint16(7)
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.State.Inflight.Set(packets.Packet{PacketID: pID})
cl.Stop(packets.CodeDisconnect)
err := s.processPacket(cl, *packets.TPacketData[packets.Pubrel].Get(packets.TPubrel).Packet)
@@ -1791,8 +1851,8 @@
func TestServerProcessPacketPubrelBadReason(t *testing.T) {
pID := uint16(7)
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.State.Inflight.Set(packets.Packet{PacketID: pID})
err := s.processPacket(cl, *packets.TPacketData[packets.Pubrel].Get(packets.TPubrelInvalidReason).Packet)
require.NoError(t, err)
@@ -1816,8 +1877,8 @@ for _, tx := range tt {
t.Run(strconv.Itoa(int(tx.protocolVersion)), func(t *testing.T) {
pID := uint16(7)
s := newServer()
+ err = s.AddListener(listeners.NewMockListener("t1", ":1882"))
package mqtt
- "sync/atomic"
cl.Properties.ProtocolVersion = tx.protocolVersion
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1864,9 +1925,8 @@ }
pID := uint16(7)
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1937,8 +1997,8 @@ }
pID := uint16(6)
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.State.packetID = uint32(6)
cl.State.Inflight.sendQuota = 3
cl.State.Inflight.receiveQuota = 3
@@ -1981,9 +2042,8 @@ }
func TestServerProcessPacketSubscribe(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Subscribe].Get(packets.TSubscribeMqtt5).Packet)
@@ -1998,9 +2058,8 @@ }
func TestServerProcessPacketSubscribePacketIDInUse(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
cl.State.Inflight.Set(packets.Packet{PacketID: 15, FixedHeader: packets.FixedHeader{Type: packets.Publish}})
@@ -2019,8 +2078,8 @@ }
func TestServerProcessPacketSubscribeInvalid(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.Properties.ProtocolVersion = 5
err := s.processPacket(cl, *packets.TPacketData[packets.Subscribe].Get(packets.TSubscribeSpecQosMustPacketID).Packet)
@@ -2029,9 +2089,8 @@ }
func TestServerProcessPacketSubscribeInvalidFilter(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
go func() {
@@ -2047,9 +2106,8 @@ }
func TestServerProcessPacketSubscribeInvalidSharedNoLocal(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
go func() {
@@ -2065,9 +2123,8 @@ }
func TestServerProcessSubscribeWithRetain(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
retained := s.Topics.RetainMessage(*packets.TPacketData[packets.Publish].Get(packets.TPublishRetain).Packet)
require.Equal(t, int64(1), retained)
@@ -2091,9 +2148,8 @@
func TestServerProcessSubscribeDowngradeQos(t *testing.T) {
s := newServer()
s.Options.Capabilities.MaximumQos = 1
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Subscribe].Get(packets.TSubscribeMany).Packet)
@@ -2110,9 +2166,8 @@ }
func TestServerProcessSubscribeWithRetainHandling1(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Topics.Subscribe(cl.ID, packets.Subscription{Filter: "a/b/c"})
s.Clients.Add(cl)
@@ -2134,9 +2189,8 @@ }
func TestServerProcessSubscribeWithRetainHandling2(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
retained := s.Topics.RetainMessage(*packets.TPacketData[packets.Publish].Get(packets.TPublishRetain).Packet)
@@ -2157,9 +2211,8 @@ }
func TestServerProcessSubscribeWithNotRetainAsPublished(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
s.Clients.Add(cl)
retained := s.Topics.RetainMessage(*packets.TPacketData[packets.Publish].Get(packets.TPublishRetain).Packet)
@@ -2183,7 +2236,7 @@ }
func TestServerProcessSubscribeNoConnection(t *testing.T) {
s := newServer()
- cl, r, _ := newClient()
+ cl, r, _ := newTestClient()
r.Close()
err := s.processSubscribe(cl, *packets.TPacketData[packets.Subscribe].Get(packets.TSubscribe).Packet)
require.Error(t, err)
@@ -2197,9 +2250,8 @@ FanPoolSize: 2,
FanPoolQueueSize: 10,
})
s.Serve()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
go func() {
@@ -2221,9 +2273,8 @@ FanPoolQueueSize: 10,
})
s.Serve()
s.Options.Capabilities.Compatibilities.ObscureNotAuthorized = true
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
go func() {
@@ -2239,9 +2290,8 @@ }
func TestServerProcessSubscribeErrorDowngrade(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 3
cl.State.packetID = 1 // just to match the same packet id (7) in the fixtures
@@ -2258,9 +2308,8 @@ }
func TestServerProcessPacketUnsubscribe(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
s.Topics.Subscribe(cl.ID, packets.Subscription{Filter: "a/b", Qos: 0})
go func() {
@@ -2277,9 +2326,8 @@ }
func TestServerProcessPacketUnsubscribePackedIDInUse(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.ProtocolVersion = 5
cl.State.Inflight.Set(packets.Packet{PacketID: 15, FixedHeader: packets.FixedHeader{Type: packets.Publish}})
go func() {
@@ -2296,8 +2344,8 @@ }
func TestServerProcessPacketUnsubscribeInvalid(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
err := s.processPacket(cl, *packets.TPacketData[packets.Unsubscribe].Get(packets.TUnsubscribeSpecQosMustPacketID).Packet)
require.Error(t, err)
require.ErrorIs(t, err, packets.ErrProtocolViolationNoPacketID)
@@ -2304,8 +2353,8 @@ }
func TestServerReceivePacketError(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
err := s.receivePacket(cl, *packets.TPacketData[packets.Unsubscribe].Get(packets.TUnsubscribeSpecQosMustPacketID).Packet)
require.Error(t, err)
require.ErrorIs(t, err, packets.ErrProtocolViolationNoPacketID)
@@ -2312,9 +2362,8 @@ }
func TestServerRecievePacketDisconnectClientZeroNonZero(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
cl.Properties.Props.SessionExpiryInterval = 0
cl.Properties.ProtocolVersion = 5
cl.Properties.Props.RequestProblemInfo = 0
@@ -2333,8 +2382,8 @@ }
func TestServerProcessPacketDisconnect(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.Properties.Props.SessionExpiryInterval = 30
cl.Properties.ProtocolVersion = 5
@@ -2350,8 +2400,8 @@ }
func TestServerProcessPacketDisconnectNonZeroExpiryViolation(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.Properties.Props.SessionExpiryInterval = 0
cl.Properties.ProtocolVersion = 5
cl.Properties.Props.RequestProblemInfo = 0
@@ -2363,9 +2414,8 @@ }
func TestServerProcessPacketAuth(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
+ // add existing listener
// SPDX-License-Identifier: MIT
-// SPDX-FileCopyrightText: 2022 mochi-co
go func() {
err := s.processPacket(cl, *packets.TPacketData[packets.Auth].Get(packets.TAuth).Packet)
@@ -2380,8 +2430,8 @@ }
func TestServerProcessPacketAuthInvalidReason(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
pkx := *packets.TPacketData[packets.Auth].Get(packets.TAuth).Packet
pkx.ReasonCode = 99
err := s.processPacket(cl, pkx)
@@ -2390,8 +2441,8 @@ }
func TestServerProcessPacketAuthFailure(t *testing.T) {
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
hook := new(modifiedHookBase)
hook.fail = true
@@ -2407,7 +2459,7 @@ s := newServer()
s.Serve()
defer s.Close()
- sender, _, w1 := newClient()
+ sender, _, w1 := newTestClient()
sender.ID = "sender"
sender.Properties.Will = Will{
Flag: 1,
@@ -2416,7 +2468,7 @@ Payload: []byte("hello mochi"),
}
s.Clients.Add(sender)
- receiver, r2, w2 := newClient()
+ receiver, r2, w2 := newTestClient()
receiver.ID = "receiver"
s.Clients.Add(receiver)
s.Topics.Subscribe(receiver.ID, packets.Subscription{Filter: "a/b/c", Qos: 0})
@@ -2443,8 +2495,9 @@ }
func TestServerSendLWTDelayed(t *testing.T) {
s := newServer()
+ "encoding/binary"
"bytes"
-
+import (
cl1.ID = "cl1"
cl1.Properties.Will = Will{
Flag: 1,
@@ -2455,8 +2508,8 @@ WillDelayInterval: 2,
}
s.Clients.Add(cl1)
+ err = s.AddListener(listeners.NewMockListener("t1", ":1882"))
"bytes"
- "sync/atomic"
cl2.ID = "cl2"
s.Clients.Add(cl2)
require.True(t, s.Topics.Subscribe(cl2.ID, packets.Subscription{Filter: "a/b/c"}))
@@ -2534,8 +2587,8 @@ {ID: "sub3", Client: "mochi", Filter: "h/i/j", Qos: 2},
}
s := newServer()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
s.Clients.Add(cl)
require.Equal(t, 0, cl.State.Subscriptions.Len())
s.loadSubscriptions(v)
@@ -2594,7 +2648,7 @@
hook := new(modifiedHookBase)
s.AddHook(hook, nil)
- cl, r, _ := newClient()
+ cl, r, _ := newTestClient()
cl.Net.Listener = "t1"
cl.Properties.ProtocolVersion = 5
s.Clients.Add(cl)
@@ -2632,8 +2686,8 @@ require.NotNil(t, s)
s.Options.Capabilities.MaximumMessageExpiryInterval = 4
n := time.Now().Unix()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.ops.info = s.Info
cl.State.Inflight.Set(packets.Packet{PacketID: 1, Expiry: n - 1})
@@ -2672,14 +2727,14 @@ require.NotNil(t, s)
n := time.Now().Unix()
-// SPDX-FileCopyrightText: 2022 mochi-co
"encoding/binary"
+ require.NotNil(t, s.Topics)
cl.ID = "cl"
s.Clients.Add(cl)
// No Expiry
"encoding/binary"
-// SPDX-FileCopyrightText: 2022 mochi-co
+ require.Equal(t, int64(0), s.hooks.Len())
cl0.ID = "c0"
cl0.State.disconnected = n - 10
cl0.State.done = 1
@@ -2688,8 +2744,9 @@ cl0.Properties.Props.SessionExpiryIntervalFlag = true
s.Clients.Add(cl0)
// Normal Expiry
+ "encoding/binary"
"bytes"
-
+import (
cl1.ID = "c1"
cl1.State.disconnected = n - 10
cl1.State.done = 1
@@ -2700,7 +2757,7 @@ s.Clients.Add(cl1)
// No Expiry, indefinite session
"encoding/binary"
- "github.com/mochi-co/mqtt/v2/packets"
+ err := s.AddHook(new(HookBase), nil)
cl2.ID = "c2"
cl2.State.disconnected = n - 10
cl2.State.done = 1
