Repository: nats-io/nats-server (go)

Issue:
Subject mapping doesn't work with MQTT
## Defect

#### Versions of `nats-server` and affected client libraries used:

```
[53] 2022/10/12 12:53:59.709111 [INF] Starting nats-server
[53] 2022/10/12 12:53:59.709136 [INF]   Version:  2.9.1
[53] 2022/10/12 12:53:59.709145 [INF]   Git:      [2363a2c]
[53] 2022/10/12 12:53:59.709151 [DBG]   Go build: go1.19.1
```

#### OS/Container environment:

Ubuntu 22.04.1 LTS

#### Steps or code to reproduce the issue:

I'm trying to use subject mapping with MQTT and the mapping doesn't work.

My nats server is running with this config:
```
listen: 0.0.0.0:4222
http: 0.0.0.0:8222

server_name: "mqtt_server"

jetstream {
    store_dir=/var/lib/nats/data
    max_file=10Gi
}

mappings = {
  foo: bar
}

mqtt {
   port: 1883
}
```

I subscribe on the topic `bar`:
```sh
docker run --rm --net host hivemq/mqtt-cli sub --topic bar -V 3
```

In another terminal I publish a message to `foo`:
```sh
docker run --rm --net host hivemq/mqtt-cli pub --topic foo -m "test" -V 3
```

#### Expected result:

I expect the message "test" to be received on the topic `bar`.

#### Actual result:

The message is not received on `bar`. I can see the message if I subscribe on `foo` instead but it means there is no mapping.


Pull request:
[FIXED] MQTT: Subjects mapping were not handled
A simple configuration like this:
```
...
mappings = {
  foo: bar
}

mqtt {
   port: 1883
}
```
would cause an MQTT subscription on "bar" to not receive messages published on "foo".

In otherwords, the subject transformation was not done when parsing a PUBLISH packet.

This PR also handles the case of service imports where transformation occurs after the initial publish parsing.

Resolves #3547

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>


Patch:
diff --git a/server/client.go b/server/client.go
index a7aa6356..45db8f86 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3927,7 +3927,8 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// Now check to see if this account has mappings that could affect the service import.
 	// Can't use non-locked trick like in processInboundClientMsg, so just call into selectMappedSubject
 	// so we only lock once.
-	if nsubj, changed := si.acc.selectMappedSubject(to); changed {
+	nsubj, changed := si.acc.selectMappedSubject(to)
+	if changed {
 		c.pa.mapped = []byte(to)
 		to = nsubj
 	}
@@ -3983,6 +3984,10 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	}
 	c.pa.reply = nrr
 
+	if changed && c.isMqtt() && c.pa.hdr > 0 {
+		c.srv.mqttStoreQoS1MsgForAccountOnNewSubject(c.pa.hdr, msg, si.acc.GetName(), to)
+	}
+
 	// FIXME(dlc) - Do L1 cache trick like normal client?
 	rr := si.acc.sl.Match(to)
 
diff --git a/server/mqtt.go b/server/mqtt.go
index 8d36130c..6ca25c36 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -347,6 +347,7 @@ type mqttWriter struct {
 type mqttWill struct {
 	topic   []byte
 	subject []byte
+	mapped  []byte
 	message []byte
 	qos     byte
 	retain  bool
@@ -362,6 +363,7 @@ type mqttFilter struct {
 type mqttPublish struct {
 	topic   []byte
 	subject []byte
+	mapped  []byte
 	msg     []byte
 	sz      int
 	pi      uint16
@@ -616,6 +618,7 @@ func (c *client) mqttParse(buf []byte) error {
 			r.reader.SetReadDeadline(time.Time{})
 		}
 	}
+	hasMappings := c.in.flags.isSet(hasMappings)
 	c.mu.Unlock()
 
 	r.reset(buf)
@@ -661,7 +664,7 @@ func (c *client) mqttParse(buf []byte) error {
 		case mqttPacketPub:
 			pp := c.mqtt.pp
 			pp.flags = b & mqttPacketFlagMask
-			err = c.mqttParsePub(r, pl, pp)
+			err = c.mqttParsePub(r, pl, pp, hasMappings)
 			if trace {
 				c.traceInOp("PUBLISH", errOrTrace(err, mqttPubTrace(pp)))
 				if err == nil {
@@ -737,7 +740,7 @@ func (c *client) mqttParse(buf []byte) error {
 			var rc byte
 			var cp *mqttConnectProto
 			var sessp bool
-			rc, cp, err = c.mqttParseConnect(r, pl)
+			rc, cp, err = c.mqttParseConnect(r, pl, hasMappings)
 			// Add the client id to the client's string, regardless of error.
 			// We may still get the client_id if the call above fails somewhere
 			// after parsing the client ID itself.
@@ -874,6 +877,41 @@ func (s *Server) mqttUpdateMaxAckPending(newmaxp uint16) {
 	})
 }
 
+func (s *Server) mqttGetJSAForAccount(acc string) *mqttJSA {
+	sm := &s.mqtt.sessmgr
+
+	sm.mu.RLock()
+	asm := sm.sessions[acc]
+	sm.mu.RUnlock()
+
+	if asm == nil {
+		return nil
+	}
+
+	asm.mu.RLock()
+	jsa := &asm.jsa
+	asm.mu.RUnlock()
+	return jsa
+}
+
+func (s *Server) mqttStoreQoS1MsgForAccountOnNewSubject(hdr int, msg []byte, acc, subject string) {
+	if s == nil || hdr <= 0 {
+		return
+	}
+	nhv := getHeader(mqttNatsHeader, msg[:hdr])
+	if len(nhv) < 1 {
+		return
+	}
+	if qos := nhv[0] - '0'; qos != 1 {
+		return
+	}
+	jsa := s.mqttGetJSAForAccount(acc)
+	if jsa == nil {
+		return
+	}
+	jsa.storeMsg(mqttStreamSubjectPrefix+subject, hdr, msg)
+}
+
 // Returns the MQTT sessions manager for a given account.
 // If new, creates the required JetStream streams/consumers
 // for handling of sessions and messages.
@@ -2464,7 +2502,7 @@ func (sess *mqttSession) deleteConsumer(cc *ConsumerConfig) {
 //////////////////////////////////////////////////////////////////////////////
 
 // Parse the MQTT connect protocol
-func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProto, error) {
+func (c *client) mqttParseConnect(r *mqttReader, pl int, hasMappings bool) (byte, *mqttConnectProto, error) {
 	// Protocol name
 	proto, err := r.readBytes("protocol name", false)
 	if err != nil {
@@ -2582,11 +2620,28 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProt
 		if !utf8.Valid(topic) {
 			return 0, nil, fmt.Errorf("invalid utf8 for Will topic %q", topic)
 		}
-		cp.will.topic = topic
 		// Convert MQTT topic to NATS subject
-		if cp.will.subject, err = mqttTopicToNATSPubSubject(topic); err != nil {
+		cp.will.subject, err = mqttTopicToNATSPubSubject(topic)
+		if err != nil {
 			return 0, nil, err
 		}
+		// Check for subject mapping.
+		if hasMappings {
+			// For selectMappedSubject to work, we need to have c.pa.subject set.
+			// If there is a change, c.pa.mapped will be set after the call.
+			c.pa.subject = cp.will.subject
+			if changed := c.selectMappedSubject(); changed {
+				// We need to keep track of the NATS subject/mapped in the `cp` structure.
+				cp.will.subject = c.pa.subject
+				cp.will.mapped = c.pa.mapped
+				// We also now need to map the original MQTT topic to the new topic
+				// based on the new subject.
+				topic = natsSubjectToMQTTTopic(string(cp.will.subject))
+			}
+			// Reset those now.
+			c.pa.subject, c.pa.mapped = nil, nil
+		}
+		cp.will.topic = topic
 		// Now "will" message.
 		// Ask for a copy since we need to hold to this after parsing of this protocol.
 		cp.will.message, err = r.readBytes("Will message", true)
@@ -2866,6 +2921,7 @@ func (s *Server) mqttHandleWill(c *client) {
 	pp := c.mqtt.pp
 	pp.topic = will.topic
 	pp.subject = will.subject
+	pp.mapped = will.mapped
 	pp.msg = will.message
 	pp.sz = len(will.message)
 	pp.pi = 0
@@ -2884,7 +2940,7 @@ func (s *Server) mqttHandleWill(c *client) {
 //
 //////////////////////////////////////////////////////////////////////////////
 
-func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish) error {
+func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish, hasMappings bool) error {
 	qos := mqttGetQoS(pp.flags)
 	if qos > 1 {
 		return fmt.Errorf("publish QoS=%v not supported", qos)
@@ -2910,6 +2966,23 @@ func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish) error {
 		return err
 	}
 
+	// Check for subject mapping.
+	if hasMappings {
+		// For selectMappedSubject to work, we need to have c.pa.subject set.
+		// If there is a change, c.pa.mapped will be set after the call.
+		c.pa.subject = pp.subject
+		if changed := c.selectMappedSubject(); changed {
+			// We need to keep track of the NATS subject/mapped in the `pp` structure.
+			pp.subject = c.pa.subject
+			pp.mapped = c.pa.mapped
+			// We also now need to map the original MQTT topic to the new topic
+			// based on the new subject.
+			pp.topic = natsSubjectToMQTTTopic(string(pp.subject))
+		}
+		// Reset those now.
+		c.pa.subject, c.pa.mapped = nil, nil
+	}
+
 	if qos > 0 {
 		pp.pi, err = r.readUint16("packet identifier")
 		if err != nil {
@@ -2952,7 +3025,7 @@ func mqttPubTrace(pp *mqttPublish) string {
 // Runs from the client's readLoop.
 // No lock held on entry.
 func (s *Server) mqttProcessPub(c *client, pp *mqttPublish) error {
-	c.pa.subject, c.pa.hdr, c.pa.size, c.pa.reply = pp.subject, -1, pp.sz, nil
+	c.pa.subject, c.pa.mapped, c.pa.hdr, c.pa.size, c.pa.reply = pp.subject, pp.mapped, -1, pp.sz, nil
 
 	bb := bytes.Buffer{}
 	bb.WriteString(hdrLine)
@@ -2974,7 +3047,7 @@ func (s *Server) mqttProcessPub(c *client, pp *mqttPublish) error {
 	if _, permIssue := c.processInboundClientMsg(msgToSend); !permIssue && mqttGetQoS(pp.flags) > 0 {
 		_, err = c.mqtt.sess.jsa.storeMsg(mqttStreamSubjectPrefix+string(c.pa.subject), c.pa.hdr, msgToSend)
 	}
-	c.pa.subject, c.pa.hdr, c.pa.size, c.pa.szb, c.pa.reply = nil, -1, 0, nil, nil
+	c.pa.subject, c.pa.mapped, c.pa.hdr, c.pa.size, c.pa.szb, c.pa.reply = nil, nil, -1, 0, nil, nil
 	return err
 }
 
@@ -3362,6 +3435,10 @@ func mqttDeliverMsgCbQos0(sub *subscription, pc *client, _ *Account, subject, _
 			return
 		}
 		topic = pc.mqtt.pp.topic
+		// Check for service imports where subject mapping is in play.
+		if len(pc.pa.mapped) > 0 && len(pc.pa.psi) > 0 {
+			topic = natsSubjectToMQTTTopic(subject)
+		}
 		retained = mqttIsRetained(pc.mqtt.pp.flags)
 
 	} else {


Test patch:
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index acfa516e..60043e3e 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -1622,7 +1622,7 @@ func TestMQTTParseConnect(t *testing.T) {
 			r.reset(test.proto)
 			mqtt := &mqtt{r: r}
 			c := &client{mqtt: mqtt}
-			if _, _, err := c.mqttParseConnect(r, test.pl); err == nil || !strings.Contains(err.Error(), test.err) {
+			if _, _, err := c.mqttParseConnect(r, test.pl, false); err == nil || !strings.Contains(err.Error(), test.err) {
 				t.Fatalf("Expected error %q, got %v", test.err, err)
 			}
 		})
@@ -2075,7 +2075,7 @@ func TestMQTTParsePub(t *testing.T) {
 			mqtt := &mqtt{r: r}
 			c := &client{mqtt: mqtt}
 			pp := &mqttPublish{flags: test.flags}
-			if err := c.mqttParsePub(r, test.pl, pp); err == nil || !strings.Contains(err.Error(), test.err) {
+			if err := c.mqttParsePub(r, test.pl, pp, false); err == nil || !strings.Contains(err.Error(), test.err) {
 				t.Fatalf("Expected error %q, got %v", test.err, err)
 			}
 		})
@@ -6069,6 +6069,252 @@ func TestMQTTConsumerInactiveThreshold(t *testing.T) {
 	}
 }
 
+func TestMQTTSubjectMapping(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: 127.0.0.1:-1
+		server_name: mqtt
+		jetstream: enabled
+
+		mappings = {
+			foo0: bar0
+			foo1: bar1
+		}
+
+		mqtt {
+			listen: 127.0.0.1:-1
+		}
+	`))
+	defer removeFile(t, conf)
+	s, o := RunServerWithConfig(conf)
+	defer testMQTTShutdownServer(s)
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	for _, qos := range []byte{0, 1} {
+		t.Run(fmt.Sprintf("qos%d", qos), func(t *testing.T) {
+			mc, r := testMQTTConnect(t, &mqttConnInfo{clientID: "sub", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+			defer mc.Close()
+
+			bar := fmt.Sprintf("bar%v", qos)
+			foo := fmt.Sprintf("foo%v", qos)
+
+			testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
+			testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: bar, qos: qos}}, []byte{qos})
+			testMQTTFlush(t, mc, nil, r)
+
+			mcp, rp := testMQTTConnect(t, &mqttConnInfo{clientID: "pub", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+			defer mcp.Close()
+			testMQTTCheckConnAck(t, rp, mqttConnAckRCConnectionAccepted, false)
+
+			natsPub(t, nc, foo, []byte("msg0"))
+			testMQTTCheckPubMsgNoAck(t, mc, r, bar, 0, []byte("msg0"))
+
+			testMQTTPublish(t, mcp, rp, 0, false, false, foo, 0, []byte("msg1"))
+			testMQTTCheckPubMsgNoAck(t, mc, r, bar, 0, []byte("msg1"))
+
+			testMQTTPublish(t, mcp, rp, 0, false, true, foo, 0, []byte("msg1_retained"))
+			testMQTTCheckPubMsgNoAck(t, mc, r, bar, 0, []byte("msg1_retained"))
+
+			testMQTTPublish(t, mcp, rp, 1, false, false, foo, 1, []byte("msg2"))
+			// For the receiving side, the expected QoS is based on the subscription's QoS,
+			// not the publisher.
+			expected := byte(0)
+			if qos == 1 {
+				expected = mqttPubQos1
+			}
+			testMQTTCheckPubMsgNoAck(t, mc, r, bar, expected, []byte("msg2"))
+
+			testMQTTPublish(t, mcp, rp, 1, false, true, foo, 1, []byte("msg2_retained"))
+			testMQTTCheckPubMsgNoAck(t, mc, r, bar, expected, []byte("msg2_retained"))
+
+			testMQTTDisconnect(t, mcp, nil)
+
+			// Try the with the "will" with QoS0 first
+			mcp, rp = testMQTTConnect(t, &mqttConnInfo{
+				clientID:  "pub",
+				cleanSess: true,
+				will:      &mqttWill{topic: []byte(foo), qos: 0, message: []byte("willmsg1")}},
+				o.MQTT.Host, o.MQTT.Port)
+			defer mcp.Close()
+			testMQTTCheckConnAck(t, rp, mqttConnAckRCConnectionAccepted, false)
+			testMQTTFlush(t, mcp, nil, rp)
+
+			// Close the connection without proper disconnect for will to be sent
+			mcp.Close()
+			testMQTTCheckPubMsgNoAck(t, mc, r, bar, 0, []byte("willmsg1"))
+
+			// Try the with the "will" with QoS1 now
+			mcp, rp = testMQTTConnect(t, &mqttConnInfo{
+				clientID:  "pub",
+				cleanSess: true,
+				will:      &mqttWill{topic: []byte(foo), qos: 1, message: []byte("willmsg2")}},
+				o.MQTT.Host, o.MQTT.Port)
+			defer mcp.Close()
+			testMQTTCheckConnAck(t, rp, mqttConnAckRCConnectionAccepted, false)
+			testMQTTFlush(t, mcp, nil, rp)
+
+			// Close the connection without proper disconnect for will to be sent
+			mcp.Close()
+			testMQTTCheckPubMsgNoAck(t, mc, r, bar, expected, []byte("willmsg2"))
+
+			si, err := js.StreamInfo("$MQTT_msgs")
+			require_NoError(t, err)
+			if qos == 0 {
+				require_True(t, si.State.Msgs == 0)
+				require_True(t, si.State.NumSubjects == 0)
+			} else {
+				// Number of QoS1 messages: 1 regular, 1 retained, 1 from will.
+				require_True(t, si.State.Msgs == 3)
+				require_True(t, si.State.NumSubjects == 1)
+			}
+			testMQTTDisconnect(t, mc, nil)
+		})
+	}
+}
+
+func TestMQTTSubjectMappingWithImportExport(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: 127.0.0.1:-1
+		server_name: mqtt
+		jetstream: enabled
+
+		accounts {
+			A {
+				users = [{user: "a", password: "pwd"}]
+				mappings = {
+					bar: foo
+				}
+				exports = [{service: "bar"}]
+				jetstream: enabled
+			}
+			B {
+				users = [{user: "b", password: "pwd"}]
+				mappings = {
+					foo: bar
+				}
+				imports = [{service: {account: "A", subject: "bar"}}]
+				jetstream: enabled
+			}
+			$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+		}
+
+		mqtt {
+			listen: 127.0.0.1:-1
+		}
+	`))
+	defer removeFile(t, conf)
+	s, o := RunServerWithConfig(conf)
+	defer testMQTTShutdownServer(s)
+
+	c1, r1 := testMQTTConnect(t, &mqttConnInfo{user: "a", pass: "pwd", clientID: "a1", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer c1.Close()
+	testMQTTCheckConnAck(t, r1, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, c1, r1, []*mqttFilter{{filter: "foo", qos: 0}}, []byte{0})
+	testMQTTFlush(t, c1, nil, r1)
+
+	c2, r2 := testMQTTConnect(t, &mqttConnInfo{user: "a", pass: "pwd", clientID: "a2", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer c2.Close()
+	testMQTTCheckConnAck(t, r2, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, c2, r2, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+	testMQTTFlush(t, c2, nil, r2)
+
+	c3, r3 := testMQTTConnect(t, &mqttConnInfo{user: "a", pass: "pwd", clientID: "a3", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer c3.Close()
+	testMQTTCheckConnAck(t, r3, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, c3, r3, []*mqttFilter{{filter: "bar", qos: 0}}, []byte{0})
+	testMQTTFlush(t, c3, nil, r3)
+
+	c4, r4 := testMQTTConnect(t, &mqttConnInfo{user: "a", pass: "pwd", clientID: "a4", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer c4.Close()
+	testMQTTCheckConnAck(t, r4, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, c4, r4, []*mqttFilter{{filter: "bar", qos: 1}}, []byte{1})
+	testMQTTFlush(t, c4, nil, r4)
+
+	bc, br := testMQTTConnect(t, &mqttConnInfo{user: "b", pass: "pwd", clientID: "b0", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer bc.Close()
+	testMQTTCheckConnAck(t, br, mqttConnAckRCConnectionAccepted, false)
+
+	bc1, br1 := testMQTTConnect(t, &mqttConnInfo{user: "b", pass: "pwd", clientID: "b1", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer bc1.Close()
+	testMQTTCheckConnAck(t, br1, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, bc1, br1, []*mqttFilter{{filter: "foo", qos: 0}}, []byte{0})
+	testMQTTFlush(t, bc1, nil, br1)
+
+	bc2, br2 := testMQTTConnect(t, &mqttConnInfo{user: "b", pass: "pwd", clientID: "b2", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer bc2.Close()
+	testMQTTCheckConnAck(t, br2, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, bc2, br2, []*mqttFilter{{filter: "b", qos: 1}}, []byte{1})
+	testMQTTFlush(t, bc2, nil, br2)
+
+	bc3, br3 := testMQTTConnect(t, &mqttConnInfo{user: "b", pass: "pwd", clientID: "b3", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer bc3.Close()
+	testMQTTCheckConnAck(t, br3, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, bc3, br3, []*mqttFilter{{filter: "bar", qos: 0}}, []byte{0})
+	testMQTTFlush(t, bc3, nil, br3)
+
+	bc4, br4 := testMQTTConnect(t, &mqttConnInfo{user: "b", pass: "pwd", clientID: "b4", cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+	defer bc4.Close()
+	testMQTTCheckConnAck(t, br4, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, bc4, br4, []*mqttFilter{{filter: "bar", qos: 1}}, []byte{1})
+	testMQTTFlush(t, bc4, nil, br4)
+
+	nc := natsConnect(t, fmt.Sprintf("nats://b:pwd@%s:%d", o.Host, o.Port))
+	defer nc.Close()
+
+	natsPub(t, nc, "foo", []byte("msg0"))
+	testMQTTCheckPubMsgNoAck(t, c1, r1, "foo", 0, []byte("msg0"))
+	testMQTTCheckPubMsgNoAck(t, c2, r2, "foo", 0, []byte("msg0"))
+	testMQTTExpectNothing(t, r3)
+	testMQTTExpectNothing(t, r4)
+	testMQTTExpectNothing(t, br1)
+	testMQTTExpectNothing(t, br2)
+	testMQTTCheckPubMsgNoAck(t, bc3, br3, "bar", 0, []byte("msg0"))
+	testMQTTCheckPubMsgNoAck(t, bc4, br4, "bar", 0, []byte("msg0"))
+
+	testMQTTPublish(t, bc, br, 0, false, false, "foo", 0, []byte("msg1"))
+	testMQTTCheckPubMsgNoAck(t, c1, r1, "foo", 0, []byte("msg1"))
+	testMQTTCheckPubMsgNoAck(t, c2, r2, "foo", 0, []byte("msg1"))
+	testMQTTExpectNothing(t, r3)
+	testMQTTExpectNothing(t, r4)
+	testMQTTExpectNothing(t, br1)
+	testMQTTExpectNothing(t, br2)
+	testMQTTCheckPubMsgNoAck(t, bc3, br3, "bar", 0, []byte("msg1"))
+	testMQTTCheckPubMsgNoAck(t, bc4, br4, "bar", 0, []byte("msg1"))
+
+	testMQTTPublish(t, bc, br, 1, false, false, "foo", 1, []byte("msg2"))
+	testMQTTCheckPubMsgNoAck(t, c1, r1, "foo", 0, []byte("msg2"))
+	testMQTTCheckPubMsgNoAck(t, c2, r2, "foo", mqttPubQos1, []byte("msg2"))
+	testMQTTExpectNothing(t, r3)
+	testMQTTExpectNothing(t, r4)
+	testMQTTExpectNothing(t, br1)
+	testMQTTExpectNothing(t, br2)
+	testMQTTCheckPubMsgNoAck(t, bc3, br3, "bar", 0, []byte("msg2"))
+	testMQTTCheckPubMsgNoAck(t, bc4, br4, "bar", mqttPubQos1, []byte("msg2"))
+
+	// Connection nc is for account B
+	check := func(nc *nats.Conn, subj string) {
+		t.Helper()
+		req, err := json.Marshal(&JSApiStreamInfoRequest{SubjectsFilter: ">"})
+		require_NoError(t, err)
+		resp, err := nc.Request(fmt.Sprintf(JSApiStreamInfoT, "$MQTT_msgs"), req, time.Second)
+		require_NoError(t, err)
+		var si StreamInfo
+		err = json.Unmarshal(resp.Data, &si)
+		require_NoError(t, err)
+		require_True(t, si.State.Msgs == 1)
+		require_True(t, si.State.NumSubjects == 1)
+		require_True(t, si.State.Subjects[subj] == 1)
+	}
+	// Currently, nc is for account B
+	check(nc, "$MQTT.msgs.bar")
+
+	nc.Close()
+	nc = natsConnect(t, fmt.Sprintf("nats://a:pwd@%s:%d", o.Host, o.Port))
+	defer nc.Close()
+	check(nc, "$MQTT.msgs.foo")
+}
+
 //////////////////////////////////////////////////////////////////////////
 //
 // Benchmarks