ws_example.erl

websocket_handle_incoming_data(Data, #state{ws = WsState = #websocket_state{module = Mod, parser_state = WsParserState}, socket=Socket} = State) ->
    WsCallback = fun handle_stream/2,
    case Mod:handle_data(Data, WsParserState, {Socket, http}, State, WsCallback) of
        {State2, websocket_close} ->
            {stop, normal, State2};
        {State2, websocket_close, CloseData} ->
            State3 = websocket_send(CloseData, State2),
            {stop, normal, State3};
        {State2, continue, WsParsesState2} ->
            inet:setopts(Socket, [{packet, raw}, {active, once}]),
            {next_state, websocket, State2#state{ws = WsState#websocket_state{parser_state = WsParsesState2}}}
    end.


handle_stream(Bin, #state{counter = C} = State) when is_binary(Bin) ->
    {struct, L} = mochijson2:decode(Bin),
    Action = proplists:get_value(<<"action">>, L),
    Data = proplists:get_value(<<"data">>, L),
    Token = proplists:get_value(<<"token">>, L),
    case Action of
        <<"publish">> ->
            topic:publish_json(self(), Data),
            State#state{counter + 1};
    end.