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