./ct_report/coverage/mod_bosh_socket.COVER.html

1 -module(mod_bosh_socket).
2
3 -behaviour(gen_fsm_compat).
4 -behaviour(mongoose_c2s_socket).
5
6 %% API
7 -export([start/5,
8 start_link/5,
9 start_supervisor/0,
10 is_supervisor_started/0,
11 stop_supervisor/0,
12 handle_request/2,
13 pause/2]).
14
15 %% Private API
16 -export([get_handlers/1,
17 get_pending/1,
18 get_client_acks/1,
19 set_client_acks/2,
20 get_cached_responses/1]).
21
22
23 %% mongoose_c2s_socket callbacks
24 -export([socket_new/2,
25 socket_peername/1,
26 tcp_to_tls/2,
27 socket_handle_data/2,
28 socket_activate/1,
29 socket_send_xml/2,
30 socket_close/1,
31 get_peer_certificate/2,
32 has_peer_cert/2,
33 is_channel_binding_supported/1,
34 get_tls_last_message/1,
35 is_ssl/1]).
36
37 %% gen_fsm callbacks
38 -export([init/1,
39 accumulate/2, accumulate/3,
40 normal/2, normal/3,
41 closing/2, closing/3,
42 handle_event/3,
43 handle_sync_event/4,
44 handle_info/3,
45 terminate/3,
46 code_change/4]).
47
48 -ignore_xref([{mod_bosh_backend, delete_session, 1},
49 accumulate/2, accumulate/3, closing/2,
50 closing/3, get_cached_responses/1,
51 get_client_acks/1, get_handlers/1, get_peer_certificate/1,
52 get_pending/1, normal/2, normal/3,
53 pause/2, set_client_acks/2, start_link/5]).
54
55 -include("mongoose.hrl").
56 -include("jlib.hrl").
57 -include_lib("exml/include/exml_stream.hrl").
58 -include("mod_bosh.hrl").
59 -define(ACCUMULATE_PERIOD, 10).
60 -define(DEFAULT_HOLD, 1).
61 -define(CONCURRENT_REQUESTS, 2).
62 -define(DEFAULT_WAIT, 60).
63 -define(DEFAULT_CLIENT_ACKS, false).
64
65 -type cached_response() :: {rid(), TStamp :: integer(), exml:element()}.
66 -type rid() :: pos_integer().
67
68 -record(state, {from :: binary() | undefined,
69 to :: binary() | undefined,
70 c2s_pid :: pid(),
71 handlers = [] :: [{rid(), reference(), pid()}],
72 %% Elements buffered for sending to the client.
73 pending = [] :: [jlib:xmlstreamel()],
74 sid :: mod_bosh:sid(),
75 wait = ?DEFAULT_WAIT :: integer(),
76 hold = ?DEFAULT_HOLD :: integer(),
77 rid :: rid() | undefined,
78 %% Requests deferred for later processing because
79 %% of having Rid greater than expected.
80 deferred = [] :: [{rid(), {mod_bosh:event_type(), exml:element()}}],
81 client_acks = ?DEFAULT_CLIENT_ACKS :: boolean(),
82 sent = [] :: [cached_response()],
83 %% Allowed inactivity period in seconds.
84 inactivity :: pos_integer() | infinity,
85 inactivity_tref :: reference() | 'undefined',
86 %% Max pause period in seconds.
87 max_pause :: pos_integer(),
88 max_wait :: pos_integer() | infinity,
89 %% Are acknowledgements used?
90 server_acks :: boolean(),
91 last_processed :: rid() | 'undefined',
92 %% Report scheduled for sending at the earliest
93 %% possible occasion.
94 report = false :: {rid(), Time :: non_neg_integer()} | 'false'}).
95 -type state() :: #state{}.
96
97 -type statename() :: 'accumulate' | 'normal' | 'closing'.
98 -type fsm_return() :: {'next_state', statename(), state()}.
99
100 %%--------------------------------------------------------------------
101 %% API
102 %%--------------------------------------------------------------------
103
104 -spec start(mongooseim:host_type(), mod_bosh:sid(), mongoose_transport:peer(),
105 binary() | undefined, map()) ->
106 {'error', _} | {'ok', 'undefined' | pid()} | {'ok', 'undefined' | pid(), _}.
107 start(HostType, Sid, Peer, PeerCert, Opts) ->
108 125 supervisor:start_child(?BOSH_SOCKET_SUP, [HostType, Sid, Peer, PeerCert, Opts]).
109
110 -spec start_link(mongooseim:host_type(), mod_bosh:sid(), mongoose_transport:peer(),
111 binary() | undefined, map()) ->
112 'ignore' | {'error', _} | {'ok', pid()}.
113 start_link(HostType, Sid, Peer, PeerCert, Opts) ->
114 125 gen_fsm_compat:start_link(?MODULE, [{HostType, Sid, Peer, PeerCert, Opts}], []).
115
116 -spec start_supervisor() -> {ok, pid()} | {error, any()}.
117 start_supervisor() ->
118 98 ChildId = ?BOSH_SOCKET_SUP,
119 98 ChildSpec =
120 {ChildId,
121 {ejabberd_tmp_sup, start_link,
122 [ChildId, ?MODULE]},
123 transient,
124 infinity,
125 supervisor,
126 [ejabberd_tmp_sup]},
127 98 case supervisor:start_child(ejabberd_sup, ChildSpec) of
128 {ok, undefined} ->
129 
:-(
 {error, undefined};
130 {ok, Child} ->
131 98 {ok, Child};
132 {ok, Child, _Info} ->
133 
:-(
 {ok, Child};
134 {error, {already_started, Child}} ->
135 
:-(
 {ok, Child};
136 {error, Reason} ->
137 
:-(
 {error, Reason}
138 end.
139
140 -spec is_supervisor_started() -> boolean().
141 is_supervisor_started() ->
142 377 is_pid(whereis(?BOSH_SOCKET_SUP)).
143
144 -spec stop_supervisor() -> ok | {error, any()}.
145 stop_supervisor() ->
146 377 ejabberd_sup:stop_child(?BOSH_SOCKET_SUP).
147
148 -spec handle_request(Pid :: pid(),
149 {EventTag :: mod_bosh:event_type(),
150 Handler :: pid(),
151 Body :: exml:element()}) -> ok.
152 handle_request(Pid, Request) ->
153 2017 gen_fsm_compat:send_all_state_event(Pid, Request).
154
155
156 %% @doc TODO: no handler for this call is present!
157 %% No check for violating maxpause is made when calling this!
158 -spec pause(Pid :: pid(), Seconds :: pos_integer()) -> ok.
159 pause(Pid, Seconds) ->
160 
:-(
 gen_fsm_compat:send_all_state_event(Pid, {pause, Seconds}).
161
162 %%--------------------------------------------------------------------
163 %% Private API
164 %%--------------------------------------------------------------------
165
166 get_handlers(Pid) ->
167 19 gen_fsm_compat:sync_send_all_state_event(Pid, get_handlers).
168
169
170 get_pending(Pid) ->
171 
:-(
 gen_fsm_compat:sync_send_all_state_event(Pid, get_pending).
172
173
174 -spec get_client_acks(pid()) -> boolean().
175 get_client_acks(Pid) ->
176 
:-(
 gen_fsm_compat:sync_send_all_state_event(Pid, get_client_acks).
177
178
179 -spec set_client_acks(pid(), boolean()) -> any().
180 set_client_acks(Pid, Enabled) ->
181 4 gen_fsm_compat:sync_send_all_state_event(Pid, {set_client_acks, Enabled}).
182
183
184 -spec get_cached_responses(pid()) -> [cached_response()].
185 get_cached_responses(Pid) ->
186 1 gen_fsm_compat:sync_send_all_state_event(Pid, get_cached_responses).
187
188 %%--------------------------------------------------------------------
189 %% gen_fsm callbacks
190 %%--------------------------------------------------------------------
191
192 %%--------------------------------------------------------------------
193 %% @private
194 %% init(Args) -> {ok, StateName, State} |
195 %% {ok, StateName, State, Timeout} |
196 %% ignore |
197 %% {stop, StopReason}
198 %% @end
199 %%--------------------------------------------------------------------
200 -spec init([{mongooseim:host_type(), mod_bosh:sid(), mongoose_transport:peer(), undefined |
201 binary(), map()}]) ->
202 {ok, accumulate, state()}.
203 init([{HostType, Sid, Peer, PeerCert, ListenerOpts}]) ->
204 125 BoshSocket = #bosh_socket{sid = Sid, pid = self(), peer = Peer, peercert = PeerCert},
205 125 C2SOpts = ListenerOpts#{access => all,
206 shaper => none,
207 xml_socket => true,
208 max_stanza_size => 0,
209 hibernate_after => 0,
210 c2s_state_timeout => 5000,
211 backwards_compatible_session => true,
212 proto => tcp},
213 125 {ok, C2SPid} = mongoose_c2s:start({?MODULE, BoshSocket, C2SOpts}, []),
214 125 Opts = gen_mod:get_loaded_module_opts(HostType, mod_bosh),
215 125 State = new_state(Sid, C2SPid, Opts),
216 125 ?LOG_DEBUG(ls(#{what => bosh_socket_init, peer => Peer}, State)),
217 125 {ok, accumulate, State}.
218
219 new_state(Sid, C2SPid, #{inactivity := Inactivity, max_wait := MaxWait,
220 server_acks := ServerAcks, max_pause := MaxPause}) ->
221 125 #state{sid = Sid,
222 c2s_pid = C2SPid,
223 inactivity = Inactivity,
224 max_pause = MaxPause,
225 max_wait = MaxWait,
226 server_acks = ServerAcks}.
227
228 %%--------------------------------------------------------------------
229 %% @private
230 %% @doc
231 %% There should be one instance of this function for each possible
232 %% state name. Whenever a gen_fsm receives an event sent using
233 %% gen_fsm_compat:send_event/2, the instance of this function with the same
234 %% name as the current state name StateName is called to handle
235 %% the event. It is also called if a timeout occurs.
236 %%
237 %% state_name(Event, State) ->
238 %% {next_state, NextStateName, NextState} |
239 %% {next_state, NextStateName, NextState, Timeout} |
240 %% {stop, Reason, NewState}
241 %% @end
242 %%--------------------------------------------------------------------
243
244 -spec accumulate(_, state()) -> fsm_return().
245 accumulate(acc_off, #state{pending = Pending} = S) ->
246 246 NS = S#state{pending = []},
247 246 {next_state, normal, send_or_store(Pending, NS)};
248 accumulate(Event, State) ->
249 
:-(
 ?LOG_DEBUG(ls(#{what => bosh_socket_unhandled_event, state_name => accumulate,
250 
:-(
 event => Event}, State)),
251 
:-(
 {next_state, accumulate, State}.
252
253
254 -spec normal(_, state()) -> fsm_return().
255 normal(acc_off, #state{} = S) ->
256 
:-(
 {next_state, normal, S};
257 normal(Event, State) ->
258 
:-(
 ?LOG_DEBUG(ls(#{what => bosh_socket_unhandled_event, state_name => normal,
259 
:-(
 event => Event}, State)),
260 
:-(
 {next_state, normal, State}.
261
262 closing(Event, State) ->
263 
:-(
 ?LOG_DEBUG(ls(#{what => bosh_socket_unhandled_event, state_name => closing,
264 
:-(
 event => Event}, State)),
265 
:-(
 {next_state, closing, State}.
266
267 %%--------------------------------------------------------------------
268 %% @private
269 %% @doc
270 %% There should be one instance of this function for each possible
271 %% state name. Whenever a gen_fsm receives an event sent using
272 %% gen_fsm_compat:sync_send_event/[2, 3], the instance of this function with
273 %% the same name as the current state name StateName is called to
274 %% handle the event.
275 %%
276 %% state_name(Event, From, State) ->
277 %% {next_state, NextStateName, NextState} |
278 %% {next_state, NextStateName, NextState, Timeout} |
279 %% {reply, Reply, NextStateName, NextState} |
280 %% {reply, Reply, NextStateName, NextState, Timeout} |
281 %% {stop, Reason, NewState} |
282 %% {stop, Reason, Reply, NewState}
283 %% @end
284 %%--------------------------------------------------------------------
285 accumulate(Event, _From, State) ->
286 
:-(
 ?LOG_DEBUG(ls(#{what => bosh_socket_unhandled_sync_event, state_name => accumulate,
287 
:-(
 event => Event}, State)),
288 
:-(
 {reply, ok, accumulate, State}.
289
290 normal(Event, _From, State) ->
291 
:-(
 ?LOG_DEBUG(ls(#{what => bosh_socket_unhandled_sync_event, state_name => normal,
292 
:-(
 event => Event}, State)),
293 
:-(
 {reply, ok, normal, State}.
294
295 closing(Event, _From, State) ->
296 
:-(
 ?LOG_DEBUG(ls(#{what => bosh_socket_unhandled_sync_event, state_name => closing,
297 
:-(
 event => Event}, State)),
298 
:-(
 {reply, ok, closing, State}.
299
300 %%--------------------------------------------------------------------
301 %% @private
302 %% @doc
303 %% Whenever a gen_fsm receives an event sent using
304 %% gen_fsm_compat:send_all_state_event/2, this function is called to handle
305 %% the event.
306 %%
307 %% handle_event(Event, StateName, State) ->
308 %% {next_state, NextStateName, NextState} |
309 %% {next_state, NextStateName, NextState, Timeout} |
310 %% {stop, Reason, NewState}
311 %% @end
312 %%--------------------------------------------------------------------
313
314 handle_event({EventTag, Handler, #xmlel{} = Body}, SName, S) ->
315 2017 NS = cancel_inactivity_timer(S),
316 2017 Rid = binary_to_integer(exml_query:attr(Body, <<"rid">>)),
317 2017 try
318 2017 NNS = handle_stream_event({EventTag, Body, Rid}, Handler, SName, NS),
319 %% TODO: it's the event which determines the next state,
320 %% this ought to be returned from handle_stream_event
321 2014 determine_next_state(EventTag, SName, NNS)
322 catch
323 throw:{invalid_rid, TState} ->
324 2 {stop, {shutdown, invalid_rid}, TState};
325 throw:{invalid_pause, TState} ->
326 1 {stop, {shutdown, policy_violation}, TState}
327 end;
328
329 handle_event(Event, StateName, State) ->
330 
:-(
 ?LOG_DEBUG(ls(#{what => bosh_socket_unhandled_all_state,
331 
:-(
 state_name => StateName, event => Event}, State)),
332 
:-(
 {next_state, StateName, State}.
333
334
335 determine_next_state(_EventTag, closing, NNS) ->
336 12 {stop, normal, NNS};
337 determine_next_state(EventTag, SName, NNS) ->
338 2002 case EventTag of
339 _ when EventTag == streamstart; EventTag == restart ->
340 246 timer:apply_after(?ACCUMULATE_PERIOD,
341 gen_fsm_compat, send_event, [self(), acc_off]),
342 246 {next_state, accumulate, NNS};
343 _ ->
344 1756 {next_state, SName, NNS}
345 end.
346
347 %%--------------------------------------------------------------------
348 %% @private
349 %% @doc
350 %% Whenever a gen_fsm receives an event sent using
351 %% gen_fsm_compat:sync_send_all_state_event/[2, 3], this function is called
352 %% to handle the event.
353 %%
354 %% handle_sync_event(Event, From, StateName, State) ->
355 %% {next_state, NextStateName, NextState} |
356 %% {next_state, NextStateName, NextState, Timeout} |
357 %% {reply, Reply, NextStateName, NextState} |
358 %% {reply, Reply, NextStateName, NextState, Timeout} |
359 %% {stop, Reason, NewState} |
360 %% {stop, Reason, Reply, NewState}
361 %% @end
362 %%--------------------------------------------------------------------
363 handle_sync_event(get_handlers, _From, StateName,
364 #state{handlers = Handlers} = S) ->
365 19 {reply, Handlers, StateName, S};
366 handle_sync_event(get_pending, _From, StateName,
367 #state{pending = Pending} = S) ->
368 
:-(
 {reply, Pending, StateName, S};
369 handle_sync_event(get_client_acks, _From, StateName,
370 #state{client_acks = ClientAcks} = S) ->
371 
:-(
 {reply, ClientAcks, StateName, S};
372 handle_sync_event({set_client_acks, ClientAcks}, _From, StateName,
373 #state{} = S) ->
374 4 NS = S#state{client_acks = ClientAcks},
375 4 {reply, ok, StateName, NS};
376 handle_sync_event(get_cached_responses, _From, StateName,
377 #state{sent = CachedResponses} = S) ->
378 1 {reply, CachedResponses, StateName, S};
379 handle_sync_event(Event, _From, StateName, State) ->
380 
:-(
 ?LOG_DEBUG(ls(#{what => bosh_socket_unhandled_sync_all_state,
381 
:-(
 state_name => StateName, event => Event}, State)),
382 
:-(
 Reply = ok,
383 
:-(
 {reply, Reply, StateName, State}.
384
385 %%--------------------------------------------------------------------
386 %% @private
387 %% @doc
388 %% This function is called by a gen_fsm when it receives any
389 %% message other than a synchronous or asynchronous event
390 %% (or a system message).
391 %%
392 %% handle_info(Info, StateName, State)->
393 %% {next_state, NextStateName, NextState} |
394 %% {next_state, NextStateName, NextState, Timeout} |
395 %% {stop, Reason, NewState}
396 %% @end
397 %%--------------------------------------------------------------------
398
399 handle_info({send, #xmlstreamend{} = StreamEnd}, normal = SName,
400 #state{pending = Pending} = S) ->
401 120 NS = send_or_store(Pending ++ [StreamEnd], S#state{pending = []}),
402 120 {next_state, SName, NS};
403 handle_info({send, Data}, accumulate = SName, #state{} = S) ->
404 492 {next_state, SName, store([Data], S)};
405 handle_info({send, Data}, normal = SName, #state{} = S) ->
406 781 NS = send_or_store(Data, S),
407 781 {next_state, SName, NS};
408 handle_info(close, _SName, #state{pending = []} = State) ->
409 108 {stop, normal, State};
410 handle_info(close, _SName, State) ->
411 12 {next_state, closing, State};
412 handle_info(inactivity_timeout, _SName, State) ->
413 2 ?LOG_INFO(ls(#{what => bosh_socket_terminating, reason => inactivity_timeout}, State)),
414 2 {stop, {shutdown, inactivity_timeout}, State};
415 handle_info({wait_timeout, {Rid, Pid}}, SName,
416 #state{handlers = Handlers} = S) ->
417 2 ?LOG_INFO(ls(#{what => bosh_socket_wait_timeout,
418 2 handler_rid => Rid, handler_pid => Pid}, S)),
419 %% In case some message was being handled when the timer fired
420 %% it may turn out that Pid is no longer available in Handlers.
421 2 case lists:keytake(Rid, 1, Handlers) of
422 false ->
423 
:-(
 {next_state, SName, S};
424 {value, {Rid, _, Pid}, NewHandlers} ->
425 2 NS = send_to_handler({Rid, Pid}, [],
426 S#state{handlers = NewHandlers}),
427 2 {next_state, SName, NS}
428 end;
429 handle_info(Info, SName, State) ->
430 
:-(
 ?UNEXPECTED_INFO(Info),
431 
:-(
 {next_state, SName, State}.
432
433 terminate(Reason, StateName, #state{sid = Sid, handlers = Handlers} = S) ->
434 125 [Pid ! {close, Sid} || {_, _, Pid} <- lists:sort(Handlers)],
435 125 mod_bosh_backend:delete_session(Sid),
436 125 catch mongoose_c2s:stop(S#state.c2s_pid, normal),
437 125 ?LOG_DEBUG(ls(#{what => bosh_socket_closing_session, reason => Reason,
438 state_name => StateName, handlers => Handlers,
439 125 pending => S#state.pending}, S)).
440
441 code_change(_OldVsn, StateName, State, _Extra) ->
442 
:-(
 {ok, StateName, State}.
443
444 %%--------------------------------------------------------------------
445 %% callback implementations
446 %%--------------------------------------------------------------------
447
448 handle_stream_event({EventTag, Body, Rid} = Event, Handler,
449 SName, #state{rid = OldRid} = S) ->
450 2020 ExpectedRid = maybe_add(1, OldRid),
451 2020 NS = maybe_add_handler(Handler, Rid, S),
452 2020 NNS = case {EventTag,
453 maybe_is_retransmission(Rid, OldRid, S#state.sent),
454 is_expected_rid(Rid, ExpectedRid),
455 is_acceptable_rid(Rid, ExpectedRid)}
456 of
457 {_, {true, CachedResponse}, _, _} when Handler /= none ->
458 1 case CachedResponse of
459 none ->
460 
:-(
 NS;
461 _ ->
462 1 resend_cached(CachedResponse, NS)
463 end;
464 {streamstart, _, _, _} ->
465 125 process_acked_stream_event(Event, SName, NS);
466 {_, _, true, _} ->
467 1889 process_acked_stream_event(Event, SName, NS);
468 {_, _, false, true} ->
469 3 ?LOG_INFO(ls(#{what => bosh_socket_deferring,
470 event_rid => Rid, body => Body,
471 3 expected_rid => ExpectedRid, event_tag => EventTag}, S)),
472 3 NS#state{deferred = [Event | NS#state.deferred]};
473 {_, _, false, false} ->
474 2 ?LOG_ERROR(#{what => bosh_socket_invalid_rid,
475 event_rid => Rid, body => Body,
476 expected_rid => ExpectedRid, event_tag => EventTag,
477 
:-(
 difference => maybe_diff(Rid, ExpectedRid)}),
478 2 [Pid ! item_not_found
479 2 || {_, _, Pid} <- lists:sort(NS#state.handlers)],
480 2 throw({invalid_rid, NS#state{handlers = []}})
481 end,
482 2017 return_surplus_handlers(SName, NNS).
483
484 -spec maybe_is_retransmission(rid(), rid(), [cached_response()])
485 -> false | {true, none} | {true, cached_response()}.
486 maybe_is_retransmission(Rid, OldRid, Sent) ->
487 2020 case {lists:keyfind(Rid, 1, Sent), Rid =:= OldRid} of
488 {false, false} ->
489 2019 false;
490 {false, true} ->
491 
:-(
 ?LOG_INFO(#{what => bosh_socket_request_repeated,
492 text => <<"Request repeated but no response found in cache">>,
493 
:-(
 event_rid => Rid, old_rid => OldRid, sent => Sent}),
494 
:-(
 {true, none};
495 {CachedResponse, _} ->
496 1 {true, CachedResponse}
497 end.
498
499 -spec maybe_add(rid(), rid() | undefined)
500 -> rid() | undefined.
501 125 maybe_add(_, undefined) -> undefined;
502 maybe_add(Rid1, Rid2) when is_integer(Rid1),
503 1895 is_integer(Rid2) -> Rid1 + Rid2.
504
505 -spec maybe_diff(rid(), rid() | undefined)
506 -> non_neg_integer() | undefined.
507 
:-(
 maybe_diff(_, undefined) -> undefined;
508 2 maybe_diff(Rid, Expected) -> abs(Rid-Expected).
509
510 -spec resend_cached(cached_response(), state()) -> state().
511 resend_cached({_Rid, _, CachedBody}, S) ->
512 1 send_to_handler(CachedBody, S).
513
514
515 -spec process_acked_stream_event({EventTag :: mod_bosh:event_type(),
516 Body :: exml:element(),
517 Rid :: 'undefined' | rid()},
518 SName :: any(),
519 S :: state() ) -> state().
520 process_acked_stream_event({EventTag, Body, Rid}, SName,
521 #state{} = S) ->
522 2014 MaybeBAck = exml_query:attr(Body, <<"ack">>),
523 2014 {Action, Ack} = determine_report_action(MaybeBAck, S#state.client_acks,
524 Rid, S#state.last_processed),
525 2014 NS = maybe_trim_cache(Ack, S),
526 2014 case Action of
527 noreport ->
528 2013 process_stream_event(EventTag, Body, SName, rid(NS, Rid));
529 report ->
530 1 NS2 = schedule_report(Ack, NS),
531 1 NS3 = process_stream_event(EventTag, Body, SName, rid(NS2, Rid)),
532 1 maybe_send_report(NS3)
533 end.
534
535 rid(#state{} = S, Rid) when is_integer(Rid), Rid > 0 ->
536 2014 S#state{rid = Rid}.
537
538
539 -spec determine_report_action(BinAck :: 'undefined' | binary(),
540 boolean(),
541 Rid :: rid(),
542 LastProcessed :: 'undefined' | pos_integer()
543 ) -> {'noreport', _} | {'report', _}.
544 determine_report_action(undefined, false, _, _) ->
545 2009 {noreport, undefined};
546 determine_report_action(undefined, true, Rid, LastProcessed) ->
547 2 ?WARNING_MSG_IF(Rid+1 /= LastProcessed, "expected 'ack' attribute on ~p~n", [Rid]),
548 2 {noreport, undefined};
549 determine_report_action(BinAck, _, _, LastProcessed) ->
550 3 Ack = binary_to_integer(BinAck),
551 3 case {LastProcessed, is_valid_ack(Ack, LastProcessed)} of
552 {undefined, _} ->
553 
:-(
 {noreport, Ack};
554 {_, true} ->
555 2 {noreport, Ack};
556 {_, false} ->
557 1 {report, Ack}
558 end.
559
560
561 -spec is_valid_ack(Ack :: rid(), 'undefined' | pos_integer()) -> boolean().
562 is_valid_ack(Ack, LastProcessed)
563 when Ack < LastProcessed ->
564 1 false;
565 is_valid_ack(_, _) ->
566 2 true.
567
568
569 -spec maybe_trim_cache(undefined | any(), state()) -> state().
570 maybe_trim_cache(undefined, S) ->
571 2011 S;
572 maybe_trim_cache(Ack, S) ->
573 3 UpToAck = fun({R, _, _}) when R =< Ack ->
574 5 true;
575 (_) ->
576 1 false
577 end,
578 3 NewSent = lists:dropwhile(UpToAck, S#state.sent),
579 3 S#state{sent = NewSent}.
580
581
582 -spec schedule_report(rid(), state()) -> state().
583 schedule_report(Ack, #state{sent = Sent} = S) ->
584 1 ReportRid = Ack + 1,
585 1 try
586 1 {ReportRid, TimeSent, _} = lists:keyfind(ReportRid, 1, Sent),
587 1 ElapsedTimeMillis = erlang:monotonic_time(millisecond) - TimeSent,
588 1 Report = {ReportRid, ElapsedTimeMillis},
589 1 case S#state.report of
590 false ->
591 1 S#state{report = Report};
592 OldReport when OldReport < Report ->
593 
:-(
 S#state{report = OldReport};
594 _ ->
595 
:-(
 S#state{report = Report}
596 end
597 catch
598 error:{badmatch, {resp, false}} ->
599 
:-(
 ?LOG_ERROR(ls(#{what => bosh_socket_no_cached_response,
600 
:-(
 responses => Sent, rid_offender => ReportRid}, S)),
601 
:-(
 S
602 end.
603
604
605 -spec maybe_send_report(state()) -> state().
606 maybe_send_report(#state{report = false} = S) ->
607 
:-(
 S;
608 maybe_send_report(#state{} = S) ->
609 1 send_or_store([], S).
610
611
612 -spec process_stream_event(mod_bosh:event_type(), exml:element(), _SName,
613 state()) -> state().
614 process_stream_event(pause, Body, SName, State) ->
615 3 Seconds = binary_to_integer(exml_query:attr(Body, <<"pause">>)),
616 3 NewState = process_pause_event(Seconds, State),
617 2 process_deferred_events(SName, NewState);
618 process_stream_event(EventTag, Body, SName, #state{c2s_pid = C2SPid} = State) ->
619 2011 {Els, NewState} = bosh_unwrap(EventTag, Body, State),
620 2011 [forward_to_c2s(C2SPid, El) || El <- Els],
621 2011 process_deferred_events(SName, NewState).
622
623
624 -spec process_pause_event('infinity' | 'undefined' | pos_integer(),
625 state()) -> state().
626 process_pause_event(Seconds, #state{max_pause = MaxPause} = S)
627 when MaxPause == undefined;
628 Seconds > MaxPause ->
629 1 [Pid ! policy_violation || {_, _, Pid} <- S#state.handlers],
630 1 throw({invalid_pause, S#state{handlers = []}});
631 process_pause_event(Seconds, State) ->
632 2 NS = State#state{inactivity = Seconds},
633 2 F = fun(_, S) ->
634 4 send_to_handler([], S)
635 end,
636 2 lists:foldl(F, NS, lists:seq(1, length(State#state.handlers))).
637
638
639 -spec process_deferred_events(_SName, state()) -> state().
640 process_deferred_events(SName, #state{deferred = Deferred} = S) ->
641 2013 lists:foldl(fun(Event, State) ->
642 3 ?LOG_DEBUG(ls(#{what => bosh_socket_processing_deferred_event,
643 3 event => Event}, S)),
644 3 handle_stream_event(Event, none, SName, State)
645 end,
646 S#state{deferred = []},
647 lists:sort(Deferred)).
648
649
650 -spec is_expected_rid(rid(), rid() | undefined) -> boolean().
651 is_expected_rid(Rid, ExpectedRid) when Rid == ExpectedRid ->
652 1889 true;
653 is_expected_rid(_, _) ->
654 131 false.
655
656 -spec is_acceptable_rid(rid(), rid() | undefined) -> boolean().
657 is_acceptable_rid(Rid, ExpectedRid)
658 when Rid > ExpectedRid,
659 Rid < ExpectedRid + ?CONCURRENT_REQUESTS ->
660 3 true;
661 is_acceptable_rid(_, _) ->
662 2017 false.
663
664 %% @doc Send data to the client if a request handler is available, that matches next RID.
665 %% Otherwise, store for sending later.
666 -spec send_or_store(_Data, state()) -> state().
667 send_or_store(Data, State) when not is_list(Data) ->
668 781 send_or_store([Data], State);
669 send_or_store(Data, State) ->
670 2091 case send_to_handler(Data, State) of
671 no_valid_handler ->
672 87 store(Data, State);
673 NewState ->
674 2004 NewState
675 end.
676
677
678 %% @doc send_to_handler() assumes that Handlers is not empty!
679 %% Be sure that's the case if calling it.
680 -spec send_to_handler([any()] | exml:element(), state()) -> state() | no_valid_handler.
681 send_to_handler(Data, State) ->
682 2096 case pick_handler(State) of
683 {Handler, NS} ->
684 2009 send_to_handler(Handler, Data, NS);
685 false ->
686 87 no_valid_handler
687 end.
688
689
690 %% Return handler and new state if a handler is available
691 %% or `false` otherwise.
692 -spec pick_handler(state()) -> {{rid(), pid()}, state()} | false.
693 pick_handler(#state{ handlers = [] }) ->
694 87 false;
695 pick_handler(#state{ handlers = Handlers, rid = Rid } = S) ->
696 2009 case lists:sort(Handlers) of
697 [{HandlerRid, TRef, Pid} | HRest] when HandlerRid =< Rid->
698 %% The cancellation might fail if the timer already fired.
699 %% Don't worry, it's handled on receiving the timeout message.
700 2009 erlang:cancel_timer(TRef),
701 2009 {{HandlerRid, Pid}, S#state{handlers = HRest}};
702 _ ->
703 
:-(
 false
704 end.
705
706
707 -spec send_to_handler({_, atom() | pid() | port() | {atom(), atom()}},
708 Wrapped :: [any()] | exml:element(),
709 State :: state() ) -> state().
710 send_to_handler({_, Pid}, #xmlel{name = <<"body">>} = Wrapped, State) ->
711 1 send_wrapped_to_handler(Pid, Wrapped, State);
712 send_to_handler({Rid, Pid}, Data, State) ->
713 2010 {Wrapped, NS} = bosh_wrap(Data, Rid, State),
714 2010 NS2 = cache_response({Rid, erlang:monotonic_time(millisecond), Wrapped}, NS),
715 2010 send_wrapped_to_handler(Pid, Wrapped, NS2).
716
717
718 %% @doc This is the most specific variant of send_to_handler()
719 %% and the *only one* actually performing a send
720 %% to the cowboy_loop_handler serving a HTTP request.
721 -spec send_wrapped_to_handler(atom() | pid() | port() | {atom(), atom()},
722 Wrapped :: exml:element(),
723 State :: state()) -> state().
724 send_wrapped_to_handler(Pid, Wrapped, #state{handlers = []} = State) ->
725 1128 Pid ! {bosh_reply, Wrapped},
726 1128 setup_inactivity_timer(State);
727 send_wrapped_to_handler(Pid, Wrapped, State) ->
728 883 Pid ! {bosh_reply, Wrapped},
729 883 State.
730
731
732 -spec maybe_ack(rid(), state()) -> [{binary(), _}].
733 maybe_ack(HandlerRid, #state{rid = Rid} = S) ->
734 2010 case Rid > HandlerRid of
735 true ->
736 882 server_ack(S#state.server_acks, Rid);
737 false ->
738 1128 []
739 end.
740
741
742 -spec maybe_report(state()) -> {[{binary(), _}], state()}.
743 maybe_report(#state{report = false} = S) ->
744 2009 {[], S};
745 maybe_report(#state{report = Report} = S) ->
746 1 {ReportRid, ElapsedTime} = Report,
747 1 NewAttrs = [{<<"report">>, integer_to_binary(ReportRid)},
748 {<<"time">>, integer_to_binary(ElapsedTime)}],
749 1 {NewAttrs, S#state{report = false}}.
750
751
752 -spec cache_response(cached_response(), state()) -> state().
753 cache_response({Rid, _, _} = Response, #state{sent = Sent} = S) ->
754 2010 NewSent = lists:keymerge(1, [Response], Sent),
755 2010 CacheUpTo = case S#state.client_acks of
756 true ->
757 %% Acknowledgements are on - there's no limit on the number
758 %% of cached responses.
759 4 infinity;
760 false ->
761 %% Leave up to ?CONCURRENT_REQUESTS responses in cache.
762 2006 ?CONCURRENT_REQUESTS
763 end,
764 2010 S#state{sent = cache_up_to(CacheUpTo, NewSent),
765 last_processed = last_processed(Rid, S#state.last_processed)}.
766
767
768 -spec cache_up_to('infinity' | 2, Responses :: [cached_response()])
769 -> [cached_response()].
770 cache_up_to(infinity, Responses) ->
771 4 Responses;
772 cache_up_to(N, Responses) ->
773 2006 lists:nthtail(max(0, length(Responses) - N), Responses).
774
775
776 -spec last_processed(rid(), 'undefined' | pos_integer()) -> rid().
777 last_processed(Rid, undefined) ->
778 125 Rid;
779 last_processed(Rid1, Rid2) ->
780 1885 max(Rid1, Rid2).
781
782
783 -spec setup_inactivity_timer(state()) -> state().
784 setup_inactivity_timer(#state{inactivity = infinity} = S) ->
785 
:-(
 S;
786 setup_inactivity_timer(S) ->
787 1128 cancel_inactivity_timer(S),
788 1128 TRef = erlang:send_after(timer:seconds(S#state.inactivity), self(),
789 inactivity_timeout),
790 1128 S#state{inactivity_tref = TRef}.
791
792
793 -spec cancel_inactivity_timer(state()) -> state().
794 cancel_inactivity_timer(#state{inactivity_tref = undefined} = S) ->
795 2139 S;
796 cancel_inactivity_timer(S) ->
797 1006 erlang:cancel_timer(S#state.inactivity_tref),
798 1006 S#state{inactivity_tref = undefined}.
799
800
801 %% @doc Store data for sending later.
802 -spec store([jlib:xmlstreamel()], state()) -> state().
803 store(Data, #state{pending = Pending} = S) ->
804 579 S#state{pending = Pending ++ Data}.
805
806
807 -spec forward_to_c2s(pid() | undefined, jlib:xmlstreamel()) -> ok.
808 forward_to_c2s(C2SPid, StreamElement) ->
809 1258 C2SPid ! {tcp, undefined, StreamElement},
810 1258 ok.
811
812
813 -spec maybe_add_handler(_, rid(), state()) -> state().
814 maybe_add_handler(Handler, Rid, S) when is_pid(Handler) ->
815 2017 add_handler({Rid, Handler}, S);
816 maybe_add_handler(_, _, S) ->
817 3 S.
818
819
820 -spec add_handler({rid(), pid()}, state()) -> state().
821 add_handler({Rid, Pid}, #state{handlers = Handlers} = S) ->
822 2017 TRef = erlang:send_after(timer:seconds(S#state.wait), self(),
823 {wait_timeout, {Rid, Pid}}),
824 2017 S#state{handlers = [{Rid, TRef, Pid} | Handlers]}.
825
826
827 %% @doc Keep in mind the hardcoding for hold == 1.
828 -spec return_surplus_handlers('accumulate' | 'normal' | 'closing', state()) -> state().
829 return_surplus_handlers(SName, #state{handlers = []} = State)
830 when SName == accumulate; SName == normal; SName == closing ->
831 2 State;
832 return_surplus_handlers(SName, #state{handlers = []} = State)
833 when SName == normal; SName == closing ->
834 
:-(
 State;
835 return_surplus_handlers(accumulate, #state{handlers = [_]} = State) ->
836 125 State;
837 return_surplus_handlers(SName, #state{handlers = [_], pending = []} = State)
838 when SName == normal; SName == closing ->
839 947 State;
840 return_surplus_handlers(accumulate, #state{handlers = _} = S) ->
841 
:-(
 case send_to_handler([], S) of
842 
:-(
 no_valid_handler -> S;
843 
:-(
 NS -> return_surplus_handlers(accumulate, NS)
844 end;
845 return_surplus_handlers(SName, #state{pending = Pending} = S)
846 when SName == normal; SName == closing ->
847 943 send_or_store(Pending, S#state{pending = []}).
848
849
850 -spec bosh_unwrap(EventTag :: mod_bosh:event_type(), exml:element(), state())
851 -> {[jlib:xmlstreamel()], state()}.
852 bosh_unwrap(StreamEvent, Body, #state{} = S)
853 when StreamEvent =:= streamstart ->
854 125 Wait = min(get_attr(<<"wait">>, Body, S#state.wait), S#state.max_wait),
855 125 Hold = get_attr(<<"hold">>, Body, S#state.hold),
856 125 ClientAcks = get_client_acks(StreamEvent, Body, S#state.client_acks, S),
857 125 From = exml_query:attr(Body, <<"from">>),
858 125 To = exml_query:attr(Body, <<"to">>),
859 125 E = stream_start(From, To),
860 125 S2 = S#state{wait = Wait, hold = Hold, client_acks = ClientAcks,
861 from = From, to = To},
862 125 {[E], S2};
863
864 bosh_unwrap(StreamEvent, _Body, #state{} = S)
865 when StreamEvent =:= restart ->
866 121 {[stream_start(S#state.from, S#state.to)], S};
867
868 bosh_unwrap(streamend, Body, State) ->
869 118 {Els, NewState} = bosh_unwrap(normal, Body, State),
870 118 {Els ++ [#xmlstreamend{name = <<>>}], NewState};
871
872 bosh_unwrap(normal, Body, #state{sid = Sid} = State) ->
873 1765 Sid = exml_query:attr(Body, <<"sid">>),
874 1765 ?NS_HTTPBIND = exml_query:attr(Body, <<"xmlns">>),
875
876 1765 {[El || El <- Body#xmlel.children,
877 %% Ignore whitespace keepalives.
878 894 El /= #xmlcdata{content = <<" ">>}],
879 State}.
880
881
882 -spec get_client_acks(streamstart, exml:element(), boolean(), #state{}) -> boolean().
883 get_client_acks(streamstart, Element, Default, State) ->
884 125 case exml_query:attr(Element, <<"ack">>) of
885 undefined ->
886 125 Default;
887 <<"1">> ->
888 
:-(
 true;
889 _ ->
890 
:-(
 ?LOG_INFO(ls(#{what => bosh_socket_ignore_ack,
891 
:-(
 text => <<"Ignoring invalid client ack on stream start">>}, State)),
892 
:-(
 false
893 end.
894
895
896 -spec get_attr(Attr :: binary(), exml:element(), integer()) -> any().
897 get_attr(Attr, Element, Default) ->
898 250 case exml_query:attr(Element, Attr) of
899 undefined ->
900 
:-(
 Default;
901 Value ->
902 250 binary_to_integer(Value)
903 end.
904
905
906 -spec stream_start(binary(), binary()) -> jlib:xmlstreamstart().
907 stream_start(From, To) ->
908 246 #xmlstreamstart{name = <<"stream:stream">>,
909 attrs = [{<<"from">>, From},
910 {<<"to">>, To},
911 {<<"version">>, <<"1.0">>},
912 {<<"xml:lang">>, <<"en">>},
913 {<<"xmlns">>, ?NS_CLIENT},
914 {<<"xmlns:stream">>, ?NS_STREAM}]}.
915
916
917 -spec bosh_wrap([any()], rid(), state()) -> {exml:element(), state()}.
918 bosh_wrap(Elements, Rid, #state{} = S) ->
919 2010 EventsStanzas = lists:partition(fun is_stream_event/1, Elements),
920 2010 {{Body, Children}, NS} = case EventsStanzas of
921 {[], Stanzas} ->
922 1643 {{bosh_body(S), Stanzas}, S};
923 {[#xmlstreamstart{} = StreamStart], Stanzas} ->
924 246 {{bosh_stream_start_body(StreamStart, S), Stanzas}, S};
925 {[#xmlstreamend{}], []} ->
926 %% No stanzas except stream end - OK.
927 119 {{bosh_stream_end_body(), []}, S};
928 {[#xmlstreamend{} = StreamEnd], Stanzas} ->
929 %% Can't wrap remaining stanzas in a stream end body.
930 %% Send Stanzas and forfeit sending stream end.
931 2 ?LOG_DEBUG(#{what => bosh_socket_cannot_send_stream_end,
932 text => <<"Can't send stream end yet. Still have pending stanzas">>,
933 2 stanzas => Stanzas}),
934 2 Pending = S#state.pending,
935 2 {{bosh_body(S), Stanzas},
936 S#state{pending = Pending ++ [StreamEnd]}}
937 end,
938 2010 MaybeAck = maybe_ack(Rid, NS),
939 2010 {MaybeReport, NNS} = maybe_report(NS),
940 2010 HasStreamPrefix = (exml_query:attr(Body, <<"xmlns:stream">>) /= undefined),
941 2010 MaybeStreamPrefix = maybe_stream_prefix(HasStreamPrefix, Children),
942 2010 ExtraAttrs = MaybeAck ++ MaybeReport ++ MaybeStreamPrefix,
943 2010 {Body#xmlel{attrs = Body#xmlel.attrs ++ ExtraAttrs,
944 children = maybe_add_default_ns_to_children(Children)}, NNS}.
945
946
947 -spec is_stream_event(jlib:xmlstreamel()) -> boolean().
948 is_stream_event(#xmlstreamstart{}) ->
949 246 true;
950 is_stream_event(#xmlstreamend{}) ->
951 121 true;
952 is_stream_event(_) ->
953 1027 false.
954
955
956 %% @doc Bosh body for a session creation response.
957 -spec bosh_stream_start_body(jlib:xmlstreamstart(), state()) -> exml:element().
958 bosh_stream_start_body(#xmlstreamstart{attrs = Attrs}, #state{} = S) ->
959 246 #xmlel{name = <<"body">>,
960 attrs = [{<<"wait">>, integer_to_binary(S#state.wait)},
961 {<<"requests">>,
962 integer_to_binary(?CONCURRENT_REQUESTS)},
963 {<<"hold">>, integer_to_binary(S#state.hold)},
964 {<<"from">>, proplists:get_value(<<"from">>, Attrs)},
965 %% TODO: how to support these with cowboy?
966 {<<"accept">>, <<"deflate, gzip">>},
967 {<<"sid">>, S#state.sid},
968 {<<"xmpp:restartlogic">>, <<"true">>},
969 {<<"xmpp:version">>, <<"1.0">>},
970 {<<"xmlns">>, ?NS_HTTPBIND},
971 {<<"xmlns:xmpp">>, <<"urn:xmpp:xbosh">>},
972 {<<"xmlns:stream">>, ?NS_STREAM}] ++
973 inactivity(S#state.inactivity) ++
974 maxpause(S#state.max_pause) ++
975 %% TODO: shouldn't an ack be sent on restart?
976 server_ack(S#state.server_acks, S#state.rid),
977 children = []}.
978
979
980 -spec inactivity('infinity' | 'undefined' | pos_integer()) -> [{binary(), _}].
981 inactivity(I) ->
982 246 [{<<"inactivity">>, integer_to_binary(I)} || is_integer(I)].
983
984
985 -spec maxpause('undefined' | pos_integer()) -> [{binary(), _}].
986 maxpause(MP) ->
987 246 [{<<"maxpause">>, integer_to_binary(MP)} || is_integer(MP)].
988
989
990 -spec server_ack('false' | 'true' | 'undefined', 'undefined' | rid())
991 -> [{binary(), _}].
992 server_ack(ServerAcks, Rid) ->
993 1128 [{<<"ack">>, integer_to_binary(Rid)} || ServerAcks =:= true].
994
995
996 %% @doc Bosh body for an ordinary stream element(s).
997 -spec bosh_body(state()) -> exml:element().
998 bosh_body(#state{} = S) ->
999 1645 #xmlel{name = <<"body">>,
1000 attrs = [{<<"sid">>, S#state.sid},
1001 {<<"xmlns">>, ?NS_HTTPBIND}],
1002 children = []}.
1003
1004
1005 -spec bosh_stream_end_body() -> exml:element().
1006 bosh_stream_end_body() ->
1007 119 #xmlel{name = <<"body">>,
1008 attrs = [{<<"type">>, <<"terminate">>},
1009 {<<"xmlns">>, ?NS_HTTPBIND}],
1010 children = []}.
1011
1012 maybe_stream_prefix(true, _) ->
1013 246 [];
1014 maybe_stream_prefix(_, Stanzas) ->
1015 1764 case lists:any(fun is_stream_prefix/1, Stanzas) of
1016 false ->
1017 1762 [];
1018 true ->
1019 2 [{<<"xmlns:stream">>, ?NS_STREAM}]
1020 end.
1021
1022 2 is_stream_prefix(#xmlel{name = <<"stream:error">>}) -> true;
1023 
:-(
 is_stream_prefix(#xmlel{name = <<"stream:features">>}) -> true;
1024 779 is_stream_prefix(_) -> false.
1025
1026
1027 %%--------------------------------------------------------------------
1028 %% Helpers
1029 %%--------------------------------------------------------------------
1030
1031 maybe_add_default_ns_to_children(Children) ->
1032 2010 lists:map(fun maybe_add_default_ns/1, Children).
1033
1034 maybe_add_default_ns(#xmlel{name = Name, attrs = Attrs} = El)
1035 when Name =:= <<"message">>; Name =:= <<"presence">>; Name =:= <<"iq">> ->
1036 658 case xml:get_attr(<<"xmlns">>, Attrs) of
1037 false ->
1038 636 El#xmlel{attrs = [{<<"xmlns">>, ?NS_CLIENT} | Attrs]};
1039 _ ->
1040 22 El
1041 end;
1042 maybe_add_default_ns(El) ->
1043 369 El.
1044
1045 ls(LogMap, State) ->
1046 
:-(
 S = #{sid => State#state.sid,
1047 c2s_pid => State#state.c2s_pid,
1048 from_jid => State#state.from,
1049 to_jid => State#state.to,
1050 rid => State#state.rid},
1051 
:-(
 maps:merge(LogMap, ignore_undefined(S)).
1052
1053 ignore_undefined(Map) ->
1054 
:-(
 maps:filter(fun(_, V) -> V =/= undefined end, Map).
1055
1056 %% mongoose_c2s_socket callbacks
1057
1058 -spec socket_new(mod_bosh:socket(), mongoose_listener:options()) -> mod_bosh:socket().
1059 socket_new(Socket, _LOpts) ->
1060 125 Socket.
1061
1062 -spec socket_peername(mod_bosh:socket()) -> {inet:ip_address(), inet:port_number()}.
1063 socket_peername(#bosh_socket{peer = Peer}) ->
1064 246 Peer.
1065
1066 -spec tcp_to_tls(mod_bosh:socket(), mongoose_listener:options()) ->
1067 {ok, mod_bosh:socket()} | {error, term()}.
1068 tcp_to_tls(_Socket, _LOpts) ->
1069 
:-(
 {error, tls_not_allowed_on_bosh}.
1070
1071 -spec socket_handle_data(mod_bosh:socket(), {tcp | ssl, term(), iodata()}) ->
1072 iodata() | {raw, [exml:element()]} | {error, term()}.
1073 socket_handle_data(_Socket, {_Kind, _Term, Packet}) ->
1074 1258 {raw, [Packet]}.
1075
1076 -spec socket_activate(mod_bosh:socket()) -> ok.
1077 socket_activate(_Socket) ->
1078 1383 ok.
1079
1080 -spec socket_send_xml(mod_bosh:socket(),
1081 iodata() | exml_stream:element() | [exml_stream:element()]) ->
1082 ok | {error, term()}.
1083 socket_send_xml(#bosh_socket{pid = Pid}, XMLs) when is_list(XMLs) ->
1084 1393 [Pid ! {send, XML} || XML <- XMLs],
1085 1393 ok;
1086 socket_send_xml(#bosh_socket{pid = Pid}, XML) ->
1087 
:-(
 Pid ! {send, XML},
1088 
:-(
 ok.
1089
1090 -spec socket_close(mod_bosh:socket()) -> ok.
1091 socket_close(#bosh_socket{pid = Pid}) ->
1092 125 Pid ! close,
1093 125 ok.
1094
1095 -spec get_peer_certificate(mod_bosh:socket(), mongoose_listener:options()) ->
1096 mongoose_transport:peercert_return().
1097 get_peer_certificate(#bosh_socket{peercert = undefined}, _) ->
1098 
:-(
 no_peer_cert;
1099 get_peer_certificate(#bosh_socket{peercert = PeerCert}, _) ->
1100 9 Decoded = public_key:pkix_decode_cert(PeerCert, plain),
1101 9 {ok, Decoded}.
1102
1103 -spec has_peer_cert(mod_bosh:socket(), mongoose_listener:options()) -> boolean().
1104 has_peer_cert(Socket, LOpts) ->
1105 6 get_peer_certificate(Socket, LOpts) /= no_peer_cert.
1106
1107 -spec is_channel_binding_supported(mod_bosh:socket()) -> boolean().
1108 is_channel_binding_supported(_Socket) ->
1109 240 false.
1110
1111 -spec get_tls_last_message(mod_bosh:socket()) -> {ok, binary()} | {error, term()}.
1112 get_tls_last_message(_Socket) ->
1113 
:-(
 {error, tls_not_allowed_on_bosh}.
1114
1115 -spec is_ssl(mod_bosh:socket()) -> boolean().
1116 is_ssl(_Socket) ->
1117 367 false.
1118
1119 %%--------------------------------------------------------------------
1120 %% Tests
1121 %%--------------------------------------------------------------------
1122
1123 -ifdef(TEST).
1124
1125 -include_lib("eunit/include/eunit.hrl").
1126
1127 cache_up_to_test_() ->
1128 [?_test(?assertEqual( [4, 5], cache_up_to(2, [1, 2, 3, 4, 5]) ))].
1129
1130 -endif.
Line Hits Source