使用cowboy实现websocket主要实现以下回调函数 下面的函数返回值要具体弄清楚原因参考 websocket具体协议 主要就是两个部分 握手和数据传输
-export([init/3]). 常见所有处理程序回调。建立WebSocket连接，这个函数必须返回 upgrade 的元组。
-export([websocket_init/3]). 初始socket状态，也可以用于注册的过程，启动一个定时器，等返回参考doc
.然后和WebSocket握手。 -export([websocket_handle/3]).处理客户端发送的对应的消息。返回格式 参考cow对应doc
-export([websocket_info/3]). 处理erlang之间派发的消息
-export([websocket_terminate/3]). websocket 断开 用于扫尾工作处理
我在这里还是采用otp方式构建整个项目 主要代码 以及效果
-module(websocket_app). -behaviour(application). %% Application callbacks
-export([start/, stop/]). %% ===================================================================
%% Application callbacks
%% =================================================================== start(_StartType, _StartArgs) ->
%%启动存储pid的树据 可以采用 ets 表格处理 但是为了方便集群处理 我采用的mnesia
ok = websocket_store:init(),
%% 配置路由
Dispatch = cowboy_router:compile([
{'_', [
{"/", cowboy_static, {priv_file, websocket, "index.html"}},
{"/websocket", websocket_hander, []},
{"/static/[...]", cowboy_static, {priv_dir, websocket, "static"}}
]}
]),
{ok, _} = cowboy:start_http(http, , [{port, }],
[{env, [{dispatch, Dispatch}]}]),
%%启动监督树
websocket_sup:start_link(). stop(_State) ->
ok.
-module(websocket_sup). -behaviour(supervisor). %% API
-export([start_link/]). %% Supervisor callbacks
-export([init/]). %% Helper macro for declaring children of supervisor
-define(CHILD(I, Type), {I, {I, start_link, []}, permanent, , Type, [I]}). %% ===================================================================
%% API functions
%% =================================================================== start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []). %% ===================================================================
%% Supervisor callbacks
%% =================================================================== init([]) ->
Children = [],
{ok, { {one_for_one, , }, Children} }.
-module(websocket_hander). -behaviour(cowboy_websocket_handler). %% ------------------------------------------------------------------
%% API Function Exports
%% ------------------------------------------------------------------ %% ------------------------------------------------------------------
%% gen_server Function Exports
%% ------------------------------------------------------------------ -export([init/]).
-export([websocket_init/]).
-export([websocket_handle/]).
-export([websocket_info/]).
-export([websocket_terminate/]). %% ------------------------------------------------------------------
%% API Function Definitions
%% ------------------------------------------------------------------ %% ------------------------------------------------------------------
%% gen_server Function Definitions
%% ------------------------------------------------------------------ %%websocket 握手
init({tcp, http}, _Req, _Opts) ->
%%插入数据
websocket_store:dirty_insert(self(),web),
{upgrade, protocol, cowboy_websocket}.
%%连接初始
websocket_init(_TransportName, Req, _Opts) ->
%%连接一秒后 发送消息
erlang:start_timer(, self(), <<"Hello!">>),
{ok, Req, undefined_state}.
%%处理客户端发送投递的消息
websocket_handle({text, Msg}, Req, State) ->
%% 把消息投递到其余的进程 这里可以 投递到 tcp socket 也可以投递的 websocket 这样就可以实现 服务支持多协议处理
lists:foreach(fun([Pid,SocketType]) ->
case SocketType of
tcp -> ok;
web ->
case Pid == self() of
true -> ok;
false -> Pid!{chat,Msg}
end
end
end,websocket_store:dirty_lookall()),
{reply, {text, << "That's what she said! ", Msg/binary >>}, Req, State};
websocket_handle(_Data, Req, State) ->
{ok, Req, State}.
%% 处理erlang 发送的消息 这里是接收 行发送的数据
websocket_info({chat,Msg},_Req,State)->
{reply,{text,Msg},_Req,State};
%% 处理 erlang:start_timer(, self(), <<"Hello!">>) 发送的消息
websocket_info({timeout, _Ref, Msg}, Req, State) ->
{reply, {text, Msg}, Req, State}; websocket_info(_Info, Req, State) ->
{ok, Req, State}.
%%断开socket
websocket_terminate(_Reason, _Req, _State) ->
websocket_store:dirty_delete(self()),
ok. %% ------------------------------------------------------------------
%% Internal Function Definitions
%% ------------------------------------------------------------------
%%%-------------------------------------------------------------------
%%% @author thinkpad <>
%%% @copyright (C) , thinkpad
%%% @doc
%%%
%%% @end
%%% Created : Jun by thinkpad <>
%%%-------------------------------------------------------------------
-module(websocket_store). -include_lib("stdlib/include/qlc.hrl" ).
%% API
-export([dirty_insert/,dirty_lookall/,dirty_lookall_record/,dirty_delete/]). -export([init/,insert/,delete/,lookup/,lookall/]). -record(socket_to_pid, {pid,socket_type}). -define(WAIT_FOR_TABLES, ). %%%===================================================================
%%% API
%%%===================================================================
init()->
dynamic_db_init(). %--------------------------------------------------------------------
%% @doc Insert a key and pid.
%% @spec insert(Key, Pid) -> void()
%% @end
%%--------------------------------------------------------------------
insert(Pid,SocketType) when is_pid(Pid) ->
Fun = fun() -> mnesia:write(#socket_to_pid{pid = Pid, socket_type = SocketType}) end,
{atomic, _} = mnesia:transaction(Fun). %--------------------------------------------------------------------
%% @doc dirty insert pid and socket
%% @spec dirty_insert(Pid socket)
%% @end
%%-------------------------------------------------------------------- dirty_insert(Pid,SocketType) when is_pid(Pid) ->
mnesia:dirty_write(#socket_to_pid{pid = Pid, socket_type = SocketType}).
%--------------------------------------------------------------------
%% @doc dirty_read data
%% @spec dirty_lookall()
%% @end
%%-------------------------------------------------------------------- dirty_lookall()->
mnesia:dirty_select(socket_to_pid,[{#socket_to_pid{pid='$1',socket_type = '$2'},[],['$$']}]). dirty_delete(Pid)->
mnesia:dirty_delete(socket_to_pid,Pid). %--------------------------------------------------------------------
%% @doc look all record info
%% @spec
%% @end
%%-------------------------------------------------------------------- dirty_lookall_record()->
mnesia:dirty_select(socket_to_pid,[{#socket_to_pid{pid='$1',socket_type = '$2'},[],['$_']}]).
%%--------------------------------------------------------------------
%% @doc Find a pid given a key.
%% @spec lookup(Key) -> {ok, Pid} | {error, not_found}
%% @end
%%--------------------------------------------------------------------
lookup(Pid) ->
do(qlc:q([{X#socket_to_pid.pid,X#socket_to_pid.socket_type} || X <- mnesia:table(socket_to_pid),X#socket_to_pid.pid==Pid])). %%--------------------------------------------------------------------
%% @doc Find all list
%% @spec lookall() -> {List} | {error, not_found}
%% @end
%%--------------------------------------------------------------------
lookall() ->
do(qlc:q([[X#socket_to_pid.pid,X#socket_to_pid.socket_type] || X <- mnesia:table(socket_to_pid)])). %%--------------------------------------------------------------------
%% @doc Delete an element by pid from the registrar.
%% @spec delete(Pid) -> void()
%% @end
%%--------------------------------------------------------------------
delete(Pid) ->
try
[#socket_to_pid{} = Record] = mnesia:dirty_read(socket_to_pid, Pid, #socket_to_pid.pid),
mnesia:dirty_delete_object(Record)
catch
_C:_E -> ok
end. %%--------------------------------------------------------------------
%% @doc
%% @spec
%% @end
%%-------------------------------------------------------------------- %%%===================================================================
%%% Internal functions
%%%===================================================================
dynamic_db_init() ->
delete_schema(),
{atomic, ok} = mnesia:create_table(socket_to_pid, [{attributes, record_info(fields, socket_to_pid)}]),
ok. %% deletes a local schema.
delete_schema() ->
mnesia:stop(),
mnesia:delete_schema([node()]),
mnesia:start(). do(Query) ->
F = fun() -> qlc:e(Query) end,
{atomic, Value} = mnesia:transaction(F),
Value.