Skip to content

Commit 6ef2239

Browse files
JavaSaBrJavaSaBr
committed
fix tests
1 parent 271a9f0 commit 6ef2239

File tree

31 files changed

+361
-316
lines changed

31 files changed

+361
-316
lines changed

‎application/src/test/groovy/javasabr/mqtt/broker/application/ConnectSubscribePublishTest.groovy‎

Lines changed: 32 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,14 @@ import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PayloadFormatIndicator
1111
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
1212
import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAckReasonCode
1313

14+
import java.util.concurrent.CompletableFuture
1415
import java.util.concurrent.atomic.AtomicReference
1516

1617
class ConnectSubscribePublishTest extends IntegrationSpecification {
1718

1819
def "should deliver publish message QoS 0 using mqtt 3.1.1"() {
1920
given:
20-
def received = new AtomicReference<Mqtt3Publish>()
21+
def received = new CompletableFuture<Mqtt3Publish>()
2122
def subscriber = buildExternalMqtt311Client()
2223
def subscriberId = subscriber.getConfig().clientIdentifier.get().toString()
2324
def publisher = buildExternalMqtt311Client()
@@ -34,17 +35,17 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
3435
publishResult != null
3536
publishResult.qos == MqttQos.AT_MOST_ONCE
3637
publishResult.type == Mqtt3MessageType.PUBLISH
37-
received.get() != null
38-
received.get().qos == MqttQos.AT_MOST_ONCE
39-
received.get().type == Mqtt3MessageType.PUBLISH
38+
received.join() != null
39+
received.join().qos == MqttQos.AT_MOST_ONCE
40+
received.join().type == Mqtt3MessageType.PUBLISH
4041
cleanup:
4142
subscriber.disconnect().join()
4243
publisher.disconnect().join()
4344
}
4445

4546
def "should deliver publish message QoS 0 using mqtt 5"() {
4647
given:
47-
def received = new AtomicReference<Mqtt5Publish>()
48+
def received = new CompletableFuture<Mqtt5Publish>()
4849
def subscriber = buildExternalMqtt5Client()
4950
def subscriberId = subscriber.getConfig().clientIdentifier.get().toString()
5051
def publisher = buildExternalMqtt5Client()
@@ -61,17 +62,17 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
6162
publishResult != null
6263
publishResult.publish.qos == MqttQos.AT_MOST_ONCE
6364
publishResult.publish.type == Mqtt5MessageType.PUBLISH
64-
received.get() != null
65-
received.get().qos == MqttQos.AT_MOST_ONCE
66-
received.get().type == Mqtt5MessageType.PUBLISH
65+
received.join() != null
66+
received.join().qos == MqttQos.AT_MOST_ONCE
67+
received.join().type == Mqtt5MessageType.PUBLISH
6768
cleanup:
6869
subscriber.disconnect().join()
6970
publisher.disconnect().join()
7071
}
7172

7273
def "should deliver publish message QoS 1 using mqtt 3.1.1"() {
7374
given:
74-
def received = new AtomicReference<Mqtt3Publish>()
75+
def received = new CompletableFuture<Mqtt3Publish>()
7576
def subscriber = buildExternalMqtt311Client()
7677
def subscriberId = subscriber.getConfig().clientIdentifier.get().toString()
7778
def publisher = buildExternalMqtt311Client()
@@ -88,17 +89,17 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
8889
publishResult != null
8990
publishResult.qos == MqttQos.AT_LEAST_ONCE
9091
publishResult.type == Mqtt3MessageType.PUBLISH
91-
received.get() != null
92-
received.get().qos == MqttQos.AT_LEAST_ONCE
93-
received.get().type == Mqtt3MessageType.PUBLISH
92+
received.join() != null
93+
received.join().qos == MqttQos.AT_LEAST_ONCE
94+
received.join().type == Mqtt3MessageType.PUBLISH
9495
cleanup:
9596
subscriber.disconnect().join()
9697
publisher.disconnect().join()
9798
}
9899

99-
def "publisher should publish message QoS 1 using mqtt 5"() {
100+
def "should deliver publish message QoS 1 using mqtt 5"() {
100101
given:
101-
def received = new AtomicReference<Mqtt5Publish>()
102+
def received = new CompletableFuture<Mqtt5Publish>()
102103
def subscriber = buildExternalMqtt5Client()
103104
def subscriberId = subscriber.getConfig().clientIdentifier.get().toString()
104105
def publisher = buildExternalMqtt5Client()
@@ -107,7 +108,6 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
107108
publisher.connect().join()
108109
def subscribeResult = subscribe(subscriber, subscriberId, MqttQos.AT_LEAST_ONCE, received)
109110
def publishResult = publish(publisher, subscriberId, MqttQos.AT_LEAST_ONCE)
110-
Thread.sleep(100)
111111
then:
112112
noExceptionThrown()
113113
subscribeResult != null
@@ -116,17 +116,17 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
116116
publishResult != null
117117
publishResult.publish.qos == MqttQos.AT_LEAST_ONCE
118118
publishResult.publish.type == Mqtt5MessageType.PUBLISH
119-
received.get() != null
120-
received.get().qos == MqttQos.AT_LEAST_ONCE
121-
received.get().type == Mqtt5MessageType.PUBLISH
119+
received.join() != null
120+
received.join().qos == MqttQos.AT_LEAST_ONCE
121+
received.join().type == Mqtt5MessageType.PUBLISH
122122
cleanup:
123123
subscriber.disconnect().join()
124124
publisher.disconnect().join()
125125
}
126126

127-
def "publisher should publish message QoS 2 using mqtt 3.1.1"() {
127+
def "should deliver publish message QoS 2 using mqtt 3.1.1"() {
128128
given:
129-
def received = new AtomicReference<Mqtt3Publish>()
129+
def received = new CompletableFuture<Mqtt3Publish>()
130130
def subscriber = buildExternalMqtt311Client()
131131
def subscriberId = subscriber.getConfig().clientIdentifier.get().toString()
132132
def publisher = buildExternalMqtt311Client()
@@ -135,7 +135,6 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
135135
publisher.connect().join()
136136
def subscribeResult = subscribe(subscriber, subscriberId, MqttQos.EXACTLY_ONCE, received)
137137
def publishResult = publish(publisher, subscriberId, MqttQos.EXACTLY_ONCE)
138-
Thread.sleep(100)
139138
then:
140139
noExceptionThrown()
141140
subscribeResult != null
@@ -144,14 +143,14 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
144143
publishResult != null
145144
publishResult.qos == MqttQos.EXACTLY_ONCE
146145
publishResult.type == Mqtt3MessageType.PUBLISH
147-
received.get() != null
148-
received.get().qos == MqttQos.EXACTLY_ONCE
149-
received.get().type == Mqtt3MessageType.PUBLISH
146+
received.join() != null
147+
received.join().qos == MqttQos.EXACTLY_ONCE
148+
received.join().type == Mqtt3MessageType.PUBLISH
150149
}
151150

152-
def "publisher should publish message QoS 2 using mqtt 5"() {
151+
def "should deliver publish message QoS 2 using mqtt 5"() {
153152
given:
154-
def received = new AtomicReference<Mqtt5Publish>()
153+
def received = new CompletableFuture<Mqtt5Publish>()
155154
def subscriber = buildExternalMqtt5Client()
156155
def subscriberId = subscriber.getConfig().clientIdentifier.get().toString()
157156
def publisher = buildExternalMqtt5Client()
@@ -169,9 +168,9 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
169168
publishResult != null
170169
publishResult.publish.qos == MqttQos.EXACTLY_ONCE
171170
publishResult.publish.type == Mqtt5MessageType.PUBLISH
172-
received.get() != null
173-
received.get().qos == MqttQos.EXACTLY_ONCE
174-
received.get().type == Mqtt5MessageType.PUBLISH
171+
received.join() != null
172+
received.join().qos == MqttQos.EXACTLY_ONCE
173+
received.join().type == Mqtt5MessageType.PUBLISH
175174
cleanup:
176175
subscriber.disconnect().join()
177176
publisher.disconnect().join()
@@ -191,11 +190,11 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
191190
Mqtt5AsyncClient subscriber,
192191
String subscriberId,
193192
MqttQos qos,
194-
AtomicReference<Mqtt5Publish> received) {
193+
CompletableFuture<Mqtt5Publish> received) {
195194
return subscriber.subscribeWith()
196195
.topicFilter("test/$subscriberId")
197196
.qos(qos)
198-
.callback({ publish -> received.set(publish) })
197+
.callback({ publish -> received.complete(publish) })
199198
.send()
200199
.join()
201200
}
@@ -213,11 +212,11 @@ class ConnectSubscribePublishTest extends IntegrationSpecification {
213212
Mqtt3AsyncClient subscriber,
214213
String subscriberId,
215214
MqttQos qos,
216-
AtomicReference<Mqtt3Publish> received) {
215+
CompletableFuture<Mqtt3Publish> received) {
217216
return subscriber.subscribeWith()
218217
.topicFilter("test/$subscriberId")
219218
.qos(qos)
220-
.callback({ publish -> received.set(publish) })
219+
.callback({ publish -> received.complete(publish) })
221220
.send()
222221
.join()
223222
}

‎model/src/main/java/javasabr/mqtt/model/subscribtion/RequestedSubscription.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import javasabr.mqtt.model.QoS;
44
import javasabr.mqtt.model.SubscribeRetainHandling;
5+
import javasabr.mqtt.model.topic.TopicFilter;
56

67
public record RequestedSubscription(
78
/*
@@ -29,4 +30,14 @@ public record RequestedSubscription(
2930
false, Application Messages forwarded using this subscription have the RETAIN flag set to 0. Retained messages sent
3031
when the subscription is established have the RETAIN flag set to 1.
3132
*/
32-
boolean retainAsPublished) {}
33+
boolean retainAsPublished) {
34+
35+
public static RequestedSubscription minimal(String rawTopicFilter, QoS qos) {
36+
return new RequestedSubscription(
37+
rawTopicFilter,
38+
qos,
39+
SubscribeRetainHandling.SEND,
40+
true,
41+
true);
42+
}
43+
}

‎network/src/main/java/javasabr/mqtt/network/message/in/PublishMqttInMessage.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -287,7 +287,7 @@ public class PublishMqttInMessage extends TrackableMqttInMessage {
287287

288288
public PublishMqttInMessage(byte info) {
289289
super(info);
290-
this.qos = QoS.ofCode(info & 0b0110 >> 1);
290+
this.qos = QoS.ofCode((info & 0b0110) >> 1);
291291
this.retained = (info & 0b0001) != 0;
292292
this.duplicate = (info & 0b1000) != 0;
293293
this.rawTopicName = StringUtils.EMPTY;

‎network/src/test/groovy/javasabr/mqtt/network/message/in/PublishAckMqttInMessageTest.groovy‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,15 @@ class PublishAckMqttInMessageTest extends BaseMqttInMessageTest {
1010
def "should read packet correctly as mqtt 3.1.1"() {
1111
given:
1212
def dataBuffer = BufferUtils.prepareBuffer(512) {
13-
it.putShort(packetId)
13+
it.putShort(messageId)
1414
}
1515
when:
1616
def packet = new PublishAckMqttInMessage(0b0100_0000 as byte)
1717
def result = packet.read(defaultMqtt311Connection, dataBuffer, dataBuffer.limit())
1818
then:
1919
result
2020
packet.reason() == ""
21-
packet.messageId() == packetId
21+
packet.messageId() == messageId
2222
packet.reasonCode() == PublishAckReasonCode.SUCCESS
2323
packet.userProperties() == Array.empty()
2424
}
@@ -30,7 +30,7 @@ class PublishAckMqttInMessageTest extends BaseMqttInMessageTest {
3030
it.putProperty(MqttMessageProperty.USER_PROPERTY, userProperties)
3131
}
3232
def dataBuffer = BufferUtils.prepareBuffer(512) {
33-
it.putShort(packetId)
33+
it.putShort(messageId)
3434
it.put(PublishAckReasonCode.PAYLOAD_FORMAT_INVALID.value)
3535
it.putMbi(propertiesBuffer.limit())
3636
it.put(propertiesBuffer)
@@ -41,12 +41,12 @@ class PublishAckMqttInMessageTest extends BaseMqttInMessageTest {
4141
then:
4242
result
4343
packet.reason() == reasonString
44-
packet.messageId() == packetId
44+
packet.messageId() == messageId
4545
packet.reasonCode() == PublishAckReasonCode.PAYLOAD_FORMAT_INVALID
4646
packet.userProperties() == userProperties
4747
when:
4848
dataBuffer = BufferUtils.prepareBuffer(512) {
49-
it.putShort(packetId)
49+
it.putShort(messageId)
5050
it.put(PublishAckReasonCode.UNSPECIFIED_ERROR.value)
5151
it.putMbi(0)
5252
}
@@ -55,7 +55,7 @@ class PublishAckMqttInMessageTest extends BaseMqttInMessageTest {
5555
then:
5656
result
5757
packet.reason() == ""
58-
packet.messageId() == packetId
58+
packet.messageId() == messageId
5959
packet.reasonCode() == PublishAckReasonCode.UNSPECIFIED_ERROR
6060
packet.userProperties() == Array.empty()
6161
}

‎network/src/test/groovy/javasabr/mqtt/network/message/in/PublishCompleteMqttInMessageTest.groovy‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,15 @@ class PublishCompleteMqttInMessageTest extends BaseMqttInMessageTest {
1010
def "should read packet correctly as mqtt 3.1.1"() {
1111
given:
1212
def dataBuffer = BufferUtils.prepareBuffer(512) {
13-
it.putShort(packetId)
13+
it.putShort(messageId)
1414
}
1515
when:
1616
def packet = new PublishCompleteMqttInMessage(0b0111_0000 as byte)
1717
def result = packet.read(defaultMqtt311Connection, dataBuffer, dataBuffer.limit())
1818
then:
1919
result
2020
packet.reason() == ""
21-
packet.messageId() == packetId
21+
packet.messageId() == messageId
2222
packet.reasonCode() == PublishCompletedReasonCode.SUCCESS
2323
packet.userProperties() == Array.empty()
2424
}
@@ -30,7 +30,7 @@ class PublishCompleteMqttInMessageTest extends BaseMqttInMessageTest {
3030
it.putProperty(MqttMessageProperty.USER_PROPERTY, userProperties)
3131
}
3232
def dataBuffer = BufferUtils.prepareBuffer(512) {
33-
it.putShort(packetId)
33+
it.putShort(messageId)
3434
it.put(PublishCompletedReasonCode.PACKET_IDENTIFIER_NOT_FOUND.value)
3535
it.putMbi(propertiesBuffer.limit())
3636
it.put(propertiesBuffer)
@@ -41,12 +41,12 @@ class PublishCompleteMqttInMessageTest extends BaseMqttInMessageTest {
4141
then:
4242
result
4343
packet.reason() == reasonString
44-
packet.messageId() == packetId
44+
packet.messageId() == messageId
4545
packet.reasonCode() == PublishCompletedReasonCode.PACKET_IDENTIFIER_NOT_FOUND
4646
packet.userProperties() == userProperties
4747
when:
4848
dataBuffer = BufferUtils.prepareBuffer(512) {
49-
it.putShort(packetId)
49+
it.putShort(messageId)
5050
it.put(PublishCompletedReasonCode.PACKET_IDENTIFIER_NOT_FOUND.value)
5151
it.putMbi(0)
5252
}
@@ -55,7 +55,7 @@ class PublishCompleteMqttInMessageTest extends BaseMqttInMessageTest {
5555
then:
5656
result
5757
packet.reason() == ""
58-
packet.messageId() == packetId
58+
packet.messageId() == messageId
5959
packet.reasonCode() == PublishCompletedReasonCode.PACKET_IDENTIFIER_NOT_FOUND
6060
packet.userProperties() == Array.empty()
6161
}

0 commit comments

Comments
 (0)