xmpp.chapril.org-ejabberd/src/mod_mqtt_ws.erl

%%%-------------------------------------------------------------------
%%% @author Evgeny Khramtsov <ekhramtsov@process-one.net>
%%% @copyright (C) 2002-2024 ProcessOne, SARL. All Rights Reserved.
%%%
%%% Licensed under the Apache License, Version 2.0 (the "License");
%%% you may not use this file except in compliance with the License.
%%% You may obtain a copy of the License at
%%%
%%%     http://www.apache.org/licenses/LICENSE-2.0
%%%
%%% Unless required by applicable law or agreed to in writing, software
%%% distributed under the License is distributed on an "AS IS" BASIS,
%%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%% See the License for the specific language governing permissions and
%%% limitations under the License.
%%%
%%%-------------------------------------------------------------------
-module(mod_mqtt_ws).
-ifndef(GEN_SERVER).
-define(GEN_SERVER, gen_server).
-endif.
-behaviour(?GEN_SERVER).

%% API
-export([socket_handoff/3]).
-export([start/1, start_link/1]).
-export([peername/1, setopts/2, send/2, close/1]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
	 terminate/2, code_change/3]).

-include_lib("xmpp/include/xmpp.hrl").
-include("ejabberd_http.hrl").
-include("logger.hrl").

-define(SEND_TIMEOUT, timer:seconds(15)).

-record(state, {socket :: socket(),
		ws_pid :: pid(),
		mqtt_session :: undefined | pid()}).

-type peername() :: {inet:ip_address(), inet:port_number()}.
-type socket() :: {http_ws, pid(), peername()}.
-export_type([socket/0]).

%%%===================================================================
%%% API
%%%===================================================================
socket_handoff(LocalPath, Request, Opts) ->
    ejabberd_websocket:socket_handoff(
      LocalPath, Request, Opts, ?MODULE, fun get_human_html_xmlel/0).

start({#ws{http_opts = Opts}, _} = WS) ->
    ?GEN_SERVER:start(?MODULE, [WS], ejabberd_config:fsm_limit_opts(Opts)).

start_link({#ws{http_opts = Opts}, _} = WS) ->
    ?GEN_SERVER:start_link(?MODULE, [WS], ejabberd_config:fsm_limit_opts(Opts)).

-spec peername(socket()) -> {ok, peername()}.
peername({http_ws, _, IP}) ->
    {ok, IP}.

-spec setopts(socket(), list()) -> ok.
setopts(_WSock, _Opts) ->
    ok.

-spec send(socket(), iodata()) -> ok | {error, timeout | einval}.
send({http_ws, Pid, _}, Data) ->
    try ?GEN_SERVER:call(Pid, {send, Data}, ?SEND_TIMEOUT)
    catch exit:{timeout, {?GEN_SERVER, _, _}} ->
	    {error, timeout};
	  exit:{_, {?GEN_SERVER, _, _}} ->
	    {error, einval}
    end.

-spec close(socket()) -> ok.
close({http_ws, Pid, _}) ->
    ?GEN_SERVER:cast(Pid, close).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
init([{#ws{ip = IP, http_opts = ListenOpts}, WsPid}]) ->
    Socket = {http_ws, self(), IP},
    case mod_mqtt_session:start(?MODULE, Socket, ListenOpts) of
	{ok, Pid} ->
	    erlang:monitor(process, Pid),
	    erlang:monitor(process, WsPid),
	    mod_mqtt_session:accept(Pid),
	    State = #state{socket = Socket,
			   ws_pid = WsPid,
			   mqtt_session = Pid},
	    {ok, State};
	{error, Reason} ->
	    {stop, Reason};
	ignore ->
	    ignore
    end.

handle_call({send, Data}, _From, #state{ws_pid = WsPid} = State) ->
    WsPid ! {data, Data},
    {reply, ok, State};
handle_call(Request, From, State) ->
    ?WARNING_MSG("Unexpected call from ~p: ~p", [From, Request]),
    {noreply, State}.

handle_cast(close, State) ->
    {stop, normal, State#state{mqtt_session = undefined}};
handle_cast(Request, State) ->
    ?WARNING_MSG("Unexpected cast: ~p", [Request]),
    {noreply, State}.

handle_info(closed, State) ->
    {stop, normal, State};
handle_info({received, Data}, State) ->
    State#state.mqtt_session ! {tcp, State#state.socket, Data},
    {noreply, State};
handle_info({'DOWN', _, process, Pid, _}, State)
  when Pid == State#state.mqtt_session orelse Pid == State#state.ws_pid ->
    {stop, normal, State};
handle_info(Info, State) ->
    ?WARNING_MSG("Unexpected info: ~p", [Info]),
    {noreply, State}.

terminate(_Reason, State) ->
    if State#state.mqtt_session /= undefined ->
	    State#state.mqtt_session ! {tcp_closed, State#state.socket};
       true ->
	    ok
    end.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%%===================================================================
%%% Internal functions
%%%===================================================================
-spec get_human_html_xmlel() -> xmlel().
get_human_html_xmlel() ->
    Heading = <<"ejabberd mod_mqtt">>,
    #xmlel{name = <<"html">>,
           attrs =
               [{<<"xmlns">>, <<"http://www.w3.org/1999/xhtml">>}],
           children =
               [#xmlel{name = <<"head">>, attrs = [],
                       children =
                           [#xmlel{name = <<"title">>, attrs = [],
                                   children = [{xmlcdata, Heading}]}]},
                #xmlel{name = <<"body">>, attrs = [],
                       children =
                           [#xmlel{name = <<"h1">>, attrs = [],
                                   children = [{xmlcdata, Heading}]},
                            #xmlel{name = <<"p">>, attrs = [],
                                   children =
                                       [{xmlcdata, <<"An implementation of ">>},
                                        #xmlel{name = <<"a">>,
                                               attrs =
                                                   [{<<"href">>,
                                                     <<"http://tools.ietf.org/html/rfc6455">>}],
                                               children =
                                                   [{xmlcdata,
                                                     <<"WebSocket protocol">>}]}]},
                            #xmlel{name = <<"p">>, attrs = [],
                                   children =
                                       [{xmlcdata,
                                         <<"This web page is only informative. To "
                                           "use WebSocket connection you need an MQTT "
                                           "client that supports it.">>}]}]}]}.