Чего не хватает для разбора udp?

[Анастасия Субботина]

Чего не хватает для разбора udp  в следующем коде
-module(fast_server).
-behaviour(gen_server).

-compile(export_all).

-include("../include/fast_context.hrl").


start_link() ->
    gen_server:start_link(?MODULE, [], []).

root() ->
  case code:lib_dir(fast) of
    {error, _Error} -> filename:dirname(code:which(?MODULE)) ++ "/../";
    Root_ when is_list(Root_) -> Root_
  end.

init([]) ->
   
    Source = "172.27.129.77",
    Port = "24027",
    Group = "239.192.7.27",

    {ok,SourceAddress} = inet:parse_address(Source),
    {ok,GroupAddress} = inet:parse_address(Group),
        GroupIp = ip_to_binary(GroupAddress),
        LocalIp = ip_to_binary({0,0,0,0}),
        SourceIp = ip_to_binary(SourceAddress),
        Bin = << GroupIp/binary,LocalIp/binary,SourceIp/binary >>,

    {ok, Socket} = gen_udp:open(erlang:list_to_integer(Port),
      [
         inet,
         binary,
         {active, true},
         {reuseaddr, true},
    {multicast_ttl, 30},
         {raw, 0, 39, Bin}
      ]),

    ok = gen_udp:controlling_process(Socket, self()), 
    F = fun([], _) ->
        ok;
    (Err, Val) ->
        io:format("~p ~p~n", [Err, Val])
    end,
    try
              {Dicts, Templates} = fast_xml:parse({file, root() ++"/spec/templates.xml"}, []),
              {ok, #context{dicts = Dicts, templates = Templates, logger = F, options = [], socket = Socket}}
       catch
              _:Err ->
                 Err
       end.

ip_to_binary(Ip) ->
    list_to_binary(tuple_to_list(Ip)).

handle_info({udp, _Socket, _Ip, _Port, Bin}, Context = #context{}) ->
    case fast_segment:decode(Bin, Context) of
        {ok, {TemplateName, Msg, Rest, _}} ->
            io:format("~p ~p~n", [TemplateName, Msg]),
            {ok, { _, _, _, Context1}} = fast_segment:decode(Rest, Context),% ошибка в этой стоке no_enough_data
            handle_info({udp, _Socket, _Ip, _Port, Rest}, Context1);
        {error, Reason} ->
            io:format("ERROR: ~p, Bin = ~p ~n", [Reason, Bin]), %39
            exit(failed)
       
    end;
handle_info(_Info, State) ->
    {noreply, State}.

terminate(normal, State) ->
    {noreply, State}.

то есть обрабатывается только первое сообщение и после него вылетает ошибка.

[Danil A. Zagoskin]

Не описан тривиальный случай, который сожрет пустой бинарь.

--
Вы получили это сообщение, поскольку подписаны на группу "Erlang по-русски".
Чтобы отменить подписку на эту группу и больше не получать от нее сообщения, отправьте письмо на электронный адрес erlang-russia...@googlegroups.com.
Чтобы отправлять сообщения в эту группу, отправьте письмо на электронный адрес erlang-...@googlegroups.com.
Чтобы настроить другие параметры, перейдите по ссылке https://groups.google.com/d/optout.

[Анастасия Субботина]

ни
handle_info(<<>>, _) ->
    ok;
ни
handle_info({udp, _Socket, _Ip, _Port, <<>>}, _) ->
    ok;
ни до, ни после

handle_info({udp, _Socket, _Ip, _Port, Bin}, Context = #context{}) ->
    case fast_segment:decode(Bin, Context) of
        {ok, {TemplateName, Msg, Rest, _}} ->
            io:format("~p ~p~n", [TemplateName, Msg]),

            {ok, { _, _, _, Context1}} = fast_segment:decode(Rest, Context),% ошибка в этой стоке not_enough_data

            handle_info({udp, _Socket, _Ip, _Port, Rest}, Context1);
       
        {error, Reason} ->
            io:format("ERROR: ~p, Bin = ~p ~n", [Reason, Bin]), %39
            exit(failed)
       
    end;

ситуацию не меняют, ошибка все та же
понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

[Andy]

Здравствуйте, Анастасия!

Если этот вопрос еще актуален (а судя по Вашему репозиторию это так), могу высказать свои предположения. Поскольку я не знаком с протоколом, но подобную задачу я решал (разбор пакета на блоки данных, которые могут разбиваться между пакетами), то предположу, что второй вызов fast_segment:decode либо не нужен по причине того, что все возможные данные уже обработаны, а оставшийся буфер (начало другого сегмента) надо будет склеить со следующим пакетом и передать в декодер, для чего сохранить его в состоянии генсервера. Либо же надо принимать и четвертый параметр ответа функции (обновленный контекст) и передавать его даже не во второй вызов функции, а организовать рекурсию, пока не станет возвращать not_enough_data, что и станет критерием перехода к первому совету (склеиванию пакетов). Общими словами: представьте себе несколько пакетов UDP, в которых лежат несколько сегментов данных, и попытайтесь обработать все возможные варианты.

Немного оффтопика: похоже, что Эрланг-программистов у нас вполне хватает. Послал резюме в одну фирму, которая ищет программистов на Эрланге для подобных задач (не там ли Вы работаете?) - статус просмотра так и висит в "непросмотренные". Хотя дата вакансии регулярно обновляется...

P.S.: Перепроверил - просмотрено, но не отвечено... Все равно вопрос остается

С уважением,
Андрей

понедельник, 13 октября 2014 г., 16:04:59 UTC+3 пользователь Анастасия Субботина написал:

[Анастасия Субботина]

Context - это состояние, которое сохраняет путь к шаблону xml по которому надо декодировать сообщения, в том виде в котором сейчас это происходит было лишним его добавление к состоянию сервера (на мой взгляд), если в результате оптимизации будет найден другой подход декодирования, то будем искать в том направлении. Сейчас первоочередная задача состоит в том, чтобы собрать работающий образец. Этот проект к моей работе никак не относится.

понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

Чего не хватает для разбора udp  в следующем коде

[Andy]

Ваше дело, конечно, но автор библиотеки не зря контекст не только принимает, но и возвращает. Успехов Вам.

понедельник, 3 ноября 2014 г., 8:02:33 UTC+2 пользователь Анастасия Субботина написал:

[Анастасия Субботина]

Чего не хватает в следующем коде чтобы сегмент можно было декодировать:
handle_info({udp, _Socket, _Ip, _Port, << _Head:32, MsgSeqNum:16/binary, Bin/binary>>}, #server{name = Name, seq = SeqNum}) ->
    io:format("~p~n", [MsgSeqNum]),
    case fast_decode_types:decode_type(uInt32, MsgSeqNum, true) of
        {SeqNum, _L, _Nullable} ->
            io:format("~p~n", [SeqNum]),
            {noreply,#server{buffer = Bin, name = Name, seq = SeqNum + 1}};
        _ ->
            {noreply,#server{buffer = Bin, name = Name, seq = SeqNum + 1}}
    end;
MsgSeqNum даже не появляется на печати. Пакет имеет следующий вид:
<<192,248,192,172,10,162,35,101,88,109,47,14,7,181>>
я пытаюсь декодировать 5 и 6 байты (<<10, 162>>), которые по спецификации являются одной переменной MsgSeqNum?

понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

Чего не хватает для разбора udp  в следующем коде

[Andrew Gopienko]

MsgSeqNum:16/binary

может быть

MsgSeqNum:16/bytes ?

4 мая 2015 г., 16:05 пользователь Анастасия Субботина <spasen...@gmail.com> написал:

--

[Andrew Gopienko]

Скорее всего для номера сообщения просто MsgSeqNum:16,...

4 мая 2015 г., 16:36 пользователь Andrew Gopienko <gopi...@gmail.com> написал:

[Анастасия Субботина]

 MsgSeqNum:16/bytes, ничего не изменилось, MsgSeqNum:16, один раз срабатывает и вылетает с ошибкой.

[Dmitry Belyaev]

Для типа binary по умолчанию длина указывается в байтах, в отличие от целых, для которых в обычно она указывается в битах. Весь клоз не срабатывает потому что входных данных недостаточно - нужно 4+16 байт минимум.
--
Best wishes,
Dmitry Belyaev

[Анастасия Субботина]

То есть необходимо добавить за счет другого сегмента или как то иначе это решается?

понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

Чего не хватает для разбора udp  в следующем коде

[Анастасия Субботина]

Я изменила на << _Head:32, MsgSeqNum:2/bytes, Bin/bytes >> печать MsgSeqNum пошла:
<<14,211>> из <<192,248,192,172,14,211,35,101,88,109,70,1,83,229>>, но вот декодирование и получение SeqNum не получается.

понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

Чего не хватает для разбора udp  в следующем коде

[Анастасия Субботина]

В результате поисков получила следующее:
handle_info({udp, _Socket, _Ip, _Port, << _Head:32, MsgSeqNum:16/bits, Bin/bytes>>}, #server{name = Name, seq = SeqNum}) ->

    io:format("~p~n", [MsgSeqNum]),

    Null = <<0>>,
    << Seq:24/bits>> = << Null:8/bits, MsgSeqNum:16/bits>>,
    Qwe = fast_decode:decode_type( Seq),
    io:format("~p~n", [Qwe]),
        case Qwe of
        {SeqNum, _L, _Rest} ->

            io:format("~p~n", [SeqNum]),
            {noreply,#server{buffer = Bin, name = Name, seq = SeqNum + 1}};
        _ ->
            {noreply,#server{buffer = Bin, name = Name, seq = SeqNum + 1}}
    end;

печать идет с первых двух io:format, то есть
<<17,178>>
{2225,[],<<>>}
при этом
case Qwe of
        {SeqNum, _L, _Rest}
не работает, хотя Qwe = {2229,[],<<>>},
В чем ошибка?

понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

Чего не хватает для разбора udp  в следующем коде

[Ilya I. Ashchepkov]

У вас SeqNum определена выше и, вероятно, не совпадает с содержащейся в Qwe

--

Вы получили это сообщение, поскольку подписаны на группу "Erlang по-русски".
Чтобы отменить подписку на эту группу и больше не получать от нее сообщения, отправьте письмо на электронный адрес erlang-russia...@googlegroups.com.
Чтобы отправлять сообщения в эту группу, отправьте письмо на электронный адрес erlang-...@googlegroups.com.
Чтобы настроить другие параметры, перейдите по ссылке https://groups.google.com/d/optout.

--

С уважением,
Ащепков Илья koc...@gmail.com

[Mikl Kurkov]

Анастасия, а вы не FAST/FIX протокол таким образом пытаетесь парсить?

В этом случае вы все делаете не так. Протокол не использует поля фиксированной ширины, а выставляет стоп бит (первый в каждом байте) как признак того что данные для поля закончены.

Вот прототип кода который парсит такой формат:

F = fun(<<0:1, C:7>>, [Cur|Acc]) -> [<<0:1,Cur/binary,C:7>> | Acc];
       (<<1:1, C:7>>, [Cur|Acc]) -> [<<>>, binary:decode_unsigned(<<0:1,Cur/binary,C:7>>) | Acc] end.

Parse = fun(Pkt) -> lists:reverse(lists:foldl(F, [<<>>], [<<C:8>> || <<C:8>> <= Pkt])) end.

Для ваших данных он выдает

Parse(<<192,248,192,172,14,211,35,101,88,109,70,1,83,229>>).
[64,120,64,44,1875,20150504130636261,<<>>]

Похоже что тут два пакета, один 120 типа без полей, второй 44 типа с двумя полями - Id и Timestamp.

P.S. с FAST/FIX никогда не сталкивался, так что кто в теме поправляйте

Mikl

[Анастасия Субботина]

Спасибо всем. Да, это FAST протокол, Mikl Kurkov этот фрагмент кода, который вы показали, он из открытых источников?

понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

Чего не хватает для разбора udp  в следующем коде

[Анастасия Субботина]

Итоговый код следующий:

handle_info({udp, _Socket, _Ip, _Port, << _Head:32, MsgSeqNum:16/bits, Bin/bytes>>}, #server{name = Name, seq = SeqNum}) ->

    { Seq, _L, _Rest} = fast_decode:decode_type( MsgSeqNum),
           case Seq of
        SeqNum ->

            io:format("~p~n", [SeqNum]),
            {noreply,#server{buffer = Bin, name = Name, seq = SeqNum + 1}};
        _ ->

            {noreply,#server{buffer = Bin, name = Name, seq = Seq + 1}}
    end;
Предпологалось, что если я запущу получение данных из 2-ух источников, то первый обработанный пакет будет увеличивать SeqNum на 1 и второй будет игнорироваться, но печать говорит о том, что в первом условии срабатывают оба пакета, почему так?
4716
4716
4717
4717

понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

Чего не хватает для разбора udp  в следующем коде

[Mikl Kurkov]

Код из головы, думаю можно считать что из открытых.
Не советую использовать его для обработки реальных данных, это только прототип для демонстрации работы со стоп-битом.

--

Вы получили это сообщение, поскольку подписаны на группу "Erlang по-русски".
Чтобы отменить подписку на эту группу и больше не получать от нее сообщения, отправьте письмо на электронный адрес erlang-russia...@googlegroups.com.
Чтобы отправлять сообщения в эту группу, отправьте письмо на электронный адрес erlang-...@googlegroups.com.
Чтобы настроить другие параметры, перейдите по ссылке https://groups.google.com/d/optout.

--

Mikl

[Mikl Kurkov]

Проблема во второй клаузе case думаю, зачем там используется номер игнорируемого пакета?

Из-за этого если мы после обработки 4716 ждем 4717, а прилетит 4715 мы не проигнорируем его, а снова станем ждать 4716, вот и дубль. UDP пакеты могут приходить в случайном порядке, так что надо как-то обработать пакеты из будущего по-хорошему.

Ну и синтаксис рекордов поддерживает обновление отдельных полей - не обязательно каждый раз его целиком копировать:

handle_info({udp, _Socket, _Ip, _Port, << _Head:32, MsgSeqNum:16/bits, Bin/bytes>>}, #server{seq = SeqNum} = State) ->

    { Seq, _L, _Rest} = fast_decode:decode_type( MsgSeqNum),
    case Seq of
        SeqNum ->
            io:format("~p~n", [SeqNum]),

            {noreply,State#server{seq = SeqNum + 1}};
        _ ->
            {noreply,State}
    end;

--

Вы получили это сообщение, поскольку подписаны на группу "Erlang по-русски".
Чтобы отменить подписку на эту группу и больше не получать от нее сообщения, отправьте письмо на электронный адрес erlang-russia...@googlegroups.com.
Чтобы отправлять сообщения в эту группу, отправьте письмо на электронный адрес erlang-...@googlegroups.com.
Чтобы настроить другие параметры, перейдите по ссылке https://groups.google.com/d/optout.

--

Mikl

[Анастасия Субботина]

Вторая клауза нужна для присваивания seq текущего номера сообщений, по другому трудно будет угадать.

понедельник, 13 октября 2014 г., 13:19:12 UTC+4 пользователь Анастасия Субботина написал:

Чего не хватает для разбора udp  в следующем коде