Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include "xmpp/xmpp_state_machine.h"
7 :
8 : #include <atomic>
9 : #include <typeinfo>
10 : #include <boost/bind/bind.hpp>
11 : #include <boost/date_time/posix_time/posix_time.hpp>
12 : #include <boost/statechart/custom_reaction.hpp>
13 : #include <boost/statechart/event.hpp>
14 : #include <boost/statechart/simple_state.hpp>
15 : #include <boost/statechart/state.hpp>
16 : #include <boost/statechart/state_machine.hpp>
17 : #include <boost/statechart/transition.hpp>
18 :
19 : #include "base/logging.h"
20 : #include "base/task_annotations.h"
21 : #include "io/event_manager.h"
22 : #include "io/ssl_session.h"
23 : #include "sandesh/sandesh_trace.h"
24 : #include "sandesh/common/vns_types.h"
25 : #include "sandesh/common/vns_constants.h"
26 : #include "sandesh/xmpp_client_server_sandesh_types.h"
27 : #include "sandesh/xmpp_peer_info_types.h"
28 : #include "sandesh/xmpp_state_machine_sandesh_types.h"
29 : #include "sandesh/xmpp_trace_sandesh_types.h"
30 : #include "xmpp/xmpp_connection.h"
31 : #include "xmpp/xmpp_factory.h"
32 : #include "xmpp/xmpp_log.h"
33 : #include "xmpp/xmpp_server.h"
34 : #include "xmpp/xmpp_session.h"
35 :
36 : using namespace std;
37 : using namespace boost::placeholders;
38 :
39 : namespace mpl = boost::mpl;
40 : namespace sc = boost::statechart;
41 :
42 : #define SM_LOG(_sm, _msg) do { \
43 : XMPP_UTDEBUG(XmppStateMachineDebug, \
44 : (_sm)->connection() ? (_sm)->connection()->ToUVEKey() : "", \
45 : XMPP_PEER_DIR_NA, (_sm)->ChannelType(), _msg); \
46 : } while (false)
47 :
48 : namespace xmsm {
49 : // events
50 :
51 : struct EvStart : sc::event<EvStart> {
52 : static const char *Name() {
53 : return "EvStart";
54 : }
55 : };
56 :
57 : struct EvStop : sc::event<EvStop> {
58 2866 : static const char *Name() {
59 2866 : return "EvStop";
60 : }
61 : };
62 :
63 : struct EvAdminDown : sc::event<EvAdminDown> {
64 3354 : static const char *Name() {
65 3354 : return "EvAdminDown";
66 : }
67 : };
68 :
69 : struct EvConnectTimerExpired : sc::event<EvConnectTimerExpired> {
70 7747 : static const char *Name() {
71 7747 : return "EvConnectTimerExpired";
72 : }
73 : };
74 :
75 : struct EvOpenTimerExpired : sc::event<EvOpenTimerExpired> {
76 1 : static const char *Name() {
77 1 : return "EvOpenTimerExpired";
78 : }
79 : };
80 :
81 : struct EvHoldTimerExpired : sc::event<EvHoldTimerExpired> {
82 13 : static const char *Name() {
83 13 : return "EvHoldTimerExpired";
84 : }
85 : };
86 :
87 : struct EvTcpConnected : sc::event<EvTcpConnected> {
88 6583 : EvTcpConnected(XmppSession *session) : session(session) { };
89 6583 : static const char *Name() {
90 6583 : return "EvTcpConnected";
91 : }
92 : XmppSession *session;
93 : } ;
94 :
95 : struct EvTcpConnectFail : sc::event<EvTcpConnectFail> {
96 1159 : EvTcpConnectFail(XmppSession *session) : session(session) { };
97 1159 : static const char *Name() {
98 1159 : return "EvTcpConnectFail";
99 : }
100 : XmppSession *session;
101 : } ;
102 :
103 : struct EvTcpPassiveOpen : sc::event<EvTcpPassiveOpen> {
104 6521 : EvTcpPassiveOpen(XmppSession *session) : session(session) { };
105 6520 : static const char *Name() {
106 6520 : return "EvTcpPassiveOpen";
107 : }
108 : XmppSession *session;
109 : };
110 :
111 : struct EvTcpClose : sc::event<EvTcpClose> {
112 6687 : EvTcpClose(XmppSession *session) : session(session) { };
113 6474 : static const char *Name() {
114 6474 : return "EvTcpClose";
115 : }
116 : XmppSession *session;
117 : };
118 :
119 : struct EvTcpDeleteSession : sc::event<EvTcpDeleteSession> {
120 12306 : explicit EvTcpDeleteSession(TcpSession *session) : session(session) { }
121 : static const char *Name() {
122 : return "EvTcpDeleteSession";
123 : }
124 : TcpSession *session;
125 : };
126 :
127 : struct EvXmppMessage : sc::event<EvXmppMessage> {
128 4110882 : explicit EvXmppMessage(XmppSession *session,
129 4110882 : const XmppStanza::XmppMessage *msg) : session(session), msg(msg) { }
130 : static const char *Name() {
131 : return "EvXmppMessage";
132 : }
133 : XmppSession *session;
134 : const XmppStanza::XmppMessage *msg;
135 : };
136 :
137 : struct EvXmppOpen : public sc::event<EvXmppOpen> {
138 15590 : EvXmppOpen(XmppSession *session, const XmppStanza::XmppMessage *msg) :
139 15590 : session(session),
140 15590 : msg(static_cast<const XmppStanza::XmppStreamMessage *>(msg)) {
141 15590 : }
142 15588 : static const char *Name() {
143 15588 : return "EvXmppOpen";
144 : }
145 : XmppSession *session;
146 : boost::shared_ptr<const XmppStanza::XmppStreamMessage> msg;
147 : };
148 :
149 : struct EvStreamFeatureRequest : sc::event<EvStreamFeatureRequest> {
150 1469 : EvStreamFeatureRequest(XmppSession *session,
151 1469 : const XmppStanza::XmppMessage *msg) :
152 1469 : session(session),
153 1469 : msg(static_cast<const XmppStanza::XmppStreamMessage *>(msg)) {
154 1469 : }
155 1469 : static const char *Name() {
156 1469 : return "EvStreamFeatureRequest";
157 : }
158 : XmppSession *session;
159 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
160 : };
161 :
162 : struct EvStartTls : sc::event<EvStartTls> {
163 1464 : EvStartTls(XmppSession *session,
164 1464 : const XmppStanza::XmppMessage *msg) :
165 1464 : session(session),
166 1464 : msg(static_cast<const XmppStanza::XmppStreamMessage *>(msg)) {
167 1464 : }
168 1464 : static const char *Name() {
169 1464 : return "EvStartTls";
170 : }
171 : XmppSession *session;
172 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
173 : };
174 :
175 : struct EvTlsProceed : sc::event<EvTlsProceed> {
176 1467 : EvTlsProceed(XmppSession *session,
177 1467 : const XmppStanza::XmppMessage *msg) :
178 1467 : session(session),
179 1467 : msg(static_cast<const XmppStanza::XmppStreamMessage *>(msg)) {
180 1467 : }
181 1467 : static const char *Name() {
182 1467 : return "EvTlsProceed";
183 : }
184 : XmppSession *session;
185 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
186 : };
187 :
188 : struct EvTlsHandShakeSuccess : sc::event<EvTlsHandShakeSuccess> {
189 2924 : explicit EvTlsHandShakeSuccess(XmppSession *session) :
190 2924 : session(session) { }
191 2924 : static const char *Name() {
192 2924 : return "EvTlsHandShakeSuccess";
193 : }
194 : XmppSession *session;
195 : };
196 :
197 : struct EvTlsHandShakeFailure : sc::event<EvTlsHandShakeFailure> {
198 2 : explicit EvTlsHandShakeFailure(XmppSession *session) :
199 2 : session(session) { }
200 2 : static const char *Name() {
201 2 : return "EvTlsHandShakeFailure";
202 : }
203 : XmppSession *session;
204 : };
205 :
206 : struct EvXmppKeepalive : sc::event<EvXmppKeepalive> {
207 2476835 : EvXmppKeepalive(XmppSession *session,
208 2476835 : const XmppStanza::XmppMessage *msg) :
209 2476835 : session(session), msg(msg) {
210 2476835 : }
211 : static const char *Name() {
212 : return "EvXmppKeepalive";
213 : }
214 : XmppSession *session;
215 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
216 : };
217 :
218 : struct EvXmppMessageStanza : sc::event<EvXmppMessageStanza> {
219 1556755 : EvXmppMessageStanza(XmppSession *session,
220 1556755 : const XmppStanza::XmppMessage *msg) :
221 1556755 : session(session), msg(msg) {
222 1556755 : }
223 : static const char *Name() {
224 : return "EvXmppMessageStanza";
225 : }
226 : XmppSession *session;
227 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
228 : };
229 :
230 : struct EvXmppIqStanza : sc::event<EvXmppIqStanza> {
231 56895 : EvXmppIqStanza(XmppSession *session,
232 : const XmppStanza::XmppMessage *msg)
233 56895 : : session(session), msg(msg) {
234 56895 : }
235 : static const char *Name() {
236 : return "EvXmppIqStanza";
237 : }
238 : XmppSession *session;
239 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
240 : };
241 :
242 : struct EvXmppOpenReceive : sc::event<EvXmppOpenReceive> {
243 : EvXmppOpenReceive(XmppSession *session) : session(session) {
244 : }
245 : static const char *Name() {
246 : return "EvXmppOpenReceive";
247 : }
248 : XmppSession *session;
249 : };
250 :
251 : struct Idle : public sc::state<Idle, XmppStateMachine> {
252 : typedef sc::transition<EvStart, Active,
253 : XmppStateMachine, &XmppStateMachine::OnStart> reactions;
254 21773 : Idle(my_context ctx) : my_base(ctx) {
255 21775 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
256 21780 : bool flap = (state_machine->get_state() == ESTABLISHED);
257 21779 : state_machine->set_state(IDLE);
258 21782 : state_machine->SendConnectionInfo("Start", "Active");
259 21784 : if (flap) {
260 9482 : state_machine->connection()->increment_flap_count();
261 : }
262 21783 : }
263 : };
264 :
265 : struct Active : public sc::state<Active, XmppStateMachine> {
266 : typedef mpl::list<
267 : sc::custom_reaction<EvAdminDown>,
268 : sc::custom_reaction<EvConnectTimerExpired>,
269 : sc::custom_reaction<EvOpenTimerExpired>,
270 : sc::custom_reaction<EvTcpPassiveOpen>,
271 : sc::custom_reaction<EvTcpClose>,
272 : sc::custom_reaction<EvXmppOpen>,
273 : sc::custom_reaction<EvStop>
274 : > reactions;
275 :
276 15977 : Active(my_context ctx) : my_base(ctx) {
277 15976 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
278 15978 : state_machine->keepalive_count_clear();
279 15977 : bool flap = (state_machine->get_state() == ESTABLISHED);
280 15977 : state_machine->set_state(ACTIVE);
281 15978 : if (flap) {
282 2578 : state_machine->connection()->increment_flap_count();
283 : }
284 15978 : if (state_machine->IsActiveChannel() ) {
285 9181 : if (state_machine->get_connect_attempts() >=
286 : XmppStateMachine::kMaxAttempts) {
287 285 : XmppConnection *connection = state_machine->connection();
288 285 : if (connection) {
289 285 : state_machine->SendConnectionInfo(
290 : "Connect failed after retries");
291 : // Notify clients if any action to be taken
292 285 : connection->ChannelMux()->HandleStateEvent(xmsm::ACTIVE);
293 : }
294 : }
295 9181 : state_machine->StartConnectTimer(state_machine->GetConnectTime());
296 : }
297 15978 : }
298 31957 : ~Active() {
299 15979 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
300 15979 : state_machine->CancelConnectTimer();
301 15979 : state_machine->CancelOpenTimer();
302 31957 : }
303 :
304 : //event on client only
305 7746 : sc::result react(const EvConnectTimerExpired &event) {
306 7746 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
307 7746 : if (state_machine->ConnectTimerCancelled()) {
308 0 : SM_LOG(state_machine, "Discard EvConnectTimerExpired in (Active) State");
309 0 : return discard_event();
310 : } else {
311 7746 : state_machine->SendConnectionInfo(event.Name(), "Connect");
312 7746 : SM_LOG(state_machine, "EvConnectTimerExpired in (Active) State");
313 7746 : return transit<Connect>();
314 : }
315 : }
316 :
317 : // event on server only
318 6521 : sc::result react(const EvTcpPassiveOpen &event) {
319 6521 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
320 6520 : event.session->AsyncReadStart();
321 6521 : assert(state_machine->session() == event.session);
322 6521 : event.session->set_observer(
323 : boost::bind(&XmppStateMachine::OnSessionEvent,
324 : state_machine, _1, _2));
325 6520 : state_machine->StartOpenTimer(XmppStateMachine::kOpenTime);
326 6521 : XmppConnectionInfo info;
327 6521 : info.set_local_port(event.session->local_port());
328 6519 : info.set_remote_port(event.session->remote_port());
329 6521 : state_machine->SendConnectionInfo(&info, event.Name());
330 13042 : return discard_event();
331 6521 : }
332 :
333 : // event on server only
334 182 : sc::result react(const EvTcpClose &event) {
335 182 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
336 182 : if (event.session != state_machine->session()) {
337 0 : return discard_event();
338 : }
339 182 : state_machine->CancelOpenTimer();
340 182 : state_machine->ResetSession();
341 182 : state_machine->SendConnectionInfo(event.Name(), "Idle");
342 182 : return transit<Idle>();
343 : }
344 :
345 : //event on server only
346 1 : sc::result react(const EvOpenTimerExpired &event) {
347 1 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
348 1 : if (state_machine->OpenTimerCancelled()) {
349 0 : SM_LOG(state_machine,
350 : "Discard EvOpenTimerExpired in (Active) State");
351 0 : return discard_event();
352 : }
353 :
354 : // At this point session on connection is not set, hence SendClose
355 : // using session on the state_machine.
356 1 : XmppSession *session = state_machine->session();
357 1 : XmppConnection *connection = state_machine->connection();
358 1 : connection->SendClose(session);
359 1 : state_machine->ResetSession();
360 1 : state_machine->SendConnectionInfo(event.Name(), "Idle");
361 1 : return transit<Idle>();
362 : }
363 :
364 : //event on server only
365 6358 : sc::result react(const EvXmppOpen &event) {
366 6358 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
367 6358 : if (event.session != state_machine->session()) {
368 0 : return discard_event();
369 : }
370 6358 : state_machine->AssignSession();
371 :
372 6358 : XmppConnection *connection = state_machine->connection();
373 6358 : if (connection->IsDeleted()) {
374 0 : state_machine->ResetSession();
375 0 : return discard_event();
376 : }
377 :
378 6358 : XmppSession *session = state_machine->session();
379 6358 : if (!state_machine->IsAuthEnabled())
380 4846 : state_machine->ResurrectOldConnection(connection, session);
381 :
382 : // Get the possibly updated XmppConnection information.
383 6358 : connection = state_machine->connection();
384 6358 : state_machine->CancelOpenTimer();
385 6358 : if (!connection->SendOpenConfirm(session)) {
386 0 : connection->SendClose(session);
387 0 : state_machine->ResetSession();
388 0 : state_machine->SendConnectionInfo("Send Open Confirm Failed",
389 : "Idle");
390 0 : return transit<Idle>();
391 : } else {
392 6358 : XmppConnectionInfo info;
393 6358 : info.set_identifier(event.msg->from);
394 6358 : if (state_machine->IsAuthEnabled()) {
395 1512 : state_machine->SendConnectionInfo(&info, event.Name(),
396 : "Open Confirm");
397 1512 : return transit<OpenConfirm>();
398 : } else {
399 4846 : connection->StartKeepAliveTimer();
400 4846 : state_machine->SendConnectionInfo(&info, event.Name(),
401 : "Established");
402 4846 : return transit<XmppStreamEstablished>();
403 : }
404 6358 : }
405 : }
406 :
407 66 : sc::result react(const EvStop &event) {
408 66 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
409 66 : XmppConnection *connection = state_machine->connection();
410 66 : state_machine->CancelOpenTimer();
411 66 : state_machine->CancelConnectTimer();
412 66 : connection->StopKeepAliveTimer();
413 :
414 66 : if (state_machine->IsActiveChannel() ) {
415 1 : state_machine->set_session(NULL);
416 1 : state_machine->SendConnectionInfo(event.Name(), "Active");
417 1 : return transit<Active>();
418 : } else {
419 65 : state_machine->ResetSession();
420 65 : state_machine->SendConnectionInfo(event.Name(), "Idle");
421 65 : return transit<Idle>();
422 : }
423 : }
424 :
425 99 : sc::result react(const EvAdminDown &event) {
426 99 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
427 99 : XmppConnectionInfo info;
428 99 : info.set_close_reason("Administratively down");
429 99 : state_machine->connection()->set_close_reason("Administratively down");
430 99 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
431 198 : return transit<Idle>();
432 99 : }
433 :
434 : };
435 :
436 : //State valid only for client side, connection in Active state
437 : struct Connect : public sc::state<Connect, XmppStateMachine> {
438 : typedef mpl::list<
439 : sc::custom_reaction<EvAdminDown>,
440 : sc::custom_reaction<EvConnectTimerExpired>,
441 : sc::custom_reaction<EvTcpConnected>,
442 : sc::custom_reaction<EvTcpConnectFail>,
443 : sc::custom_reaction<EvTcpClose>,
444 : sc::custom_reaction<EvStop>
445 : > reactions;
446 :
447 : static const int kConnectTimeout = 60; // seconds
448 :
449 7746 : Connect(my_context ctx) : my_base(ctx) {
450 7746 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
451 7746 : StartSession(state_machine);
452 7746 : state_machine->connect_attempts_inc();
453 7746 : state_machine->set_state(CONNECT);
454 7746 : state_machine->StartConnectTimer(state_machine->GetConnectTime());
455 7746 : XmppSession *session = state_machine->session();
456 7746 : if (session != NULL) {
457 7746 : XmppConnectionInfo info;
458 7746 : info.set_local_port(session->local_port());
459 7746 : info.set_remote_port(session->remote_port());
460 7746 : state_machine->SendConnectionInfo(&info, "Connect Event");
461 7746 : }
462 7746 : }
463 15492 : ~Connect() {
464 7746 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
465 7746 : state_machine->CancelConnectTimer();
466 15492 : }
467 :
468 1 : sc::result react(const EvConnectTimerExpired &event) {
469 1 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
470 1 : if (state_machine->ConnectTimerCancelled()) {
471 0 : SM_LOG(state_machine,
472 : "Discard EvConnectTimerExpired in (Connect) State");
473 0 : return discard_event();
474 : }
475 1 : CloseSession(state_machine);
476 1 : XmppConnectionInfo info;
477 1 : info.set_close_reason("Connect timer expired");
478 1 : state_machine->connection()->set_close_reason("Connect timer expired");
479 1 : state_machine->SendConnectionInfo(&info, event.Name(), "Active");
480 1 : return transit<Active>();
481 1 : }
482 :
483 6583 : sc::result react(const EvTcpConnected &event) {
484 6583 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
485 6583 : if (event.session != state_machine->session()) {
486 0 : return discard_event();
487 : }
488 6583 : XmppConnection *connection = state_machine->connection();
489 6583 : state_machine->CancelConnectTimer();
490 6583 : XmppSession *session = state_machine->session();
491 6583 : XmppConnectionInfo info;
492 6583 : info.set_local_port(session->local_port());
493 6583 : info.set_remote_port(session->remote_port());
494 6583 : if (connection->SendOpen(session)) {
495 6583 : state_machine->StartHoldTimer();
496 6583 : state_machine->SendConnectionInfo(&info, event.Name(), "OpenSent");
497 6583 : return transit<OpenSent>();
498 : } else {
499 0 : SM_LOG(state_machine, "SendOpen failed in (Connect) State");
500 0 : CloseSession(state_machine);
501 0 : info.set_close_reason("SendOpen failed");
502 0 : state_machine->connection()->set_close_reason("Send Open failed");
503 0 : state_machine->SendConnectionInfo(&info, "Send Open failed",
504 : "Active");
505 0 : return transit<Active>();
506 : }
507 6583 : }
508 :
509 1159 : sc::result react(const EvTcpConnectFail &event) {
510 : // delete session; restart connect timer.
511 1159 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
512 1159 : if (event.session != state_machine->session()) {
513 0 : return discard_event();
514 : }
515 1159 : state_machine->set_session(NULL);
516 1159 : state_machine->CancelConnectTimer();
517 1159 : state_machine->SendConnectionInfo(event.Name(), "Active");
518 1159 : return transit<Active>();
519 : }
520 :
521 1 : sc::result react(const EvTcpClose &event) {
522 1 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
523 1 : if (event.session != state_machine->session()) {
524 0 : return discard_event();
525 : }
526 : // close the tcp sessions.
527 1 : CloseSession(state_machine);
528 1 : state_machine->CancelConnectTimer();
529 1 : state_machine->SendConnectionInfo(event.Name(), "Active");
530 1 : return transit<Active>();
531 : }
532 :
533 1 : sc::result react(const EvStop &event) {
534 1 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
535 1 : CloseSession(state_machine);
536 1 : state_machine->CancelConnectTimer();
537 1 : XmppConnectionInfo info;
538 1 : info.set_close_reason("EvStop received");
539 1 : state_machine->connection()->set_close_reason("EvStop received");
540 1 : state_machine->SendConnectionInfo(&info, event.Name(), "Active");
541 2 : return transit<Active>();
542 1 : }
543 :
544 0 : sc::result react(const EvAdminDown &event) {
545 0 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
546 0 : CloseSession(state_machine);
547 0 : XmppConnectionInfo info;
548 0 : info.set_close_reason("Administratively down");
549 0 : state_machine->connection()->set_close_reason("Administratively down");
550 0 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
551 0 : return transit<Idle>();
552 0 : }
553 :
554 : // Create an active connection request.
555 7746 : void StartSession(XmppStateMachine *state_machine) {
556 7746 : XmppConnection *connection = state_machine->connection();
557 7746 : XmppSession *session = connection->CreateSession();
558 7746 : state_machine->set_session(session);
559 7746 : session->set_observer(boost::bind(&XmppStateMachine::OnSessionEvent,
560 : state_machine, _1, _2));
561 7746 : boost::system::error_code err;
562 7746 : session->socket()->bind(connection->local_endpoint(), err);
563 7746 : if (err) {
564 0 : LOG(WARN, "Bind failure for local address " <<
565 : connection->local_endpoint() << " : " << err.message());
566 0 : assert(false);
567 : }
568 7746 : connection->server()->Connect(session, connection->endpoint());
569 7746 : }
570 :
571 3 : void CloseSession(XmppStateMachine *state_machine) {
572 3 : state_machine->set_session(NULL);
573 3 : }
574 : };
575 :
576 : // The client reaches OpenSent after sending an immediate OPEN on a active
577 : // connection. Server should not come in this state.
578 : struct OpenSent : public sc::state<OpenSent, XmppStateMachine> {
579 : typedef mpl::list<
580 : sc::custom_reaction<EvAdminDown>,
581 : sc::custom_reaction<EvTcpClose>,
582 : sc::custom_reaction<EvXmppOpen>,
583 : sc::custom_reaction<EvHoldTimerExpired>,
584 : sc::custom_reaction<EvStop>
585 : > reactions;
586 :
587 6583 : OpenSent(my_context ctx) : my_base(ctx) {
588 6583 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
589 6583 : state_machine->set_state(OPENSENT);
590 6583 : }
591 13166 : ~OpenSent() {
592 6583 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
593 6583 : state_machine->CancelHoldTimer();
594 13166 : }
595 :
596 175 : sc::result react(const EvTcpClose &event) {
597 175 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
598 175 : if (event.session != state_machine->session()) {
599 0 : return discard_event();
600 : }
601 175 : state_machine->CancelHoldTimer();
602 175 : if (event.session) {
603 175 : state_machine->set_session(NULL);
604 175 : state_machine->SendConnectionInfo(event.Name(), "Active");
605 175 : return transit<Active>();
606 : }
607 0 : return discard_event();
608 : }
609 :
610 6312 : sc::result react(const EvXmppOpen &event) {
611 6312 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
612 6312 : XmppConnection *connection = state_machine->connection();
613 6312 : if (event.session == state_machine->session()) {
614 6312 : state_machine->AssignSession();
615 6312 : XmppConnectionInfo info;
616 6312 : info.set_identifier(event.msg->from);
617 6312 : if (state_machine->IsAuthEnabled()) {
618 1472 : state_machine->SendConnectionInfo(&info, event.Name(),
619 : "Open Confirm");
620 1472 : return transit<OpenConfirm>();
621 : } else {
622 4840 : connection->SendKeepAlive();
623 4840 : connection->StartKeepAliveTimer();
624 4840 : state_machine->StartHoldTimer();
625 4840 : state_machine->SendConnectionInfo(&info, event.Name(),
626 : "Established");
627 4840 : return transit<XmppStreamEstablished>();
628 : }
629 6312 : }
630 0 : return discard_event();
631 : }
632 :
633 2 : sc::result react(const EvHoldTimerExpired &event) {
634 2 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
635 2 : if (state_machine->HoldTimerCancelled()) {
636 0 : SM_LOG(state_machine,
637 : "Discard EvHoldTimerExpired in (OpenSent) State");
638 0 : return discard_event();
639 : }
640 2 : CloseSession(state_machine);
641 2 : XmppConnectionInfo info;
642 2 : info.set_close_reason("Hold timer expired");
643 2 : state_machine->connection()->set_close_reason("Hold timer expired");
644 2 : state_machine->SendConnectionInfo(&info, event.Name(), "Active");
645 2 : return transit<Active>();
646 2 : }
647 :
648 2 : sc::result react(const EvStop &event) {
649 2 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
650 2 : state_machine->CancelHoldTimer();
651 2 : CloseSession(state_machine);
652 2 : XmppConnectionInfo info;
653 2 : info.set_close_reason("EvStop received");
654 2 : state_machine->connection()->set_close_reason("EvStop received");
655 2 : state_machine->SendConnectionInfo(&info, event.Name(), "Active");
656 4 : return transit<Active>();
657 2 : }
658 :
659 72 : sc::result react(const EvAdminDown &event) {
660 72 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
661 72 : CloseSession(state_machine);
662 72 : XmppConnectionInfo info;
663 72 : info.set_close_reason("Administratively down");
664 72 : state_machine->connection()->set_close_reason("Administratively down");
665 72 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
666 144 : return transit<Idle>();
667 72 : }
668 :
669 76 : void CloseSession(XmppStateMachine *state_machine) {
670 76 : state_machine->set_session(NULL);
671 76 : }
672 : };
673 :
674 : struct OpenConfirm : public sc::state<OpenConfirm, XmppStateMachine> {
675 : typedef mpl::list<
676 : sc::custom_reaction<EvAdminDown>,
677 : sc::custom_reaction<EvTcpClose>,
678 : sc::custom_reaction<EvHoldTimerExpired>,
679 : sc::custom_reaction<EvStreamFeatureRequest>, //received by client
680 : sc::custom_reaction<EvStartTls>, //received by server
681 : sc::custom_reaction<EvTlsProceed>, //received by client
682 : sc::custom_reaction<EvTlsHandShakeSuccess>,
683 : sc::custom_reaction<EvTlsHandShakeFailure>,
684 : sc::custom_reaction<EvXmppOpen>, //received by server
685 : sc::custom_reaction<EvStop>
686 : > reactions;
687 :
688 2984 : OpenConfirm(my_context ctx) : my_base(ctx) {
689 2984 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
690 2984 : state_machine->StartHoldTimer();
691 2984 : XmppConnectionInfo info;
692 2984 : if (!state_machine->IsActiveChannel()) { //server
693 1512 : if (state_machine->IsAuthEnabled()) {
694 1512 : XmppConnection *connection = state_machine->connection();
695 1512 : XmppSession *session = state_machine->session();
696 1512 : if (!connection->SendStreamFeatureRequest(session)) {
697 0 : connection->SendClose(session);
698 0 : SM_LOG(state_machine,
699 : "Xmpp Send Stream Feature Request Failed, IDLE");
700 0 : state_machine->ResetSession();
701 0 : info.set_close_reason("Send Stream Feature Request Failed");
702 0 : state_machine->connection()->set_close_reason(
703 : "Send Stream Feature Request Failed");
704 0 : state_machine->SendConnectionInfo(&info,
705 : "Send Stream Feature Request failed", "Idle");
706 : // cannot transition state as this is the constructor
707 : // of new state
708 : }
709 : }
710 : }
711 2984 : state_machine->set_state(OPENCONFIRM);
712 2984 : state_machine->set_openconfirm_state(OPENCONFIRM_INIT);
713 2984 : }
714 :
715 46 : sc::result react(const EvTcpClose &event) {
716 46 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
717 46 : state_machine->CancelHoldTimer();
718 46 : if (state_machine->IsActiveChannel() ) {
719 1 : CloseSession(state_machine);
720 1 : state_machine->SendConnectionInfo(event.Name(), "Active");
721 1 : return transit<Active>();
722 : } else {
723 45 : state_machine->ResetSession();
724 45 : state_machine->SendConnectionInfo(event.Name(), "Idle");
725 45 : return transit<Idle>();
726 : }
727 : }
728 :
729 6 : sc::result react(const EvHoldTimerExpired &event) {
730 6 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
731 6 : if (state_machine->HoldTimerCancelled()) {
732 0 : SM_LOG(state_machine,
733 : "Discard EvHoldTimerExpired in (OpenConfirm) State");
734 0 : return discard_event();
735 : }
736 6 : XmppConnectionInfo info;
737 6 : info.set_close_reason("Hold timer expired");
738 6 : state_machine->connection()->set_close_reason("Hold timer expired");
739 6 : if (state_machine->IsActiveChannel() ) {
740 4 : CloseSession(state_machine);
741 4 : state_machine->SendConnectionInfo(&info, event.Name(), "Active");
742 4 : return transit<Active>();
743 : } else {
744 2 : state_machine->ResetSession();
745 2 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
746 2 : return transit<Idle>();
747 : }
748 6 : }
749 :
750 : // received by the client
751 1469 : sc::result react(const EvStreamFeatureRequest &event) {
752 1469 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
753 1469 : XmppConnection *connection = state_machine->connection();
754 1469 : XmppSession *session = state_machine->session();
755 : // TODO, we need to have a supported stream feature list
756 : // and compare against the requested stream feature list
757 : // which will enable us to send start of various features
758 1469 : if (!connection->SendStartTls(session)) {
759 0 : connection->SendClose(session);
760 0 : state_machine->ResetSession();
761 0 : XmppConnectionInfo info;
762 0 : info.set_close_reason("Send Start Tls Failed");
763 0 : state_machine->SendConnectionInfo(&info, event.Name(), "Active");
764 0 : return transit<Active>();
765 0 : } else {
766 1469 : state_machine->StartHoldTimer();
767 1469 : state_machine->SendConnectionInfo(event.Name(),
768 : "Sent Start Tls, OpenConfirm Feature Negotiation");
769 1469 : state_machine->set_openconfirm_state(
770 : OPENCONFIRM_FEATURE_NEGOTIATION);
771 1469 : return discard_event();
772 : }
773 : }
774 :
775 : // received by client
776 1467 : sc::result react(const EvTlsProceed &event) {
777 1467 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
778 1467 : XmppConnection *connection = state_machine->connection();
779 1467 : XmppSession *session = state_machine->session();
780 1467 : state_machine->StartHoldTimer();
781 1467 : XmppConnectionInfo info;
782 1467 : info.set_identifier(connection->GetTo());
783 : // Trigger Ssl Handshake on client side
784 1467 : session->TriggerSslHandShake(state_machine->HandShakeCbHandler());
785 1467 : state_machine->SendConnectionInfo( &info, event.Name(),
786 : "Trigger Client Ssl Handshake");
787 2934 : return discard_event();
788 1467 : }
789 :
790 : //received by server
791 1464 : sc::result react(const EvStartTls &event) {
792 1464 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
793 1464 : if (event.session != state_machine->session()) {
794 0 : return discard_event();
795 : }
796 1464 : XmppConnection *connection = state_machine->connection();
797 1464 : XmppSession *session = state_machine->session();
798 1464 : XmppConnectionInfo info;
799 1464 : info.set_identifier(connection->GetTo());
800 1464 : if (!connection->SendProceedTls(session)) {
801 0 : connection->SendClose(session);
802 0 : state_machine->ResetSession();
803 0 : info.set_close_reason("Send Proceed Tls Failed");
804 0 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
805 0 : return transit<Idle>();
806 : } else {
807 1464 : state_machine->StartHoldTimer();
808 : // Trigger Ssl Handshake on server side
809 1464 : session->TriggerSslHandShake(state_machine->HandShakeCbHandler());
810 1464 : state_machine->SendConnectionInfo(&info, event.Name(),
811 : "Trigger Server Ssl Handshake");
812 1464 : return discard_event();
813 : }
814 1464 : }
815 :
816 1 : sc::result react(const EvStop &event) {
817 1 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
818 1 : state_machine->CancelHoldTimer();
819 1 : if (state_machine->IsActiveChannel() ) {
820 1 : CloseSession(state_machine);
821 1 : state_machine->SendConnectionInfo(event.Name(), "Active");
822 1 : return transit<Active>();
823 : } else {
824 0 : state_machine->ResetSession();
825 0 : state_machine->SendConnectionInfo(event.Name(), "Idle");
826 0 : return transit<Idle>();
827 : }
828 : }
829 :
830 2924 : sc::result react(const EvTlsHandShakeSuccess &event) {
831 2924 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
832 2924 : if (event.session != state_machine->session()) {
833 0 : return discard_event();
834 : }
835 2924 : XmppConnection *connection = state_machine->connection();
836 2924 : XmppSession *session = state_machine->session();
837 2924 : session->AsyncReadStart();
838 2924 : state_machine->set_openconfirm_state(OPENCONFIRM_FEATURE_SUCCESS);
839 2924 : if (state_machine->IsActiveChannel()) { //client
840 1462 : if (!connection->SendOpen(session)) {
841 0 : connection->SendClose(session);
842 0 : state_machine->ResetSession();
843 0 : XmppConnectionInfo info;
844 0 : info.set_close_reason("Open send failed in OpenConfirm State");
845 0 : state_machine->SendConnectionInfo(event.Name(), "Active");
846 0 : return transit<Active>();
847 0 : }
848 : }
849 : // both server and client
850 2924 : state_machine->StartHoldTimer();
851 2924 : state_machine->SendConnectionInfo(event.Name(),
852 : "OpenConfirm Feature Negotiation Success");
853 2924 : return discard_event();
854 : }
855 :
856 2 : sc::result react(const EvTlsHandShakeFailure &event) {
857 2 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
858 2 : if (event.session != state_machine->session()) {
859 0 : return discard_event();
860 : }
861 : // Do not send stream close as error occured at TLS layer
862 2 : state_machine->ResetSession();
863 2 : if (state_machine->IsActiveChannel()) { // client
864 1 : state_machine->SendConnectionInfo(event.Name(), "Active");
865 1 : return transit<Active>();
866 : } else {
867 1 : state_machine->SendConnectionInfo(event.Name(), "Idle");
868 1 : return transit<Idle>();
869 : }
870 : }
871 :
872 : //event on server and client
873 2919 : sc::result react(const EvXmppOpen &event) {
874 2919 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
875 2919 : if (event.session != state_machine->session()) {
876 0 : return discard_event();
877 : }
878 2919 : XmppConnectionInfo info;
879 2919 : info.set_identifier(event.msg->from);
880 2919 : XmppConnection *connection = state_machine->connection();
881 2919 : XmppSession *session = state_machine->session();
882 2919 : if (connection->IsActiveChannel()) { //client
883 1460 : connection->SendKeepAlive();
884 1460 : connection->StartKeepAliveTimer();
885 1460 : state_machine->StartHoldTimer();
886 1460 : state_machine->SendConnectionInfo(&info, event.Name(),
887 : "Established");
888 1460 : return transit<XmppStreamEstablished>();
889 : } else { //server
890 1459 : if (!connection->SendOpenConfirm(session)) {
891 0 : connection->SendClose(session);
892 0 : SM_LOG(state_machine,
893 : "Xmpp Send Open Confirm Failed, IDLE");
894 0 : state_machine->ResetSession();
895 0 : info.set_close_reason("Send Open Confirm Failed");
896 0 : state_machine->connection()->set_close_reason(
897 : "Send Open Confirm Failed");
898 0 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
899 0 : return transit<Idle>();
900 : } else {
901 1459 : if (state_machine->IsAuthEnabled())
902 1459 : state_machine->ResurrectOldConnection(connection, session);
903 1458 : connection = state_machine->connection();
904 1458 : connection->StartKeepAliveTimer();
905 1459 : state_machine->StartHoldTimer();
906 1459 : state_machine->SendConnectionInfo(&info, event.Name(),
907 : "Established");
908 1459 : return transit<XmppStreamEstablished>();
909 : }
910 : }
911 2919 : }
912 :
913 0 : sc::result react(const EvAdminDown &event) {
914 0 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
915 0 : CloseSession(state_machine);
916 0 : XmppConnectionInfo info;
917 0 : info.set_close_reason("Administratively down");
918 0 : state_machine->connection()->set_close_reason("Administratively down");
919 0 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
920 0 : return transit<Idle>();
921 0 : }
922 :
923 6 : void CloseSession(XmppStateMachine *state_machine) {
924 6 : XmppConnection *connection = state_machine->connection();
925 6 : if (connection != NULL) connection->StopKeepAliveTimer();
926 6 : state_machine->set_session(NULL);
927 6 : }
928 : };
929 :
930 : struct XmppStreamEstablished :
931 : public sc::state<XmppStreamEstablished, XmppStateMachine> {
932 : typedef mpl::list<
933 : sc::custom_reaction<EvAdminDown>,
934 : sc::custom_reaction<EvTcpClose>,
935 : sc::custom_reaction<EvXmppKeepalive>,
936 : sc::custom_reaction<EvXmppMessageStanza>,
937 : sc::custom_reaction<EvXmppIqStanza>,
938 : sc::custom_reaction<EvHoldTimerExpired>,
939 : sc::custom_reaction<EvStop>
940 : > reactions;
941 :
942 12605 : XmppStreamEstablished(my_context ctx) : my_base(ctx) {
943 12605 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
944 12605 : XmppConnection *connection = state_machine->connection();
945 12605 : state_machine->StartHoldTimer();
946 12605 : state_machine->set_state(ESTABLISHED);
947 12604 : connection->ChannelMux()->HandleStateEvent(xmsm::ESTABLISHED);
948 12605 : }
949 25198 : ~XmppStreamEstablished() {
950 12599 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
951 12595 : state_machine->CancelHoldTimer();
952 25200 : }
953 :
954 6105 : sc::result react(const EvTcpClose &event) {
955 6105 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
956 6106 : if (event.session != state_machine->session()) {
957 36 : return discard_event();
958 : }
959 6070 : state_machine->ResetSession();
960 6070 : if (state_machine->IsActiveChannel()) {
961 2574 : state_machine->SendConnectionInfo(event.Name(), "Active");
962 2574 : return transit<Active>();
963 : } else {
964 3496 : state_machine->SendConnectionInfo(event.Name(), "Idle");
965 3496 : return transit<Idle>();
966 : }
967 : }
968 :
969 2472467 : sc::result react(const EvXmppKeepalive &event) {
970 2472467 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
971 2472466 : if (event.session != state_machine->session()) {
972 0 : return discard_event();
973 : }
974 2472466 : state_machine->keepalive_count_inc();
975 2472466 : if (state_machine->get_keepalive_count() == 3) {
976 3012 : state_machine->connect_attempts_clear();
977 : }
978 2472466 : state_machine->StartHoldTimer();
979 2472467 : return discard_event();
980 : }
981 :
982 1556755 : sc::result react(const EvXmppMessageStanza &event) {
983 1556755 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
984 1556755 : if (event.session != state_machine->session()) {
985 0 : return discard_event();
986 : }
987 1556755 : state_machine->StartHoldTimer();
988 3113510 : state_machine->connection()->ProcessXmppChatMessage(
989 1556755 : static_cast<const XmppStanza::XmppChatMessage *>(event.msg.get()));
990 1556755 : return discard_event();
991 : }
992 :
993 56895 : sc::result react(const EvXmppIqStanza &event) {
994 56895 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
995 56895 : if (event.session != state_machine->session()) {
996 0 : return discard_event();
997 : }
998 56894 : state_machine->StartHoldTimer();
999 113790 : state_machine->connection()->ProcessXmppIqMessage(
1000 56895 : static_cast<const XmppStanza::XmppMessage *>(event.msg.get()));
1001 56894 : return discard_event();
1002 : }
1003 :
1004 5 : sc::result react(const EvHoldTimerExpired &event) {
1005 5 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
1006 5 : if (state_machine->HoldTimerCancelled()) {
1007 0 : SM_LOG(state_machine,
1008 : "Discard EvHoldTimerExpired in (Established) State");
1009 0 : return discard_event();
1010 : }
1011 5 : XMPP_NOTICE(XmppStateMachineNotice,
1012 : state_machine->connection()->ToUVEKey(),
1013 : XMPP_PEER_DIR_NA,
1014 : state_machine->ChannelType(),
1015 : "EvHoldTimerExpired in (Established) State. Transit to IDLE");
1016 5 : state_machine->SendConnectionInfo(event.Name());
1017 5 : state_machine->AssertOnHoldTimeout();
1018 5 : state_machine->ResetSession();
1019 5 : if (state_machine->IsActiveChannel()) {
1020 4 : return transit<Active>();
1021 : } else {
1022 1 : return transit<Idle>();
1023 : }
1024 : }
1025 :
1026 2796 : sc::result react(const EvStop &event) {
1027 2796 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
1028 2801 : state_machine->SendConnectionInfo(event.Name());
1029 2799 : state_machine->ResetSession();
1030 2803 : if (state_machine->IsActiveChannel()) {
1031 0 : return transit<Active>();
1032 : } else {
1033 2803 : return transit<Idle>();
1034 : }
1035 : }
1036 :
1037 3183 : sc::result react(const EvAdminDown &event) {
1038 3183 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
1039 3183 : state_machine->ResetSession();
1040 3183 : XmppConnectionInfo info;
1041 3183 : info.set_close_reason("Administratively down");
1042 3183 : state_machine->connection()->set_close_reason("Administratively down");
1043 3183 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
1044 6366 : return transit<Idle>();
1045 3183 : }
1046 : };
1047 :
1048 : } // namespace xmsm
1049 :
1050 5 : void XmppStateMachine::AssertOnHoldTimeout() {
1051 : static bool init_ = false;
1052 : static bool assert_ = false;
1053 :
1054 5 : if (!init_) {
1055 3 : char *str = getenv("XMPP_ASSERT_ON_HOLD_TIMEOUT");
1056 3 : if (str && strtoul(str, NULL, 0) != 0) assert_ = true;
1057 3 : init_ = true;
1058 : }
1059 :
1060 5 : if (!assert_) return;
1061 :
1062 0 : if (connection()) {
1063 0 : connection()->LogMsg("HOLD TIMER EXPIRED: ");
1064 : } else {
1065 0 : LOG4CPLUS_DEBUG(log4cplus::Logger::getRoot(), "HOLD TIMER EXPIRED: ");
1066 : }
1067 :
1068 0 : assert(!assert_);
1069 : }
1070 :
1071 12351 : void XmppStateMachine::ResetSession() {
1072 12351 : XmppConnection *connection = this->connection();
1073 12350 : set_session(NULL);
1074 12358 : CancelHoldTimer();
1075 :
1076 12357 : if (!connection)
1077 0 : return;
1078 :
1079 : // Stop keepalives, transition to IDLE and notify registered entities.
1080 12357 : connection->StopKeepAliveTimer();
1081 12358 : connection->ChannelMux()->HandleStateEvent(xmsm::IDLE);
1082 12358 : if (IsActiveChannel())
1083 5762 : return;
1084 :
1085 : // Retain the connection if graceful restart is supported.
1086 6596 : XmppServer *server = dynamic_cast<XmppServer *>(connection->server());
1087 6596 : if (!server->IsPeerCloseGraceful())
1088 6383 : connection->ManagedDelete();
1089 : }
1090 :
1091 11606 : XmppStateMachine::XmppStateMachine(XmppConnection *connection, bool active,
1092 11606 : bool auth_enabled, int config_hold_time)
1093 11606 : : work_queue_(TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
1094 : connection->GetTaskInstance(),
1095 : boost::bind(&XmppStateMachine::DequeueEvent, this, _1)),
1096 11606 : connection_(connection),
1097 11606 : session_(NULL),
1098 11606 : server_(connection->server()),
1099 11606 : connect_timer_(
1100 11606 : TimerManager::CreateTimer(*server_->event_manager()->io_service(),
1101 : "Connect timer",
1102 : TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
1103 : connection->GetTaskInstance())),
1104 11606 : open_timer_(
1105 11606 : TimerManager::CreateTimer(*server_->event_manager()->io_service(),
1106 : "Open timer",
1107 : TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
1108 : connection->GetTaskInstance())),
1109 11606 : hold_timer_(
1110 11606 : TimerManager::CreateTimer(*server_->event_manager()->io_service(),
1111 : "Hold timer",
1112 : TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
1113 : connection->GetTaskInstance())),
1114 11606 : config_hold_time_(config_hold_time),
1115 11606 : hold_time_(GetConfiguredHoldTime()),
1116 11606 : attempts_(0),
1117 11606 : keepalive_count_(0),
1118 11606 : deleted_(false),
1119 11606 : in_dequeue_(false),
1120 11606 : is_active_(active),
1121 11606 : auth_enabled_(auth_enabled),
1122 11606 : state_(xmsm::IDLE),
1123 11606 : last_state_(xmsm::IDLE),
1124 23212 : openconfirm_state_(xmsm::OPENCONFIRM_INIT) {
1125 : handshake_cb_ = boost::bind(
1126 11606 : &XmppConnection::ProcessSslHandShakeResponse, connection, _1, _2);
1127 11606 : }
1128 :
1129 11701 : XmppStateMachine::~XmppStateMachine() {
1130 11606 : assert(!deleted_);
1131 11606 : deleted_ = true;
1132 11606 : work_queue_.Shutdown();
1133 11606 : set_session(NULL);
1134 :
1135 : // Explicitly call the state destructor before the state machine itself.
1136 : // This is needed because some of the destructors access the state machine
1137 : // context.
1138 11606 : terminate();
1139 :
1140 : // Delete timer after state machine is terminated so that there is no
1141 : // possible reference to the timers being deleted any more
1142 11606 : TimerManager::DeleteTimer(connect_timer_);
1143 11606 : TimerManager::DeleteTimer(open_timer_);
1144 11606 : TimerManager::DeleteTimer(hold_timer_);
1145 11701 : }
1146 :
1147 11836 : void XmppStateMachine::Initialize() {
1148 11836 : initiate();
1149 11836 : Enqueue(xmsm::EvStart());
1150 11836 : }
1151 :
1152 : // Note this api does not enqueue the deletion of TCP session
1153 11606 : void XmppStateMachine::clear_session() {
1154 11606 : if (session_ != NULL) {
1155 2006 : session_->set_observer(NULL);
1156 2006 : session_->ClearConnection();
1157 2006 : session_->Close();
1158 2006 : connection_->clear_session();
1159 2006 : session_ = NULL;
1160 : }
1161 11606 : }
1162 :
1163 12306 : void XmppStateMachine::DeleteSession(XmppSession *session) {
1164 : // If there is a session assigned to this state machine, reset the observer
1165 : // so that tcp does not have a reference to 'this' which is going away
1166 12306 : if (session != NULL) {
1167 12306 : session->set_observer(NULL);
1168 12301 : session->ClearConnection();
1169 12301 : session->Close();
1170 12306 : Enqueue(xmsm::EvTcpDeleteSession(session));
1171 : }
1172 12306 : }
1173 :
1174 39689 : void XmppStateMachine::set_session(TcpSession *session) {
1175 39689 : if (session_ != NULL) {
1176 12299 : connection_->clear_session();
1177 12306 : DeleteSession(session_);
1178 : }
1179 39696 : session_ = static_cast<XmppSession *>(session);
1180 39696 : }
1181 :
1182 0 : void XmppStateMachine::TimerErrorHandler(std::string name, std::string error) {
1183 0 : }
1184 :
1185 165 : void XmppStateMachine::StartConnectTimer(int seconds) {
1186 165 : CancelConnectTimer();
1187 :
1188 : // Add up to +/- kJitter percentage to reduce connection collisions.
1189 165 : int ms = ((seconds)? seconds * 1000 : 50);
1190 165 : ms = (ms * (100 - kJitter)) / 100;
1191 165 : ms += (ms * (rand() % (kJitter * 2))) / 100;
1192 165 : connect_timer_->Start(ms,
1193 : boost::bind(&XmppStateMachine::ConnectTimerExpired, this),
1194 : boost::bind(&XmppStateMachine::TimerErrorHandler, this, _1, _2));
1195 165 : }
1196 :
1197 31700 : void XmppStateMachine::CancelConnectTimer() {
1198 31700 : connect_timer_->Cancel();
1199 31699 : }
1200 :
1201 14 : void XmppStateMachine::StartOpenTimer(int seconds) {
1202 14 : CancelOpenTimer();
1203 14 : open_timer_->Start(seconds * 1000,
1204 : boost::bind(&XmppStateMachine::OpenTimerExpired, this),
1205 : boost::bind(&XmppStateMachine::TimerErrorHandler, this, _1, _2));
1206 14 : }
1207 :
1208 22599 : void XmppStateMachine::CancelOpenTimer() {
1209 22599 : open_timer_->Cancel();
1210 22599 : }
1211 :
1212 12093 : int XmppStateMachine::GetConfiguredHoldTime() const {
1213 : static std::atomic<bool> env_checked = std::atomic<bool>();
1214 : static std::atomic<int> env_hold_time = std::atomic<int>();
1215 :
1216 : // For testing only - configure through environment variable.
1217 12093 : if (!env_checked) {
1218 65 : char *keepalive_time_str = getenv("XMPP_KEEPALIVE_SECONDS");
1219 65 : if (keepalive_time_str) {
1220 0 : env_hold_time = strtoul(keepalive_time_str, NULL, 0) * 3;
1221 0 : env_checked = true;
1222 0 : return env_hold_time;
1223 : } else {
1224 65 : env_checked = true;
1225 : }
1226 12028 : } else if (env_hold_time) {
1227 0 : return env_hold_time;
1228 12028 : } else if (config_hold_time_) {
1229 12028 : return config_hold_time_;
1230 : }
1231 :
1232 : // Use hard coded default.
1233 65 : return kHoldTime;
1234 : }
1235 :
1236 4123368 : void XmppStateMachine::StartHoldTimer() {
1237 4123368 : CancelHoldTimer();
1238 :
1239 4123364 : if (hold_time_ <= 0)
1240 0 : return;
1241 :
1242 4123364 : hold_timer_->Start(hold_time_ * 1000,
1243 : boost::bind(&XmppStateMachine::HoldTimerExpired, this),
1244 : boost::bind(&XmppStateMachine::TimerErrorHandler, this, _1, _2));
1245 : }
1246 :
1247 4155128 : void XmppStateMachine::CancelHoldTimer() {
1248 4155128 : hold_timer_->Cancel();
1249 4155127 : }
1250 :
1251 12051 : void XmppStateMachine::OnStart(const xmsm::EvStart &event) {
1252 12051 : }
1253 :
1254 7747 : bool XmppStateMachine::ConnectTimerExpired() {
1255 7747 : XMPP_UTDEBUG(XmppStateMachineTimerExpire,
1256 : connection() ? connection()->ToUVEKey() : "",
1257 : XMPP_PEER_DIR_NA, this->ChannelType(), "Connect", StateName());
1258 7747 : Enqueue(xmsm::EvConnectTimerExpired());
1259 7747 : return false;
1260 : }
1261 :
1262 1 : bool XmppStateMachine::OpenTimerExpired() {
1263 1 : XMPP_NOTICE(XmppEventLog, connection()->ToUVEKey(), XMPP_PEER_DIR_NA,
1264 : this->ChannelType(),
1265 : "Event: OpenTimer Expired ",
1266 : connection()->endpoint().address().to_string(),
1267 : connection()->GetTo());
1268 1 : Enqueue(xmsm::EvOpenTimerExpired());
1269 1 : return false;
1270 : }
1271 :
1272 :
1273 13 : bool XmppStateMachine::HoldTimerExpired() {
1274 13 : boost::system::error_code error;
1275 :
1276 : // Reset hold timer if there is data already present in the socket.
1277 26 : if (session() && session()->socket() &&
1278 13 : session()->socket()->available(error) > 0) {
1279 0 : return true;
1280 : }
1281 13 : XMPP_NOTICE(XmppEventLog, connection()->ToUVEKey(), XMPP_PEER_DIR_NA,
1282 : this->ChannelType(), "Event: HoldTimer Expired ",
1283 : connection()->endpoint().address().to_string(),
1284 : connection()->GetTo());
1285 13 : Enqueue(xmsm::EvHoldTimerExpired());
1286 13 : return false;
1287 : }
1288 :
1289 20792 : void XmppStateMachine::OnSessionEvent(
1290 : TcpSession *session, TcpSession::Event event) {
1291 20792 : switch (event) {
1292 6544 : case TcpSession::ACCEPT:
1293 6544 : break;
1294 6583 : case TcpSession::CONNECT_COMPLETE:
1295 6583 : XMPP_NOTICE(XmppEventLog, session->ToUVEKey(), XMPP_PEER_DIR_IN,
1296 : this->ChannelType(), "Event: Tcp Connected ",
1297 : connection()->endpoint().address().to_string(),
1298 : connection()->GetTo());
1299 6583 : Enqueue(xmsm::EvTcpConnected(static_cast<XmppSession *>(session)));
1300 6583 : break;
1301 1159 : case TcpSession::CONNECT_FAILED:
1302 1159 : XMPP_NOTICE(XmppEventLog, session->ToUVEKey(), XMPP_PEER_DIR_IN,
1303 : this->ChannelType(), "Event: Tcp Connect Fail ",
1304 : connection()->endpoint().address().to_string(),
1305 : connection()->GetTo());
1306 1159 : Enqueue(xmsm::EvTcpConnectFail(static_cast<XmppSession *>(session)));
1307 1159 : connection_->inc_connect_error();
1308 1159 : break;
1309 6506 : case TcpSession::CLOSE:
1310 6506 : XMPP_NOTICE(XmppEventLog, session->ToUVEKey(), XMPP_PEER_DIR_IN,
1311 : this->ChannelType(), "Event: Tcp Connection Closed ",
1312 : connection()->endpoint().address().to_string(),
1313 : connection()->GetTo());
1314 6506 : Enqueue(xmsm::EvTcpClose(static_cast<XmppSession *>(session)));
1315 6506 : connection_->inc_session_close();
1316 6506 : break;
1317 0 : default:
1318 0 : XMPP_WARNING(XmppUnknownEvent, session->ToUVEKey(), XMPP_PEER_DIR_IN,
1319 : this->ChannelType(), event);
1320 0 : break;
1321 : }
1322 20792 : }
1323 :
1324 6521 : bool XmppStateMachine::PassiveOpen(XmppSession *session) {
1325 6521 : string state = "PassiveOpen in state: " + StateName();
1326 6521 : XMPP_NOTICE(XmppEventLog, session->ToUVEKey(), XMPP_PEER_DIR_IN,
1327 : this->ChannelType(), state,
1328 : session->Connection()->endpoint().address().to_string(), "");
1329 13042 : return Enqueue(xmsm::EvTcpPassiveOpen(session));
1330 6521 : }
1331 :
1332 : // Process XmppStream header message received over a session. Close the stream
1333 : // if an old session is still present and undergoing graceful restart.
1334 : //
1335 : // Return true if msg is enqueued for further processing, false otherwise.
1336 15770 : void XmppStateMachine::ProcessStreamHeaderMessage(XmppSession *session,
1337 : const XmppStanza::XmppMessage *msg) {
1338 : XmppConnectionManager *connection_manager =
1339 15770 : dynamic_cast<XmppConnectionManager *>(connection_->server());
1340 15770 : std::scoped_lock lock(connection_manager->mutex());
1341 :
1342 : // Update "To" information which can be used to map an older session
1343 15771 : session->Connection()->SetTo(msg->from);
1344 :
1345 15771 : XmppServer *xmpp_server = dynamic_cast<XmppServer *>(server_);
1346 15771 : XmppConnectionEndpoint *endpoint = NULL;
1347 :
1348 : // Look for an endpoint which may already exist
1349 15771 : if (xmpp_server) {
1350 : endpoint =
1351 7999 : xmpp_server->FindConnectionEndpoint(connection_->ToString());
1352 :
1353 16006 : if (!xmpp_server->subcluster_name().empty() &&
1354 8007 : xmpp_server->subcluster_name() != msg->xmlns) {
1355 2 : string reason = "Subcluster mismatch: Agent subcluster " +
1356 4 : msg->xmlns + ", Control subcluster " +
1357 4 : xmpp_server->subcluster_name();
1358 2 : XMPP_WARNING(XmppDeleteConnection, session->ToUVEKey(),
1359 : XMPP_PEER_DIR_IN,
1360 : "Drop new xmpp connection " + session->ToString() +
1361 : " as " + reason);
1362 2 : session->Connection()->set_close_reason(reason);
1363 2 : ProcessEvent(xmsm::EvTcpClose(session));
1364 2 : delete msg;
1365 2 : return;
1366 2 : }
1367 : }
1368 :
1369 : // If there is no connection already associated with the end-point,
1370 : // process the incoming open message and move forward with the session
1371 : // establishment.
1372 16105 : if (!endpoint || !endpoint->connection() ||
1373 336 : connection_ == endpoint->connection()) {
1374 15434 : ProcessEvent(xmsm::EvXmppOpen(session, msg));
1375 15434 : return;
1376 : }
1377 :
1378 : // Check if the IP addresses match.
1379 : boost::asio::ip::address addr =
1380 335 : endpoint->connection()->endpoint().address();
1381 335 : if (connection_->endpoint().address() != addr) {
1382 36 : XMPP_WARNING(XmppDeleteConnection, session->ToUVEKey(),
1383 : XMPP_PEER_DIR_IN,
1384 : "Drop new xmpp connection " + session->ToString() +
1385 : " as another connection with same name " + msg->from +
1386 : " but with different IP address " + addr.to_string() +
1387 : " already exists");
1388 36 : ProcessEvent(xmsm::EvTcpClose(session));
1389 36 : delete msg;
1390 36 : return;
1391 : }
1392 :
1393 299 : XmppChannelMux *channel = endpoint->connection()->ChannelMux();
1394 :
1395 : // If GR Helper mode is not enabled, ignore the new connection request.
1396 : // Existing connection, if down will get cleaned up eventually. If it is
1397 : // up, it shall remain intact.
1398 299 : if (!xmpp_server->IsPeerCloseGraceful()) {
1399 125 : XMPP_NOTICE(XmppDeleteConnection, session->ToUVEKey(),
1400 : XMPP_PEER_DIR_IN,
1401 : "Drop new xmpp connection " + session->ToString() +
1402 : " as another connection is alreready present");
1403 125 : ProcessEvent(xmsm::EvTcpClose(session));
1404 125 : delete msg;
1405 125 : return;
1406 : }
1407 :
1408 174 : XmppStateMachine *old_sm = endpoint->connection()->state_machine();
1409 :
1410 : // Bring down old session if connection is already up and ready. This is
1411 : // the scenario in which old session's TCP did not learn the session down
1412 : // event, possibly due to compute cold reboot. In that case, trigger
1413 : // closure (and possibly GR) process for the old session.
1414 174 : if (channel->GetPeerState() == xmps::READY) {
1415 0 : XMPP_NOTICE(XmppDeleteConnection, old_sm->session()->ToUVEKey(),
1416 : XMPP_PEER_DIR_IN, "Delete old xmpp connection " +
1417 : old_sm->session()->ToString() +
1418 : " as a new connection as been initiated (GR Helper is active)");
1419 0 : old_sm->Enqueue(xmsm::EvTcpClose(old_sm->session()));
1420 :
1421 : // Drop the new session until old one is deleted or marked stale.
1422 0 : ProcessEvent(xmsm::EvTcpClose(session));
1423 0 : delete msg;
1424 0 : return;
1425 : }
1426 :
1427 : // If previous closure is still in progress, drop the new connection.
1428 174 : if (channel->ReceiverCount()) {
1429 18 : XMPP_NOTICE(XmppDeleteConnection, session->ToUVEKey(),
1430 : XMPP_PEER_DIR_IN,
1431 : "Drop new xmpp connection " + session->ToString() +
1432 : " as current connection is still under deletion");
1433 18 : ProcessEvent(xmsm::EvTcpClose(session));
1434 18 : delete msg;
1435 18 : return;
1436 : }
1437 :
1438 : // Now we reach for the classic GR case, in which existing session stale
1439 : // process is complete, (peer is in GR/LLGR timer wait state).
1440 : // In this case, we should process the open message, and later switch to
1441 : // the new connection and state machine, retaining the old XmppPeer,
1442 : // BgpXmppChannel, routes, etc.
1443 156 : ProcessEvent(xmsm::EvXmppOpen(session, msg));
1444 15771 : }
1445 :
1446 4110876 : void XmppStateMachine::OnMessage(XmppSession *session,
1447 : const XmppStanza::XmppMessage *msg) {
1448 4110876 : if (!Enqueue(xmsm::EvXmppMessage(session, msg)))
1449 0 : delete msg;
1450 4110882 : }
1451 :
1452 4110882 : void XmppStateMachine::ProcessMessage(XmppSession *session,
1453 : const XmppStanza::XmppMessage *msg) {
1454 : // Bail if session is already reset and disassociated from the connection.
1455 4110882 : if (!session->Connection()) {
1456 229 : delete msg;
1457 232 : return;
1458 : }
1459 :
1460 4110655 : const XmppStanza::XmppStreamMessage *stream_msg =
1461 : static_cast<const XmppStanza::XmppStreamMessage *>(msg);
1462 :
1463 4110655 : switch (msg->type) {
1464 20170 : case XmppStanza::STREAM_HEADER:
1465 20170 : if (stream_msg->strmtype ==
1466 : XmppStanza::XmppStreamMessage::FEATURE_TLS) {
1467 :
1468 4400 : switch (stream_msg->strmtlstype) {
1469 1469 : case (XmppStanza::XmppStreamMessage::TLS_FEATURE_REQUEST):
1470 1469 : ProcessEvent(xmsm::EvStreamFeatureRequest(session,
1471 : msg));
1472 1469 : break;
1473 1464 : case (XmppStanza::XmppStreamMessage::TLS_START):
1474 1464 : ProcessEvent(xmsm::EvStartTls(session, msg));
1475 1464 : break;
1476 1467 : case (XmppStanza::XmppStreamMessage::TLS_PROCEED):
1477 1467 : ProcessEvent(xmsm::EvTlsProceed(session, msg));
1478 1467 : break;
1479 0 : default:
1480 0 : break;
1481 : }
1482 :
1483 15770 : } else if (stream_msg->strmtype ==
1484 0 : XmppStanza::XmppStreamMessage::INIT_STREAM_HEADER ||
1485 0 : stream_msg->strmtype ==
1486 : XmppStanza::XmppStreamMessage::INIT_STREAM_HEADER_RESP)
1487 15770 : ProcessStreamHeaderMessage(session, msg);
1488 20171 : break;
1489 2476835 : case XmppStanza::WHITESPACE_MESSAGE_STANZA:
1490 2476835 : ProcessEvent(xmsm::EvXmppKeepalive(session, msg));
1491 2476835 : break;
1492 56895 : case XmppStanza::IQ_STANZA:
1493 56895 : ProcessEvent(xmsm::EvXmppIqStanza(session, msg));
1494 56894 : break;
1495 1556755 : case XmppStanza::MESSAGE_STANZA:
1496 1556755 : ProcessEvent(xmsm::EvXmppMessageStanza(session, msg));
1497 1556755 : break;
1498 0 : default:
1499 0 : if (!msg->IsValidType(msg->type)) {
1500 0 : XMPP_NOTICE(XmppStateMachineUnsupportedMessage,
1501 : XMPP_PEER_DIR_IN, session->ToUVEKey(),
1502 : ChannelType(), (int)msg->type);
1503 : }
1504 0 : delete msg;
1505 0 : break;
1506 : }
1507 :
1508 : }
1509 :
1510 9131 : void XmppStateMachine::Clear() {
1511 9131 : Enqueue(xmsm::EvStop());
1512 9131 : }
1513 :
1514 3767 : void XmppStateMachine::SetAdminState(bool down) {
1515 3767 : assert(IsActiveChannel());
1516 3767 : if (down) {
1517 3508 : Enqueue(xmsm::EvAdminDown());
1518 : } else {
1519 259 : Enqueue(xmsm::EvStart());
1520 : }
1521 3767 : }
1522 :
1523 2926 : void XmppStateMachine::OnEvent(SslSession *session,
1524 : xmsm::SslHandShakeResponse resp) {
1525 2926 : XmppSession *sess = static_cast<XmppSession *>(session);
1526 2926 : switch(resp) {
1527 2 : case xmsm::EvTLSHANDSHAKE_FAILURE:
1528 2 : Enqueue(xmsm::EvTlsHandShakeFailure(sess));
1529 2 : break;
1530 2924 : case xmsm::EvTLSHANDSHAKE_SUCCESS:
1531 2924 : Enqueue(xmsm::EvTlsHandShakeSuccess(sess));
1532 2924 : break;
1533 0 : default:
1534 0 : break;
1535 : }
1536 2926 : }
1537 :
1538 67673 : void XmppStateMachine::set_state(xmsm::XmState state) {
1539 67673 : last_state_ = state_;
1540 67673 : state_ = state;
1541 67673 : state_since_ = UTCTimestampUsec();
1542 :
1543 67680 : if (!logUVE()) return;
1544 :
1545 0 : XmppPeerInfoData peer_info;
1546 0 : peer_info.set_name(connection()->ToUVEKey());
1547 0 : PeerStateInfo state_info;
1548 0 : state_info.set_state(StateName());
1549 0 : state_info.set_last_state(LastStateName());
1550 0 : state_info.set_last_state_at(state_since_);
1551 0 : peer_info.set_state_info(state_info);
1552 0 : assert(!peer_info.get_name().empty());
1553 0 : XMPPPeerInfo::Send(peer_info);
1554 0 : }
1555 :
1556 :
1557 7377 : void XmppStateMachine::set_openconfirm_state(xmsm::XmOpenConfirmState state) {
1558 7377 : openconfirm_state_ = state;
1559 7377 : }
1560 :
1561 : static const char *state_names[] = {
1562 : "Idle",
1563 : "Active",
1564 : "Connect",
1565 : "OpenSent",
1566 : "OpenConfirm",
1567 : "Established" };
1568 :
1569 4076034 : string XmppStateMachine::StateName() const {
1570 4076034 : return state_names[state_];
1571 : }
1572 :
1573 565 : string XmppStateMachine::LastStateName() const {
1574 565 : return state_names[last_state_];
1575 : }
1576 :
1577 559 : string XmppStateMachine::LastStateChangeAt() const {
1578 1118 : return integerToString(UTCUsecToPTime(state_since_));
1579 : }
1580 :
1581 7877966 : xmsm::XmState XmppStateMachine::StateType() const {
1582 7877966 : return state_;
1583 : }
1584 :
1585 4346598 : xmsm::XmOpenConfirmState XmppStateMachine::OpenConfirmStateType() const {
1586 4346598 : return openconfirm_state_;
1587 : }
1588 :
1589 12670 : void XmppStateMachine::AssignSession() {
1590 12670 : connection_->set_session(static_cast<XmppSession *>(session_));
1591 12670 : }
1592 :
1593 : const int XmppStateMachine::kConnectInterval;
1594 :
1595 16927 : int XmppStateMachine::GetConnectTime() const {
1596 16927 : int backoff = attempts_ > 6 ? 6 : attempts_;
1597 16927 : return std::min(backoff ? 1 << (backoff - 1) : 0, kConnectInterval);
1598 : }
1599 :
1600 4072288 : bool XmppStateMachine::IsActiveChannel() {
1601 4072288 : return is_active_;
1602 : }
1603 :
1604 4234081 : bool XmppStateMachine::logUVE() {
1605 4234081 : return connection()->logUVE();
1606 : }
1607 :
1608 3999495 : const char *XmppStateMachine::ChannelType() {
1609 3999495 : return (IsActiveChannel() ? " Mode Client: " : " Mode Server: " );
1610 : }
1611 :
1612 4178939 : bool XmppStateMachine::DequeueEvent(
1613 : boost::intrusive_ptr<const sc::event_base> event) {
1614 : // Process message event and enqueue additional events as necessary.
1615 : const xmsm::EvXmppMessage *ev_xmpp_message =
1616 4178939 : dynamic_cast<const xmsm::EvXmppMessage *>(event.get());
1617 4178935 : if (ev_xmpp_message) {
1618 4110882 : ProcessMessage(ev_xmpp_message->session, ev_xmpp_message->msg);
1619 : } else {
1620 68053 : ProcessEvent(*event);
1621 68045 : event.reset();
1622 : }
1623 4178952 : return true;
1624 : }
1625 :
1626 4178689 : void XmppStateMachine::ProcessEvent(const sc::event_base &event) {
1627 4178689 : const xmsm::EvTcpDeleteSession *deferred_delete =
1628 4178689 : dynamic_cast<const xmsm::EvTcpDeleteSession *>(&event);
1629 4178689 : XMPP_UTDEBUG(XmppStateMachineDequeueEvent,
1630 : connection() ? connection()->ToUVEKey() : "",
1631 : XMPP_PEER_DIR_IN, ChannelType(), TYPE_NAME(event), StateName(),
1632 : connection() ?
1633 : connection()->endpoint().address().to_string() : "",
1634 : connection() ? connection()->GetTo() : "");
1635 4178715 : if (deferred_delete) {
1636 12304 : TcpSession *session = deferred_delete->session;
1637 12304 : session->server()->DeleteSession(session);
1638 12305 : return;
1639 : }
1640 4166411 : if (deleted_) {
1641 0 : return;
1642 : }
1643 :
1644 4166411 : update_last_event(TYPE_NAME(event));
1645 4166404 : in_dequeue_ = true;
1646 4166404 : process_event(event);
1647 4166389 : in_dequeue_ = false;
1648 : }
1649 :
1650 10568 : void XmppStateMachine::unconsumed_event(const sc::event_base &event) {
1651 10568 : XMPP_INFO(XmppUnconsumedEvent, connection() ? connection()->ToUVEKey() : "",
1652 : XMPP_PEER_DIR_IN, ChannelType(), TYPE_NAME(event), StateName());
1653 10567 : }
1654 :
1655 4166408 : void XmppStateMachine::update_last_event(const std::string &event) {
1656 4166408 : set_last_event(event);
1657 :
1658 4166413 : if (!logUVE()) return;
1659 :
1660 : // Skip iq and message events.
1661 1 : if (event == "xmsm::EvXmppIqStanza" ||
1662 0 : event == "xmsm::EvXmppMessageStanza") {
1663 0 : return;
1664 : }
1665 :
1666 : // Skip keepalive events after we've reached established state.
1667 0 : if (state_ == xmsm::ESTABLISHED && event == "xmsm::EvXmppKeepalive") {
1668 0 : return;
1669 : }
1670 :
1671 0 : XmppPeerInfoData peer_info;
1672 0 : peer_info.set_name(connection()->ToUVEKey());
1673 0 : PeerEventInfo event_info;
1674 0 : event_info.set_last_event(last_event_);
1675 0 : event_info.set_last_event_at(last_event_at_);
1676 0 : peer_info.set_event_info(event_info);
1677 :
1678 0 : assert(!peer_info.get_name().empty());
1679 0 : XMPPPeerInfo::Send(peer_info);
1680 0 : }
1681 :
1682 : //
1683 : // Enqueue an event to xmpp state machine.
1684 : // Return false if the event is not enqueued.
1685 : //
1686 4179363 : bool XmppStateMachine::Enqueue(const sc::event_base &event) {
1687 4179363 : if (!deleted_) {
1688 4179361 : work_queue_.Enqueue(event.intrusive_from_this());
1689 4179382 : return true;
1690 : }
1691 :
1692 2 : return false;
1693 : }
1694 :
1695 : // Object trace routines.
1696 44702 : void XmppStateMachine::SendConnectionInfo(const string &event,
1697 : const string &nextstate) {
1698 44702 : XmppConnectionInfo info;
1699 44691 : SendConnectionInfo(&info, event, nextstate);
1700 89430 : return;
1701 44715 : }
1702 :
1703 87422 : void XmppStateMachine::SendConnectionInfo(XmppConnectionInfo *info,
1704 : const string &event, const string &nextstate) {
1705 :
1706 87422 : info->set_ip_address(this->connection()->endpoint().address().to_string());
1707 87450 : info->set_state(StateName());
1708 87442 : info->set_event(event);
1709 87444 : if (!nextstate.empty()) {
1710 70089 : info->set_next_state(nextstate);
1711 : }
1712 87444 : XMPP_CONNECTION_LOG_MSG(*info);
1713 87447 : return;
1714 : }
1715 :
1716 : // Resurrect an old xmpp connection if present (when under GracefulRestart)
1717 : //
1718 : // During Graceful Restart (or otherwise), new connections are rejected in
1719 : // ProcessStreamHeaderMessage() itself until old one's cleanup process is
1720 : // complete and the system is ready to start a new session.
1721 : //
1722 : // Hence in here, when called upon receipt of OpenMessage, we can try to reuse
1723 : // old XmppConnection if present and there by complete any pending GR process
1724 : //
1725 : // We do so by reusing XmppConnection, XmppChannel, etc. from the old connection
1726 : // and only use the XmppSession and XmppStateMachine from the new session
1727 : //
1728 : // New connection is instead associated with the old state machine and session,
1729 : // and their deletion is triggered
1730 6305 : void XmppStateMachine::ResurrectOldConnection(XmppConnection *new_connection,
1731 : XmppSession *new_session) {
1732 :
1733 : // Look for an endpoint (which is a persistent data structure) across
1734 : // xmpp session flips
1735 : bool created;
1736 : XmppConnectionEndpoint *connection_endpoint =
1737 : static_cast<XmppServer *>(
1738 6305 : new_connection->server())->LocateConnectionEndpoint(
1739 : static_cast<XmppServerConnection *>(new_connection), created);
1740 :
1741 : // If this is a new endpoint, then there is no older connection to manage.
1742 6304 : if (created)
1743 6200 : return;
1744 :
1745 : // GR Helper must be enabled when we are trying to resurrect connection.
1746 104 : XmppServer *server = dynamic_cast<XmppServer *>(new_connection->server());
1747 104 : assert(server->IsPeerCloseGraceful());
1748 :
1749 104 : XMPP_NOTICE(XmppCreateConnection, new_session->ToUVEKey(),
1750 : XMPP_PEER_DIR_IN,
1751 : "Resurrect xmpp connection " + new_session->ToString());
1752 :
1753 : // Retrieve old XmppConnection and XmppStateMachine (to reuse)
1754 104 : XmppConnection *old_xmpp_connection = connection_endpoint->connection();
1755 104 : assert(old_xmpp_connection);
1756 :
1757 104 : XmppStateMachine *old_state_machine = old_xmpp_connection->state_machine();
1758 104 : assert(old_state_machine);
1759 :
1760 : // Swap Old and New connections and state machines linkages
1761 104 : new_connection->SwapXmppStateMachine(old_xmpp_connection);
1762 104 : this->SwapXmppConnection(old_state_machine);
1763 :
1764 : // Update XmppConnection in the old session.
1765 104 : XmppSession *old_xmpp_session = old_state_machine->session();
1766 104 : if (old_xmpp_session)
1767 0 : old_xmpp_session->SetConnection(new_connection);
1768 104 : new_session->SetConnection(old_xmpp_connection);
1769 :
1770 : // Set new session with the old connection as it would be the current active
1771 : // connection from now on.
1772 104 : old_xmpp_connection->set_session(new_session);
1773 :
1774 : // Trigger deletion of the new connection which now is associated wth the
1775 : // the old_state_machine
1776 104 : new_connection->Shutdown();
1777 : }
1778 :
1779 104 : void XmppStateMachine::SwapXmppConnection(XmppStateMachine *other) {
1780 104 : swap(connection_, other->connection_);
1781 104 : connection_->SwapContents(other->connection_);
1782 104 : }
|
