1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
using Gee;
namespace Xmpp.Iq {
private const string NS_URI = "jabber:client";
public class Module : XmppStreamNegotiationModule {
public static ModuleIdentity<Module> IDENTITY = new ModuleIdentity<Module>(NS_URI, "iq_module");
public signal void preprocess_incoming_iq_set_get(XmppStream stream, Stanza iq_stanza);
public signal void preprocess_outgoing_iq_set_get(XmppStream stream, Stanza iq_stanza);
private HashMap<string, ResponseListener> responseListeners = new HashMap<string, ResponseListener>();
private HashMap<string, ArrayList<Handler>> namespaceRegistrants = new HashMap<string, ArrayList<Handler>>();
public async Iq.Stanza send_iq_async(XmppStream stream, Iq.Stanza iq, int io_priority = Priority.DEFAULT, Cancellable? cancellable = null) throws IOError {
assert(iq.type_ == Iq.Stanza.TYPE_GET || iq.type_ == Iq.Stanza.TYPE_SET);
preprocess_outgoing_iq_set_get(stream, iq);
Iq.Stanza? return_stanza = null;
IOError? return_error = null;
bool yielded = false;
responseListeners[iq.id] = new ResponseListener((_, result_iq) => {
return_stanza = result_iq;
if (yielded) send_iq_async.callback();
});
stream.write_async.begin(iq.stanza, io_priority, cancellable, (_, res) => {
try {
stream.write_async.end(res);
} catch (IOError e) {
return_error = e;
}
});
ulong cancelled_handler_id = 0;
if (cancellable != null) {
cancelled_handler_id = cancellable.connect(() => {
if (yielded) send_iq_async.callback();
});
}
if ((cancellable == null || !cancellable.is_cancelled()) && return_error == null) {
yielded = true;
yield;
yielded = false;
}
if (cancellable != null) {
if (cancellable.is_cancelled()) {
cancellable.set_error_if_cancelled();
} else {
cancellable.disconnect(cancelled_handler_id);
}
}
responseListeners.unset(iq.id);
if (return_error != null) throw return_error;
return return_stanza;
}
public delegate void OnResult(XmppStream stream, Iq.Stanza iq);
public void send_iq(XmppStream stream, Iq.Stanza iq, owned OnResult? listener = null, int io_priority = Priority.DEFAULT) {
preprocess_outgoing_iq_set_get(stream, iq);
stream.write(iq.stanza, io_priority);
if (listener != null) {
responseListeners[iq.id] = new ResponseListener((owned) listener);
}
}
public void register_for_namespace(string namespace, Handler module) {
if (!namespaceRegistrants.has_key(namespace)) {
namespaceRegistrants.set(namespace, new ArrayList<Handler>());
}
namespaceRegistrants[namespace].add(module);
}
public void unregister_from_namespace(string namespace, Handler module) {
ArrayList<Handler>? handlers = namespaceRegistrants[namespace];
if (handlers != null) handlers.remove(module);
}
public override void attach(XmppStream stream) {
stream.received_iq_stanza.connect(on_received_iq_stanza);
}
public override void detach(XmppStream stream) {
stream.received_iq_stanza.disconnect(on_received_iq_stanza);
}
public override bool mandatory_outstanding(XmppStream stream) { return false; }
public override bool negotiation_active(XmppStream stream) { return false; }
public override string get_ns() { return NS_URI; }
public override string get_id() { return IDENTITY.id; }
private async void on_received_iq_stanza(XmppStream stream, StanzaNode node) {
Iq.Stanza iq = new Iq.Stanza.from_stanza(node, stream.has_flag(Bind.Flag.IDENTITY) ? stream.get_flag(Bind.Flag.IDENTITY).my_jid : null);
if (iq.type_ == Iq.Stanza.TYPE_RESULT || iq.is_error()) {
if (responseListeners.has_key(iq.id)) {
ResponseListener? listener = responseListeners.get(iq.id);
if (listener != null) {
listener.on_result(stream, iq);
}
responseListeners.unset(iq.id);
}
} else {
Gee.List<StanzaNode> children = node.get_all_subnodes();
if (children.size == 1 && namespaceRegistrants.has_key(children[0].ns_uri)) {
preprocess_incoming_iq_set_get(stream, iq);
Gee.List<Handler> handlers = namespaceRegistrants[children[0].ns_uri];
foreach (Handler handler in handlers) {
if (iq.type_ == Iq.Stanza.TYPE_GET) {
yield handler.on_iq_get(stream, iq);
} else if (iq.type_ == Iq.Stanza.TYPE_SET) {
yield handler.on_iq_set(stream, iq);
}
}
} else {
// Send error if we don't handle the NS of the IQ get/set payload (RFC6120 10.3.3 (2))
Iq.Stanza unavailable_error = new Iq.Stanza.error(iq, new ErrorStanza.service_unavailable()) { to=iq.from };
send_iq(stream, unavailable_error);
}
}
}
private class ResponseListener {
public OnResult on_result { get; private owned set; }
public ResponseListener(owned OnResult on_result) {
this.on_result = (owned) on_result;
}
}
}
public interface Handler : Object {
public async virtual void on_iq_get(XmppStream stream, Iq.Stanza iq) {
Iq.Stanza bad_request = new Iq.Stanza.error(iq, new ErrorStanza.bad_request("unexpected IQ get for this namespace"));
stream.get_module(Module.IDENTITY).send_iq(stream, bad_request);
}
public async virtual void on_iq_set(XmppStream stream, Iq.Stanza iq) {
Iq.Stanza bad_request = new Iq.Stanza.error(iq, new ErrorStanza.bad_request("unexpected IQ set for this namespace"));
stream.get_module(Module.IDENTITY).send_iq(stream, bad_request);
}
}
}
|
