Skip to content

Commit 1165c33

Browse files
sonnypsonnyp
authored
stream-management: Split bind2, sasl2 and stream features (#1063)
1 parent dba0d4a commit 1165c33

File tree

7 files changed

+168
-141
lines changed

7 files changed

+168
-141
lines changed

‎packages/stream-management/bind2.js

+19
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
import { NS, makeEnableElement } from "./index.js";
2+
3+
export function setupBind2({ bind2, sm, failed, enabled }) {
4+
bind2.use(
5+
NS,
6+
// https://xmpp.org/extensions/xep-0198.html#inline-examples
7+
(_element) => {
8+
return makeEnableElement({ sm });
9+
},
10+
async (element) => {
11+
if (element.is("enabled")) {
12+
enabled(element.attrs);
13+
} else if (element.is("failed")) {
14+
// const error = StreamError.fromElement(element)
15+
failed();
16+
}
17+
},
18+
);
19+
}

‎packages/stream-management/index.js

+9-113
Original file line numberDiff line numberDiff line change
@@ -1,44 +1,26 @@
1-
import XMPPError from "@xmpp/error";
2-
import { EventEmitter, procedure } from "@xmpp/events";
1+
import { EventEmitter } from "@xmpp/events";
32
import xml from "@xmpp/xml";
43
import { datetime } from "@xmpp/time";
4+
import { setupBind2 } from "./bind2.js";
5+
import { setupSasl2 } from "./sasl2.js";
6+
import { setupStreamFeature } from "./stream-feature.js";
57

68
// https://xmpp.org/extensions/xep-0198.html
79

8-
const NS = "urn:xmpp:sm:3";
10+
export const NS = "urn:xmpp:sm:3";
911

10-
function makeEnableElement({ sm }) {
12+
export function makeEnableElement({ sm }) {
1113
return xml("enable", {
1214
xmlns: NS,
1315
max: sm.preferredMaximum,
14-
resume: sm.allowResume ? "true" : undefined,
16+
resume: "true",
1517
});
1618
}
1719

18-
function makeResumeElement({ sm }) {
20+
export function makeResumeElement({ sm }) {
1921
return xml("resume", { xmlns: NS, h: sm.inbound, previd: sm.id });
2022
}
2123

22-
function enable(entity, sm) {
23-
return procedure(entity, makeEnableElement({ sm }), (element, done) => {
24-
if (element.is("enabled", NS)) {
25-
return done(element);
26-
} else if (element.is("failed", NS)) {
27-
throw XMPPError.fromElement(element);
28-
}
29-
});
30-
}
31-
32-
async function resume(entity, sm) {
33-
return procedure(entity, makeResumeElement({ sm }), (element, done) => {
34-
if (element.is("resumed", NS)) {
35-
return done(element);
36-
} else if (element.is("failed", NS)) {
37-
throw XMPPError.fromElement(element);
38-
}
39-
});
40-
}
41-
4224
export default function streamManagement({
4325
streamFeatures,
4426
entity,
@@ -51,7 +33,6 @@ export default function streamManagement({
5133

5234
const sm = new EventEmitter();
5335
Object.assign(sm, {
54-
allowResume: true,
5536
preferredMaximum: null,
5637
enabled: false,
5738
id: "",
@@ -171,6 +152,7 @@ export default function streamManagement({
171152
clearTimeout(requestAckTimeout);
172153

173154
if (!sm.enabled) return;
155+
if (!timeout) return;
174156

175157
requestAckTimeout = setTimeout(requestAck, timeout);
176158
}
@@ -218,92 +200,6 @@ export default function streamManagement({
218200
return sm;
219201
}
220202

221-
function setupStreamFeature({
222-
streamFeatures,
223-
sm,
224-
entity,
225-
resumed,
226-
failed,
227-
enabled,
228-
}) {
229-
// https://xmpp.org/extensions/xep-0198.html#enable
230-
// For client-to-server connections, the client MUST NOT attempt to enable stream management until after it has completed Resource Binding unless it is resuming a previous session
231-
streamFeatures.use("sm", NS, async (context, next) => {
232-
// Resuming
233-
if (sm.id) {
234-
try {
235-
const element = await resume(entity, sm);
236-
await resumed(element);
237-
return;
238-
// If resumption fails, continue with session establishment
239-
} catch {
240-
failed();
241-
}
242-
}
243-
244-
// Enabling
245-
246-
// Resource binding first
247-
await next();
248-
249-
const promiseEnable = enable(entity, sm);
250-
251-
if (sm.outbound_q.length > 0) {
252-
throw new Error(
253-
"Stream Management assertion failure, queue should be empty after enable",
254-
);
255-
}
256-
257-
// > The counter for an entity's own sent stanzas is set to zero and started after sending either <enable/> or <enabled/>.
258-
sm.outbound = 0;
259-
260-
try {
261-
const response = await promiseEnable;
262-
enabled(response.attrs);
263-
} catch {
264-
sm.enabled = false;
265-
}
266-
267-
sm.inbound = 0;
268-
});
269-
}
270-
271-
function setupSasl2({ sasl2, sm, failed, resumed }) {
272-
sasl2.use(
273-
"urn:xmpp:sm:3",
274-
(element) => {
275-
if (!element.is("sm")) return;
276-
if (sm.id) return makeResumeElement({ sm });
277-
},
278-
(element) => {
279-
if (element.is("resumed")) {
280-
resumed(element);
281-
} else if (element.is(failed)) {
282-
// const error = StreamError.fromElement(element)
283-
failed();
284-
}
285-
},
286-
);
287-
}
288-
289-
function setupBind2({ bind2, sm, failed, enabled }) {
290-
bind2.use(
291-
"urn:xmpp:sm:3",
292-
// https://xmpp.org/extensions/xep-0198.html#inline-examples
293-
(_element) => {
294-
return makeEnableElement({ sm });
295-
},
296-
(element) => {
297-
if (element.is("enabled")) {
298-
enabled(element.attrs);
299-
} else if (element.is("failed")) {
300-
// const error = StreamError.fromElement(element)
301-
failed();
302-
}
303-
},
304-
);
305-
}
306-
307203
function queueToStanza({ entity, item }) {
308204
const { stanza, stamp } = item;
309205
if (

‎packages/stream-management/sasl2.js

+19
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
import { NS, makeResumeElement } from "./index.js";
2+
3+
export function setupSasl2({ sasl2, sm, failed, resumed }) {
4+
sasl2.use(
5+
NS,
6+
(element) => {
7+
if (!element.is("sm")) return;
8+
if (sm.id) return makeResumeElement({ sm });
9+
},
10+
(element) => {
11+
if (element.is("resumed")) {
12+
resumed(element);
13+
} else if (element.is(failed)) {
14+
// const error = StreamError.fromElement(element)
15+
failed();
16+
}
17+
},
18+
);
19+
}

‎packages/stream-management/stream-feature.js

+74
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
import XMPPError from "@xmpp/error";
2+
import { procedure } from "@xmpp/events";
3+
4+
import { NS, makeEnableElement, makeResumeElement } from "./index.js";
5+
6+
export function setupStreamFeature({
7+
streamFeatures,
8+
sm,
9+
entity,
10+
resumed,
11+
failed,
12+
enabled,
13+
}) {
14+
// https://xmpp.org/extensions/xep-0198.html#enable
15+
// For client-to-server connections, the client MUST NOT attempt to enable stream management until after it has completed Resource Binding unless it is resuming a previous session
16+
streamFeatures.use("sm", NS, async (context, next) => {
17+
// Resuming
18+
if (sm.id) {
19+
try {
20+
const element = await resume(entity, sm);
21+
await resumed(element);
22+
return;
23+
// If resumption fails, continue with session establishment
24+
} catch {
25+
failed();
26+
}
27+
}
28+
29+
// Enabling
30+
31+
// Resource binding first
32+
await next();
33+
34+
const promiseEnable = enable(entity, sm);
35+
36+
if (sm.outbound_q.length > 0) {
37+
throw new Error(
38+
"Stream Management assertion failure, queue should be empty after enable",
39+
);
40+
}
41+
42+
// > The counter for an entity's own sent stanzas is set to zero and started after sending either <enable/> or <enabled/>.
43+
sm.outbound = 0;
44+
45+
try {
46+
const response = await promiseEnable;
47+
enabled(response.attrs);
48+
} catch {
49+
sm.enabled = false;
50+
}
51+
52+
sm.inbound = 0;
53+
});
54+
}
55+
56+
export function enable(entity, sm) {
57+
return procedure(entity, makeEnableElement({ sm }), (element, done) => {
58+
if (element.is("enabled", NS)) {
59+
return done(element);
60+
} else if (element.is("failed", NS)) {
61+
throw XMPPError.fromElement(element);
62+
}
63+
});
64+
}
65+
66+
export async function resume(entity, sm) {
67+
return procedure(entity, makeResumeElement({ sm }), (element, done) => {
68+
if (element.is("resumed", NS)) {
69+
return done(element);
70+
} else if (element.is("failed", NS)) {
71+
throw XMPPError.fromElement(element);
72+
}
73+
});
74+
}

‎packages/stream-management/stream-features.test.js

-28
Original file line numberDiff line numberDiff line change
@@ -124,34 +124,6 @@ test("enable - failed", async () => {
124124
expect(entity.streamManagement.enabled).toBe(false);
125125
});
126126

127-
test("stanza ack", async () => {
128-
const { entity } = mockClient();
129-
130-
entity.streamManagement.enabled = true;
131-
132-
expect(entity.streamManagement.outbound).toBe(0);
133-
expect(entity.streamManagement.outbound_q).toBeEmpty();
134-
// expect(entity.streamManagement.enabled).toBe(true);
135-
136-
await entity.send(<message id="a" />);
137-
138-
expect(entity.streamManagement.outbound).toBe(0);
139-
expect(entity.streamManagement.outbound_q).toHaveLength(1);
140-
141-
let acks = 0;
142-
entity.streamManagement.on("ack", (stanza) => {
143-
expect(stanza.attrs.id).toBe("a");
144-
acks++;
145-
});
146-
147-
entity.mockInput(<a xmlns="urn:xmpp:sm:3" h="1" />);
148-
await tick();
149-
150-
expect(acks).toBe(1);
151-
expect(entity.streamManagement.outbound).toBe(1);
152-
expect(entity.streamManagement.outbound_q).toHaveLength(0);
153-
});
154-
155127
test("resume - resumed", async () => {
156128
const { entity } = mockClient();
157129

‎packages/stream-management/stream-management.test.js

+46
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
import { mockClient } from "@xmpp/test";
2+
3+
import { tick } from "@xmpp/events";
4+
5+
test("emits ack when the server ackownledge stanzas", async () => {
6+
const { entity } = mockClient();
7+
8+
entity.streamManagement.enabled = true;
9+
10+
expect(entity.streamManagement.outbound).toBe(0);
11+
expect(entity.streamManagement.outbound_q).toBeEmpty();
12+
// expect(entity.streamManagement.enabled).toBe(true);
13+
14+
await entity.send(<message id="a" />);
15+
16+
expect(entity.streamManagement.outbound).toBe(0);
17+
expect(entity.streamManagement.outbound_q).toHaveLength(1);
18+
19+
let acks = 0;
20+
entity.streamManagement.on("ack", (stanza) => {
21+
expect(stanza.attrs.id).toBe("a");
22+
acks++;
23+
});
24+
25+
entity.mockInput(<a xmlns="urn:xmpp:sm:3" h="1" />);
26+
await tick();
27+
28+
expect(acks).toBe(1);
29+
expect(entity.streamManagement.outbound).toBe(1);
30+
expect(entity.streamManagement.outbound_q).toHaveLength(0);
31+
});
32+
33+
test("sends an <a/> element before closing", async () => {
34+
const { entity, streamManagement } = mockClient();
35+
streamManagement.enabled = true;
36+
streamManagement.inbound = 42;
37+
entity.status = "online";
38+
39+
const promise_disconnect = entity.disconnect();
40+
41+
expect(await entity.catchOutgoing()).toEqual(
42+
<a xmlns="urn:xmpp:sm:3" h={streamManagement.inbound} />,
43+
);
44+
45+
await promise_disconnect;
46+
});

‎server/prosody.cfg.lua

+1
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ modules_enabled = {
2626
"sasl2_bind2";
2727
"sasl2_sm";
2828
"sasl2_fast";
29+
"csi_simple";
2930
};
3031

3132
modules_disabled = {

0 commit comments

Comments
 (0)