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 238 : static const char *Name() {
58 238 : return "EvStop";
59 : }
60 : };
61 :
62 : struct EvAdminDown : sc::event<EvAdminDown> {
63 48 : static const char *Name() {
64 48 : return "EvAdminDown";
65 : }
66 : };
67 :
68 : struct EvConnectTimerExpired : sc::event<EvConnectTimerExpired> {
69 418 : static const char *Name() {
70 418 : 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 317 : EvTcpConnected(XmppSession *session) : session(session) { };
88 317 : static const char *Name() {
89 317 : return "EvTcpConnected";
90 : }
91 : XmppSession *session;
92 : } ;
93 :
94 : struct EvTcpConnectFail : sc::event<EvTcpConnectFail> {
95 96 : EvTcpConnectFail(XmppSession *session) : session(session) { };
96 96 : static const char *Name() {
97 96 : return "EvTcpConnectFail";
98 : }
99 : XmppSession *session;
100 : } ;
101 :
102 : struct EvTcpPassiveOpen : sc::event<EvTcpPassiveOpen> {
103 279 : EvTcpPassiveOpen(XmppSession *session) : session(session) { };
104 279 : static const char *Name() {
105 279 : return "EvTcpPassiveOpen";
106 : }
107 : XmppSession *session;
108 : };
109 :
110 : struct EvTcpClose : sc::event<EvTcpClose> {
111 306 : EvTcpClose(XmppSession *session) : session(session) { };
112 290 : static const char *Name() {
113 290 : return "EvTcpClose";
114 : }
115 : XmppSession *session;
116 : };
117 :
118 : struct EvTcpDeleteSession : sc::event<EvTcpDeleteSession> {
119 534 : 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 602728 : explicit EvXmppMessage(XmppSession *session,
128 602728 : 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 842 : EvXmppOpen(XmppSession *session, const XmppStanza::XmppMessage *msg) :
138 842 : session(session),
139 842 : msg(static_cast<const XmppStanza::XmppStreamMessage *>(msg)) {
140 842 : }
141 841 : static const char *Name() {
142 841 : return "EvXmppOpen";
143 : }
144 : XmppSession *session;
145 : boost::shared_ptr<const XmppStanza::XmppStreamMessage> msg;
146 : };
147 :
148 : struct EvStreamFeatureRequest : sc::event<EvStreamFeatureRequest> {
149 141 : EvStreamFeatureRequest(XmppSession *session,
150 141 : const XmppStanza::XmppMessage *msg) :
151 141 : session(session),
152 141 : msg(static_cast<const XmppStanza::XmppStreamMessage *>(msg)) {
153 141 : }
154 141 : static const char *Name() {
155 141 : return "EvStreamFeatureRequest";
156 : }
157 : XmppSession *session;
158 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
159 : };
160 :
161 : struct EvStartTls : sc::event<EvStartTls> {
162 136 : EvStartTls(XmppSession *session,
163 136 : const XmppStanza::XmppMessage *msg) :
164 136 : session(session),
165 136 : msg(static_cast<const XmppStanza::XmppStreamMessage *>(msg)) {
166 136 : }
167 136 : static const char *Name() {
168 136 : return "EvStartTls";
169 : }
170 : XmppSession *session;
171 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
172 : };
173 :
174 : struct EvTlsProceed : sc::event<EvTlsProceed> {
175 139 : EvTlsProceed(XmppSession *session,
176 139 : const XmppStanza::XmppMessage *msg) :
177 139 : session(session),
178 139 : msg(static_cast<const XmppStanza::XmppStreamMessage *>(msg)) {
179 139 : }
180 139 : static const char *Name() {
181 139 : return "EvTlsProceed";
182 : }
183 : XmppSession *session;
184 : boost::shared_ptr<const XmppStanza::XmppMessage> msg;
185 : };
186 :
187 : struct EvTlsHandShakeSuccess : sc::event<EvTlsHandShakeSuccess> {
188 268 : explicit EvTlsHandShakeSuccess(XmppSession *session) :
189 268 : session(session) { }
190 268 : static const char *Name() {
191 268 : 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 366947 : EvXmppKeepalive(XmppSession *session,
207 366947 : const XmppStanza::XmppMessage *msg) :
208 366947 : session(session), msg(msg) {
209 366947 : }
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 231792 : EvXmppMessageStanza(XmppSession *session,
219 231792 : const XmppStanza::XmppMessage *msg) :
220 231792 : session(session), msg(msg) {
221 231792 : }
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 2701 : EvXmppIqStanza(XmppSession *session,
231 : const XmppStanza::XmppMessage *msg)
232 2701 : : session(session), msg(msg) {
233 2701 : }
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 918 : Idle(my_context ctx) : my_base(ctx) {
254 918 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
255 918 : bool flap = (state_machine->get_state() == ESTABLISHED);
256 918 : state_machine->set_state(IDLE);
257 918 : state_machine->SendConnectionInfo("Start", "Active");
258 918 : if (flap) {
259 319 : state_machine->connection()->increment_flap_count();
260 : }
261 918 : }
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 925 : Active(my_context ctx) : my_base(ctx) {
276 925 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
277 925 : state_machine->keepalive_count_clear();
278 925 : bool flap = (state_machine->get_state() == ESTABLISHED);
279 925 : state_machine->set_state(ACTIVE);
280 925 : if (flap) {
281 214 : state_machine->connection()->increment_flap_count();
282 : }
283 925 : if (state_machine->IsActiveChannel() ) {
284 570 : if (state_machine->get_connect_attempts() >=
285 : XmppStateMachine::kMaxAttempts) {
286 15 : XmppConnection *connection = state_machine->connection();
287 15 : if (connection) {
288 15 : state_machine->SendConnectionInfo(
289 : "Connect failed after retries");
290 : // Notify clients if any action to be taken
291 15 : connection->ChannelMux()->HandleStateEvent(xmsm::ACTIVE);
292 : }
293 : }
294 570 : state_machine->StartConnectTimer(state_machine->GetConnectTime());
295 : }
296 925 : }
297 1850 : ~Active() {
298 925 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
299 925 : state_machine->CancelConnectTimer();
300 925 : state_machine->CancelOpenTimer();
301 1850 : }
302 :
303 : //event on client only
304 417 : sc::result react(const EvConnectTimerExpired &event) {
305 417 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
306 417 : if (state_machine->ConnectTimerCancelled()) {
307 0 : SM_LOG(state_machine, "Discard EvConnectTimerExpired in (Active) State");
308 0 : return discard_event();
309 : } else {
310 417 : state_machine->SendConnectionInfo(event.Name(), "Connect");
311 417 : SM_LOG(state_machine, "EvConnectTimerExpired in (Active) State");
312 417 : return transit<Connect>();
313 : }
314 : }
315 :
316 : // event on server only
317 279 : sc::result react(const EvTcpPassiveOpen &event) {
318 279 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
319 279 : event.session->AsyncReadStart();
320 279 : assert(state_machine->session() == event.session);
321 279 : event.session->set_observer(
322 : boost::bind(&XmppStateMachine::OnSessionEvent,
323 : state_machine, _1, _2));
324 279 : state_machine->StartOpenTimer(XmppStateMachine::kOpenTime);
325 279 : XmppConnectionInfo info;
326 279 : info.set_local_port(event.session->local_port());
327 278 : info.set_remote_port(event.session->remote_port());
328 279 : state_machine->SendConnectionInfo(&info, event.Name());
329 558 : return discard_event();
330 279 : }
331 :
332 : // event on server only
333 14 : sc::result react(const EvTcpClose &event) {
334 14 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
335 14 : if (event.session != state_machine->session()) {
336 0 : return discard_event();
337 : }
338 14 : state_machine->CancelOpenTimer();
339 14 : state_machine->ResetSession();
340 14 : state_machine->SendConnectionInfo(event.Name(), "Idle");
341 14 : 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 284 : sc::result react(const EvXmppOpen &event) {
365 284 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
366 284 : if (event.session != state_machine->session()) {
367 0 : return discard_event();
368 : }
369 284 : state_machine->AssignSession();
370 :
371 284 : XmppConnection *connection = state_machine->connection();
372 284 : if (connection->IsDeleted()) {
373 0 : state_machine->ResetSession();
374 0 : return discard_event();
375 : }
376 :
377 284 : XmppSession *session = state_machine->session();
378 284 : if (!state_machine->IsAuthEnabled())
379 144 : state_machine->ResurrectOldConnection(connection, session);
380 :
381 : // Get the possibly updated XmppConnection information.
382 284 : connection = state_machine->connection();
383 284 : state_machine->CancelOpenTimer();
384 284 : 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 284 : XmppConnectionInfo info;
392 284 : info.set_identifier(event.msg->from);
393 284 : if (state_machine->IsAuthEnabled()) {
394 140 : state_machine->SendConnectionInfo(&info, event.Name(),
395 : "Open Confirm");
396 140 : return transit<OpenConfirm>();
397 : } else {
398 144 : connection->StartKeepAliveTimer();
399 144 : state_machine->SendConnectionInfo(&info, event.Name(),
400 : "Established");
401 144 : return transit<XmppStreamEstablished>();
402 : }
403 284 : }
404 : }
405 :
406 12 : sc::result react(const EvStop &event) {
407 12 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
408 12 : XmppConnection *connection = state_machine->connection();
409 12 : state_machine->CancelOpenTimer();
410 12 : state_machine->CancelConnectTimer();
411 12 : connection->StopKeepAliveTimer();
412 :
413 12 : 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 11 : state_machine->ResetSession();
419 11 : state_machine->SendConnectionInfo(event.Name(), "Idle");
420 11 : return transit<Idle>();
421 : }
422 : }
423 :
424 0 : sc::result react(const EvAdminDown &event) {
425 0 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
426 0 : XmppConnectionInfo info;
427 0 : info.set_close_reason("Administratively down");
428 0 : state_machine->connection()->set_close_reason("Administratively down");
429 0 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
430 0 : return transit<Idle>();
431 0 : }
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 417 : Connect(my_context ctx) : my_base(ctx) {
449 417 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
450 417 : StartSession(state_machine);
451 417 : state_machine->connect_attempts_inc();
452 417 : state_machine->set_state(CONNECT);
453 417 : state_machine->StartConnectTimer(state_machine->GetConnectTime());
454 417 : XmppSession *session = state_machine->session();
455 417 : if (session != NULL) {
456 417 : XmppConnectionInfo info;
457 417 : info.set_local_port(session->local_port());
458 417 : info.set_remote_port(session->remote_port());
459 417 : state_machine->SendConnectionInfo(&info, "Connect Event");
460 417 : }
461 417 : }
462 834 : ~Connect() {
463 417 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
464 417 : state_machine->CancelConnectTimer();
465 834 : }
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 317 : sc::result react(const EvTcpConnected &event) {
483 317 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
484 317 : if (event.session != state_machine->session()) {
485 0 : return discard_event();
486 : }
487 317 : XmppConnection *connection = state_machine->connection();
488 317 : state_machine->CancelConnectTimer();
489 317 : XmppSession *session = state_machine->session();
490 317 : XmppConnectionInfo info;
491 317 : info.set_local_port(session->local_port());
492 317 : info.set_remote_port(session->remote_port());
493 317 : if (connection->SendOpen(session)) {
494 317 : state_machine->StartHoldTimer();
495 317 : state_machine->SendConnectionInfo(&info, event.Name(), "OpenSent");
496 317 : 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 317 : }
507 :
508 96 : sc::result react(const EvTcpConnectFail &event) {
509 : // delete session; restart connect timer.
510 96 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
511 96 : if (event.session != state_machine->session()) {
512 0 : return discard_event();
513 : }
514 96 : state_machine->set_session(NULL);
515 96 : state_machine->CancelConnectTimer();
516 96 : state_machine->SendConnectionInfo(event.Name(), "Active");
517 96 : 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 417 : void StartSession(XmppStateMachine *state_machine) {
555 417 : XmppConnection *connection = state_machine->connection();
556 417 : XmppSession *session = connection->CreateSession();
557 417 : state_machine->set_session(session);
558 417 : session->set_observer(boost::bind(&XmppStateMachine::OnSessionEvent,
559 : state_machine, _1, _2));
560 417 : boost::system::error_code err;
561 417 : session->socket()->bind(connection->local_endpoint(), err);
562 417 : if (err) {
563 0 : LOG(WARN, "Bind failure for local address " <<
564 : connection->local_endpoint() << " : " << err.message());
565 0 : assert(false);
566 : }
567 417 : connection->server()->Connect(session, connection->endpoint());
568 417 : }
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 317 : OpenSent(my_context ctx) : my_base(ctx) {
587 317 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
588 317 : state_machine->set_state(OPENSENT);
589 317 : }
590 634 : ~OpenSent() {
591 317 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
592 317 : state_machine->CancelHoldTimer();
593 634 : }
594 :
595 15 : sc::result react(const EvTcpClose &event) {
596 15 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
597 15 : if (event.session != state_machine->session()) {
598 0 : return discard_event();
599 : }
600 15 : state_machine->CancelHoldTimer();
601 15 : if (event.session) {
602 15 : state_machine->set_session(NULL);
603 15 : state_machine->SendConnectionInfo(event.Name(), "Active");
604 15 : return transit<Active>();
605 : }
606 0 : return discard_event();
607 : }
608 :
609 294 : sc::result react(const EvXmppOpen &event) {
610 294 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
611 294 : XmppConnection *connection = state_machine->connection();
612 294 : if (event.session == state_machine->session()) {
613 294 : state_machine->AssignSession();
614 294 : XmppConnectionInfo info;
615 294 : info.set_identifier(event.msg->from);
616 294 : if (state_machine->IsAuthEnabled()) {
617 144 : state_machine->SendConnectionInfo(&info, event.Name(),
618 : "Open Confirm");
619 144 : return transit<OpenConfirm>();
620 : } else {
621 150 : connection->SendKeepAlive();
622 150 : connection->StartKeepAliveTimer();
623 150 : state_machine->StartHoldTimer();
624 150 : state_machine->SendConnectionInfo(&info, event.Name(),
625 : "Established");
626 150 : return transit<XmppStreamEstablished>();
627 : }
628 294 : }
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 0 : sc::result react(const EvAdminDown &event) {
659 0 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
660 0 : CloseSession(state_machine);
661 0 : XmppConnectionInfo info;
662 0 : info.set_close_reason("Administratively down");
663 0 : state_machine->connection()->set_close_reason("Administratively down");
664 0 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
665 0 : return transit<Idle>();
666 0 : }
667 :
668 4 : void CloseSession(XmppStateMachine *state_machine) {
669 4 : state_machine->set_session(NULL);
670 4 : }
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 284 : OpenConfirm(my_context ctx) : my_base(ctx) {
688 284 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
689 284 : state_machine->StartHoldTimer();
690 284 : XmppConnectionInfo info;
691 284 : if (!state_machine->IsActiveChannel()) { //server
692 140 : if (state_machine->IsAuthEnabled()) {
693 140 : XmppConnection *connection = state_machine->connection();
694 140 : XmppSession *session = state_machine->session();
695 140 : 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 284 : state_machine->set_state(OPENCONFIRM);
711 284 : state_machine->set_openconfirm_state(OPENCONFIRM_INIT);
712 284 : }
713 :
714 2 : sc::result react(const EvTcpClose &event) {
715 2 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
716 2 : state_machine->CancelHoldTimer();
717 2 : 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 1 : state_machine->ResetSession();
723 1 : state_machine->SendConnectionInfo(event.Name(), "Idle");
724 1 : 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 141 : sc::result react(const EvStreamFeatureRequest &event) {
751 141 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
752 141 : XmppConnection *connection = state_machine->connection();
753 141 : 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 141 : 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 141 : state_machine->StartHoldTimer();
766 141 : state_machine->SendConnectionInfo(event.Name(),
767 : "Sent Start Tls, OpenConfirm Feature Negotiation");
768 141 : state_machine->set_openconfirm_state(
769 : OPENCONFIRM_FEATURE_NEGOTIATION);
770 141 : return discard_event();
771 : }
772 : }
773 :
774 : // received by client
775 139 : sc::result react(const EvTlsProceed &event) {
776 139 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
777 139 : XmppConnection *connection = state_machine->connection();
778 139 : XmppSession *session = state_machine->session();
779 139 : state_machine->StartHoldTimer();
780 139 : XmppConnectionInfo info;
781 139 : info.set_identifier(connection->GetTo());
782 : // Trigger Ssl Handshake on client side
783 139 : session->TriggerSslHandShake(state_machine->HandShakeCbHandler());
784 139 : state_machine->SendConnectionInfo( &info, event.Name(),
785 : "Trigger Client Ssl Handshake");
786 278 : return discard_event();
787 139 : }
788 :
789 : //received by server
790 136 : sc::result react(const EvStartTls &event) {
791 136 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
792 136 : if (event.session != state_machine->session()) {
793 0 : return discard_event();
794 : }
795 136 : XmppConnection *connection = state_machine->connection();
796 136 : XmppSession *session = state_machine->session();
797 136 : XmppConnectionInfo info;
798 136 : info.set_identifier(connection->GetTo());
799 136 : 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 136 : state_machine->StartHoldTimer();
807 : // Trigger Ssl Handshake on server side
808 136 : session->TriggerSslHandShake(state_machine->HandShakeCbHandler());
809 136 : state_machine->SendConnectionInfo(&info, event.Name(),
810 : "Trigger Server Ssl Handshake");
811 136 : return discard_event();
812 : }
813 136 : }
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 268 : sc::result react(const EvTlsHandShakeSuccess &event) {
830 268 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
831 268 : if (event.session != state_machine->session()) {
832 0 : return discard_event();
833 : }
834 268 : XmppConnection *connection = state_machine->connection();
835 268 : XmppSession *session = state_machine->session();
836 268 : session->AsyncReadStart();
837 268 : state_machine->set_openconfirm_state(OPENCONFIRM_FEATURE_SUCCESS);
838 268 : if (state_machine->IsActiveChannel()) { //client
839 134 : 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 268 : state_machine->StartHoldTimer();
850 268 : state_machine->SendConnectionInfo(event.Name(),
851 : "OpenConfirm Feature Negotiation Success");
852 268 : 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 263 : sc::result react(const EvXmppOpen &event) {
873 263 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
874 263 : if (event.session != state_machine->session()) {
875 0 : return discard_event();
876 : }
877 263 : XmppConnectionInfo info;
878 263 : info.set_identifier(event.msg->from);
879 263 : XmppConnection *connection = state_machine->connection();
880 263 : XmppSession *session = state_machine->session();
881 263 : if (connection->IsActiveChannel()) { //client
882 132 : connection->SendKeepAlive();
883 132 : connection->StartKeepAliveTimer();
884 132 : state_machine->StartHoldTimer();
885 132 : state_machine->SendConnectionInfo(&info, event.Name(),
886 : "Established");
887 132 : return transit<XmppStreamEstablished>();
888 : } else { //server
889 131 : 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 131 : if (state_machine->IsAuthEnabled())
901 131 : state_machine->ResurrectOldConnection(connection, session);
902 131 : connection = state_machine->connection();
903 131 : connection->StartKeepAliveTimer();
904 131 : state_machine->StartHoldTimer();
905 131 : state_machine->SendConnectionInfo(&info, event.Name(),
906 : "Established");
907 131 : return transit<XmppStreamEstablished>();
908 : }
909 : }
910 263 : }
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 557 : XmppStreamEstablished(my_context ctx) : my_base(ctx) {
942 557 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
943 557 : XmppConnection *connection = state_machine->connection();
944 557 : state_machine->StartHoldTimer();
945 557 : state_machine->set_state(ESTABLISHED);
946 557 : connection->ChannelMux()->HandleStateEvent(xmsm::ESTABLISHED);
947 557 : }
948 1114 : ~XmppStreamEstablished() {
949 557 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
950 557 : state_machine->CancelHoldTimer();
951 1114 : }
952 :
953 258 : sc::result react(const EvTcpClose &event) {
954 258 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
955 258 : if (event.session != state_machine->session()) {
956 0 : return discard_event();
957 : }
958 258 : state_machine->ResetSession();
959 258 : if (state_machine->IsActiveChannel()) {
960 210 : state_machine->SendConnectionInfo(event.Name(), "Active");
961 210 : return transit<Active>();
962 : } else {
963 48 : state_machine->SendConnectionInfo(event.Name(), "Idle");
964 48 : return transit<Idle>();
965 : }
966 : }
967 :
968 366563 : sc::result react(const EvXmppKeepalive &event) {
969 366563 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
970 366563 : if (event.session != state_machine->session()) {
971 0 : return discard_event();
972 : }
973 366563 : state_machine->keepalive_count_inc();
974 366563 : if (state_machine->get_keepalive_count() == 3) {
975 172 : state_machine->connect_attempts_clear();
976 : }
977 366563 : state_machine->StartHoldTimer();
978 366563 : return discard_event();
979 : }
980 :
981 231792 : sc::result react(const EvXmppMessageStanza &event) {
982 231792 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
983 231792 : if (event.session != state_machine->session()) {
984 0 : return discard_event();
985 : }
986 231792 : state_machine->StartHoldTimer();
987 463584 : state_machine->connection()->ProcessXmppChatMessage(
988 231792 : static_cast<const XmppStanza::XmppChatMessage *>(event.msg.get()));
989 231792 : return discard_event();
990 : }
991 :
992 2701 : sc::result react(const EvXmppIqStanza &event) {
993 2701 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
994 2701 : if (event.session != state_machine->session()) {
995 0 : return discard_event();
996 : }
997 2701 : state_machine->StartHoldTimer();
998 5402 : state_machine->connection()->ProcessXmppIqMessage(
999 2701 : static_cast<const XmppStanza::XmppMessage *>(event.msg.get()));
1000 2701 : 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 222 : sc::result react(const EvStop &event) {
1026 222 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
1027 222 : state_machine->SendConnectionInfo(event.Name());
1028 222 : state_machine->ResetSession();
1029 222 : if (state_machine->IsActiveChannel()) {
1030 0 : return transit<Active>();
1031 : } else {
1032 222 : return transit<Idle>();
1033 : }
1034 : }
1035 :
1036 48 : sc::result react(const EvAdminDown &event) {
1037 48 : XmppStateMachine *state_machine = &context<XmppStateMachine>();
1038 48 : state_machine->ResetSession();
1039 48 : XmppConnectionInfo info;
1040 48 : info.set_close_reason("Administratively down");
1041 48 : state_machine->connection()->set_close_reason("Administratively down");
1042 48 : state_machine->SendConnectionInfo(&info, event.Name(), "Idle");
1043 96 : return transit<Idle>();
1044 48 : }
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 564 : void XmppStateMachine::ResetSession() {
1071 564 : XmppConnection *connection = this->connection();
1072 564 : set_session(NULL);
1073 564 : CancelHoldTimer();
1074 :
1075 564 : if (!connection)
1076 0 : return;
1077 :
1078 : // Stop keepalives, transition to IDLE and notify registered entities.
1079 564 : connection->StopKeepAliveTimer();
1080 564 : connection->ChannelMux()->HandleStateEvent(xmsm::IDLE);
1081 563 : if (IsActiveChannel())
1082 263 : return;
1083 :
1084 : // Retain the connection if graceful restart is supported.
1085 301 : XmppServer *server = dynamic_cast<XmppServer *>(connection->server());
1086 301 : if (!server->IsPeerCloseGraceful())
1087 252 : connection->ManagedDelete();
1088 : }
1089 :
1090 516 : XmppStateMachine::XmppStateMachine(XmppConnection *connection, bool active,
1091 516 : bool auth_enabled, int config_hold_time)
1092 516 : : work_queue_(TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
1093 : connection->GetTaskInstance(),
1094 : boost::bind(&XmppStateMachine::DequeueEvent, this, _1)),
1095 516 : connection_(connection),
1096 516 : session_(NULL),
1097 516 : server_(connection->server()),
1098 516 : connect_timer_(
1099 516 : TimerManager::CreateTimer(*server_->event_manager()->io_service(),
1100 : "Connect timer",
1101 : TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
1102 : connection->GetTaskInstance())),
1103 516 : open_timer_(
1104 516 : TimerManager::CreateTimer(*server_->event_manager()->io_service(),
1105 : "Open timer",
1106 : TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
1107 : connection->GetTaskInstance())),
1108 516 : hold_timer_(
1109 516 : TimerManager::CreateTimer(*server_->event_manager()->io_service(),
1110 : "Hold timer",
1111 : TaskScheduler::GetInstance()->GetTaskId("xmpp::StateMachine"),
1112 : connection->GetTaskInstance())),
1113 516 : config_hold_time_(config_hold_time),
1114 516 : hold_time_(GetConfiguredHoldTime()),
1115 516 : attempts_(0),
1116 516 : keepalive_count_(0),
1117 516 : deleted_(false),
1118 516 : in_dequeue_(false),
1119 516 : is_active_(active),
1120 516 : auth_enabled_(auth_enabled),
1121 516 : state_(xmsm::IDLE),
1122 516 : last_state_(xmsm::IDLE),
1123 1032 : openconfirm_state_(xmsm::OPENCONFIRM_INIT) {
1124 : handshake_cb_ = boost::bind(
1125 516 : &XmppConnection::ProcessSslHandShakeResponse, connection, _1, _2);
1126 516 : }
1127 :
1128 607 : XmppStateMachine::~XmppStateMachine() {
1129 516 : assert(!deleted_);
1130 516 : deleted_ = true;
1131 516 : work_queue_.Shutdown();
1132 516 : 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 516 : 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 516 : TimerManager::DeleteTimer(connect_timer_);
1142 516 : TimerManager::DeleteTimer(open_timer_);
1143 516 : TimerManager::DeleteTimer(hold_timer_);
1144 607 : }
1145 :
1146 569 : void XmppStateMachine::Initialize() {
1147 569 : initiate();
1148 569 : Enqueue(xmsm::EvStart());
1149 569 : }
1150 :
1151 : // Note this api does not enqueue the deletion of TCP session
1152 516 : void XmppStateMachine::clear_session() {
1153 516 : if (session_ != NULL) {
1154 183 : session_->set_observer(NULL);
1155 183 : session_->ClearConnection();
1156 183 : session_->Close();
1157 183 : connection_->clear_session();
1158 183 : session_ = NULL;
1159 : }
1160 516 : }
1161 :
1162 534 : 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 534 : if (session != NULL) {
1166 534 : session->set_observer(NULL);
1167 534 : session->ClearConnection();
1168 534 : session->Close();
1169 534 : Enqueue(xmsm::EvTcpDeleteSession(session));
1170 : }
1171 534 : }
1172 :
1173 1922 : void XmppStateMachine::set_session(TcpSession *session) {
1174 1922 : if (session_ != NULL) {
1175 534 : connection_->clear_session();
1176 534 : DeleteSession(session_);
1177 : }
1178 1922 : session_ = static_cast<XmppSession *>(session);
1179 1922 : }
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 1914 : void XmppStateMachine::CancelConnectTimer() {
1197 1914 : connect_timer_->Cancel();
1198 1914 : }
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 1249 : void XmppStateMachine::CancelOpenTimer() {
1208 1249 : open_timer_->Cancel();
1209 1249 : }
1210 :
1211 516 : 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 516 : if (!env_checked) {
1217 9 : char *keepalive_time_str = getenv("XMPP_KEEPALIVE_SECONDS");
1218 9 : 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 9 : env_checked = true;
1224 : }
1225 507 : } else if (env_hold_time) {
1226 0 : return env_hold_time;
1227 507 : } else if (config_hold_time_) {
1228 507 : return config_hold_time_;
1229 : }
1230 :
1231 : // Use hard coded default.
1232 9 : return kHoldTime;
1233 : }
1234 :
1235 603311 : void XmppStateMachine::StartHoldTimer() {
1236 603311 : CancelHoldTimer();
1237 :
1238 603311 : if (hold_time_ <= 0)
1239 0 : return;
1240 :
1241 603311 : hold_timer_->Start(hold_time_ * 1000,
1242 : boost::bind(&XmppStateMachine::HoldTimerExpired, this),
1243 : boost::bind(&XmppStateMachine::TimerErrorHandler, this, _1, _2));
1244 : }
1245 :
1246 604769 : void XmppStateMachine::CancelHoldTimer() {
1247 604769 : hold_timer_->Cancel();
1248 604769 : }
1249 :
1250 585 : void XmppStateMachine::OnStart(const xmsm::EvStart &event) {
1251 585 : }
1252 :
1253 418 : bool XmppStateMachine::ConnectTimerExpired() {
1254 418 : XMPP_UTDEBUG(XmppStateMachineTimerExpire,
1255 : connection() ? connection()->ToUVEKey() : "",
1256 : XMPP_PEER_DIR_NA, this->ChannelType(), "Connect", StateName());
1257 418 : Enqueue(xmsm::EvConnectTimerExpired());
1258 418 : 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 984 : void XmppStateMachine::OnSessionEvent(
1289 : TcpSession *session, TcpSession::Event event) {
1290 984 : switch (event) {
1291 278 : case TcpSession::ACCEPT:
1292 278 : break;
1293 317 : case TcpSession::CONNECT_COMPLETE:
1294 317 : 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 317 : Enqueue(xmsm::EvTcpConnected(static_cast<XmppSession *>(session)));
1299 317 : break;
1300 96 : case TcpSession::CONNECT_FAILED:
1301 96 : 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 96 : Enqueue(xmsm::EvTcpConnectFail(static_cast<XmppSession *>(session)));
1306 96 : connection_->inc_connect_error();
1307 96 : break;
1308 293 : case TcpSession::CLOSE:
1309 293 : 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 293 : Enqueue(xmsm::EvTcpClose(static_cast<XmppSession *>(session)));
1314 293 : connection_->inc_session_close();
1315 293 : break;
1316 0 : default:
1317 0 : XMPP_WARNING(XmppUnknownEvent, session->ToUVEKey(), XMPP_PEER_DIR_IN,
1318 : this->ChannelType(), event);
1319 0 : break;
1320 : }
1321 984 : }
1322 :
1323 279 : bool XmppStateMachine::PassiveOpen(XmppSession *session) {
1324 279 : string state = "PassiveOpen in state: " + StateName();
1325 279 : XMPP_NOTICE(XmppEventLog, session->ToUVEKey(), XMPP_PEER_DIR_IN,
1326 : this->ChannelType(), state,
1327 : session->Connection()->endpoint().address().to_string(), "");
1328 558 : return Enqueue(xmsm::EvTcpPassiveOpen(session));
1329 279 : }
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 855 : void XmppStateMachine::ProcessStreamHeaderMessage(XmppSession *session,
1336 : const XmppStanza::XmppMessage *msg) {
1337 : XmppConnectionManager *connection_manager =
1338 855 : dynamic_cast<XmppConnectionManager *>(connection_->server());
1339 855 : std::scoped_lock lock(connection_manager->mutex());
1340 :
1341 : // Update "To" information which can be used to map an older session
1342 855 : session->Connection()->SetTo(msg->from);
1343 :
1344 855 : XmppServer *xmpp_server = dynamic_cast<XmppServer *>(server_);
1345 855 : XmppConnectionEndpoint *endpoint = NULL;
1346 :
1347 : // Look for an endpoint which may already exist
1348 855 : if (xmpp_server) {
1349 : endpoint =
1350 429 : xmpp_server->FindConnectionEndpoint(connection_->ToString());
1351 :
1352 858 : if (!xmpp_server->subcluster_name().empty() &&
1353 429 : 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 899 : if (!endpoint || !endpoint->connection() ||
1372 44 : connection_ == endpoint->connection()) {
1373 812 : ProcessEvent(xmsm::EvXmppOpen(session, msg));
1374 812 : return;
1375 : }
1376 :
1377 : // Check if the IP addresses match.
1378 : boost::asio::ip::address addr =
1379 43 : endpoint->connection()->endpoint().address();
1380 43 : if (connection_->endpoint().address() != addr) {
1381 0 : 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 0 : ProcessEvent(xmsm::EvTcpClose(session));
1388 0 : delete msg;
1389 0 : return;
1390 : }
1391 :
1392 43 : 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 43 : if (!xmpp_server->IsPeerCloseGraceful()) {
1398 0 : 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 0 : ProcessEvent(xmsm::EvTcpClose(session));
1403 0 : delete msg;
1404 0 : return;
1405 : }
1406 :
1407 43 : 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 43 : 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 43 : if (channel->ReceiverCount()) {
1428 13 : 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 13 : ProcessEvent(xmsm::EvTcpClose(session));
1433 13 : delete msg;
1434 13 : 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 30 : ProcessEvent(xmsm::EvXmppOpen(session, msg));
1443 855 : }
1444 :
1445 602728 : void XmppStateMachine::OnMessage(XmppSession *session,
1446 : const XmppStanza::XmppMessage *msg) {
1447 602728 : if (!Enqueue(xmsm::EvXmppMessage(session, msg)))
1448 0 : delete msg;
1449 602728 : }
1450 :
1451 602728 : void XmppStateMachine::ProcessMessage(XmppSession *session,
1452 : const XmppStanza::XmppMessage *msg) {
1453 : // Bail if session is already reset and disassociated from the connection.
1454 602728 : if (!session->Connection()) {
1455 17 : delete msg;
1456 17 : return;
1457 : }
1458 :
1459 602711 : const XmppStanza::XmppStreamMessage *stream_msg =
1460 : static_cast<const XmppStanza::XmppStreamMessage *>(msg);
1461 :
1462 602711 : switch (msg->type) {
1463 1271 : case XmppStanza::STREAM_HEADER:
1464 1271 : if (stream_msg->strmtype ==
1465 : XmppStanza::XmppStreamMessage::FEATURE_TLS) {
1466 :
1467 416 : switch (stream_msg->strmtlstype) {
1468 141 : case (XmppStanza::XmppStreamMessage::TLS_FEATURE_REQUEST):
1469 141 : ProcessEvent(xmsm::EvStreamFeatureRequest(session,
1470 : msg));
1471 141 : break;
1472 136 : case (XmppStanza::XmppStreamMessage::TLS_START):
1473 136 : ProcessEvent(xmsm::EvStartTls(session, msg));
1474 136 : break;
1475 139 : case (XmppStanza::XmppStreamMessage::TLS_PROCEED):
1476 139 : ProcessEvent(xmsm::EvTlsProceed(session, msg));
1477 139 : break;
1478 0 : default:
1479 0 : break;
1480 : }
1481 :
1482 855 : } 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 855 : ProcessStreamHeaderMessage(session, msg);
1487 1271 : break;
1488 366947 : case XmppStanza::WHITESPACE_MESSAGE_STANZA:
1489 366947 : ProcessEvent(xmsm::EvXmppKeepalive(session, msg));
1490 366946 : break;
1491 2701 : case XmppStanza::IQ_STANZA:
1492 2701 : ProcessEvent(xmsm::EvXmppIqStanza(session, msg));
1493 2701 : break;
1494 231792 : case XmppStanza::MESSAGE_STANZA:
1495 231792 : ProcessEvent(xmsm::EvXmppMessageStanza(session, msg));
1496 231792 : 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 402 : void XmppStateMachine::Clear() {
1510 402 : Enqueue(xmsm::EvStop());
1511 402 : }
1512 :
1513 84 : void XmppStateMachine::SetAdminState(bool down) {
1514 84 : assert(IsActiveChannel());
1515 84 : if (down) {
1516 64 : Enqueue(xmsm::EvAdminDown());
1517 : } else {
1518 20 : Enqueue(xmsm::EvStart());
1519 : }
1520 84 : }
1521 :
1522 270 : void XmppStateMachine::OnEvent(SslSession *session,
1523 : xmsm::SslHandShakeResponse resp) {
1524 270 : XmppSession *sess = static_cast<XmppSession *>(session);
1525 270 : switch(resp) {
1526 2 : case xmsm::EvTLSHANDSHAKE_FAILURE:
1527 2 : Enqueue(xmsm::EvTlsHandShakeFailure(sess));
1528 2 : break;
1529 268 : case xmsm::EvTLSHANDSHAKE_SUCCESS:
1530 268 : Enqueue(xmsm::EvTlsHandShakeSuccess(sess));
1531 268 : break;
1532 0 : default:
1533 0 : break;
1534 : }
1535 270 : }
1536 :
1537 3418 : void XmppStateMachine::set_state(xmsm::XmState state) {
1538 3418 : last_state_ = state_;
1539 3418 : state_ = state;
1540 3418 : state_since_ = UTCTimestampUsec();
1541 :
1542 3418 : 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 693 : void XmppStateMachine::set_openconfirm_state(xmsm::XmOpenConfirmState state) {
1557 693 : openconfirm_state_ = state;
1558 693 : }
1559 :
1560 : static const char *state_names[] = {
1561 : "Idle",
1562 : "Active",
1563 : "Connect",
1564 : "OpenSent",
1565 : "OpenConfirm",
1566 : "Established" };
1567 :
1568 612242 : string XmppStateMachine::StateName() const {
1569 612242 : return state_names[state_];
1570 : }
1571 :
1572 0 : string XmppStateMachine::LastStateName() const {
1573 0 : return state_names[last_state_];
1574 : }
1575 :
1576 0 : string XmppStateMachine::LastStateChangeAt() const {
1577 0 : return integerToString(UTCUsecToPTime(state_since_));
1578 : }
1579 :
1580 1112041 : xmsm::XmState XmppStateMachine::StateType() const {
1581 1112041 : return state_;
1582 : }
1583 :
1584 622817 : xmsm::XmOpenConfirmState XmppStateMachine::OpenConfirmStateType() const {
1585 622817 : return openconfirm_state_;
1586 : }
1587 :
1588 578 : void XmppStateMachine::AssignSession() {
1589 578 : connection_->set_session(static_cast<XmppSession *>(session_));
1590 578 : }
1591 :
1592 : const int XmppStateMachine::kConnectInterval;
1593 :
1594 987 : int XmppStateMachine::GetConnectTime() const {
1595 987 : int backoff = attempts_ > 6 ? 6 : attempts_;
1596 987 : return std::min(backoff ? 1 << (backoff - 1) : 0, kConnectInterval);
1597 : }
1598 :
1599 613840 : bool XmppStateMachine::IsActiveChannel() {
1600 613840 : return is_active_;
1601 : }
1602 :
1603 608809 : bool XmppStateMachine::logUVE() {
1604 608809 : return connection()->logUVE();
1605 : }
1606 :
1607 609812 : const char *XmppStateMachine::ChannelType() {
1608 609812 : return (IsActiveChannel() ? " Mode Client: " : " Mode Server: " );
1609 : }
1610 :
1611 605941 : 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 605941 : dynamic_cast<const xmsm::EvXmppMessage *>(event.get());
1616 605941 : if (ev_xmpp_message) {
1617 602728 : ProcessMessage(ev_xmpp_message->session, ev_xmpp_message->msg);
1618 : } else {
1619 3213 : ProcessEvent(*event);
1620 3214 : event.reset();
1621 : }
1622 605941 : return true;
1623 : }
1624 :
1625 605925 : void XmppStateMachine::ProcessEvent(const sc::event_base &event) {
1626 605925 : const xmsm::EvTcpDeleteSession *deferred_delete =
1627 605925 : dynamic_cast<const xmsm::EvTcpDeleteSession *>(&event);
1628 605925 : 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 605925 : if (deferred_delete) {
1635 534 : TcpSession *session = deferred_delete->session;
1636 534 : session->server()->DeleteSession(session);
1637 534 : return;
1638 : }
1639 605391 : if (deleted_) {
1640 0 : return;
1641 : }
1642 :
1643 605391 : update_last_event(TYPE_NAME(event));
1644 605391 : in_dequeue_ = true;
1645 605391 : process_event(event);
1646 605391 : in_dequeue_ = false;
1647 : }
1648 :
1649 523 : void XmppStateMachine::unconsumed_event(const sc::event_base &event) {
1650 523 : XMPP_INFO(XmppUnconsumedEvent, connection() ? connection()->ToUVEKey() : "",
1651 : XMPP_PEER_DIR_IN, ChannelType(), TYPE_NAME(event), StateName());
1652 523 : }
1653 :
1654 605390 : void XmppStateMachine::update_last_event(const std::string &event) {
1655 605390 : set_last_event(event);
1656 :
1657 605391 : if (!logUVE()) return;
1658 :
1659 : // Skip iq and message events.
1660 0 : 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 606004 : bool XmppStateMachine::Enqueue(const sc::event_base &event) {
1686 606004 : if (!deleted_) {
1687 606004 : work_queue_.Enqueue(event.intrusive_from_this());
1688 606004 : return true;
1689 : }
1690 :
1691 0 : return false;
1692 : }
1693 :
1694 : // Object trace routines.
1695 2387 : void XmppStateMachine::SendConnectionInfo(const string &event,
1696 : const string &nextstate) {
1697 2387 : XmppConnectionInfo info;
1698 2387 : SendConnectionInfo(&info, event, nextstate);
1699 4776 : return;
1700 2388 : }
1701 :
1702 4576 : void XmppStateMachine::SendConnectionInfo(XmppConnectionInfo *info,
1703 : const string &event, const string &nextstate) {
1704 :
1705 4576 : info->set_ip_address(this->connection()->endpoint().address().to_string());
1706 4576 : info->set_state(StateName());
1707 4577 : info->set_event(event);
1708 4577 : if (!nextstate.empty()) {
1709 3639 : info->set_next_state(nextstate);
1710 : }
1711 4577 : XMPP_CONNECTION_LOG_MSG(*info);
1712 4577 : 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 275 : 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 275 : 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 275 : if (created)
1742 255 : return;
1743 :
1744 : // GR Helper must be enabled when we are trying to resurrect connection.
1745 20 : XmppServer *server = dynamic_cast<XmppServer *>(new_connection->server());
1746 20 : assert(server->IsPeerCloseGraceful());
1747 :
1748 20 : 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 20 : XmppConnection *old_xmpp_connection = connection_endpoint->connection();
1754 20 : assert(old_xmpp_connection);
1755 :
1756 20 : XmppStateMachine *old_state_machine = old_xmpp_connection->state_machine();
1757 20 : assert(old_state_machine);
1758 :
1759 : // Swap Old and New connections and state machines linkages
1760 20 : new_connection->SwapXmppStateMachine(old_xmpp_connection);
1761 20 : this->SwapXmppConnection(old_state_machine);
1762 :
1763 : // Update XmppConnection in the old session.
1764 20 : XmppSession *old_xmpp_session = old_state_machine->session();
1765 20 : if (old_xmpp_session)
1766 0 : old_xmpp_session->SetConnection(new_connection);
1767 20 : 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 20 : 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 20 : new_connection->Shutdown();
1776 : }
1777 :
1778 20 : void XmppStateMachine::SwapXmppConnection(XmppStateMachine *other) {
1779 20 : swap(connection_, other->connection_);
1780 20 : connection_->SwapContents(other->connection_);
1781 20 : }
|
