Библиотека для обмена сообщениями

static void read_data(connection_info_t * c) {
    size_t reserved;

    uint8_t* data = bip_buffer_reserve(&c->recv_buffer, CONNECTION_RECV_BUFFER, &reserved);
    ssize_t recv_len = recv(c->fd, data, reserved, 0);

    rx_log_write(RX_LOG_PARANOID, "recv %ld bytes from socket. %s", recv_len, get_connection_name(c));

    if (recv_len < 0) {
        bip_buffer_commit(&c->recv_buffer, 0);
        if (errno != EAGAIN) {
            rx_log_write(RX_LOG_ERROR | RX_LOG_ERRNO, "recv failed. %s", get_connection_name(c));
            destroy_connection(c);
        }
        return;
    }
    bip_buffer_commit(&c->recv_buffer, recv_len);
    if (recv_len == 0 && errno == EINTR) {
        return;
    }
    if (recv_len == 0) {
        rx_log_write(RX_LOG_INFO | RX_LOG_ERRNO, "recv return zero bytes. Close connection %s", get_connection_name(c));
        destroy_connection(c);
        return;
    }
    c->shm_connection->recv_bytes += recv_len;

    while (bip_buffer_get_data_size(&c->recv_buffer) > sizeof (diameter_cmd_t)) {
        size_t size;
        uint32_t len;
        const diameter_len_header_t* header = (const diameter_len_header_t*) bip_buffer_get_contiguous_block(&c->recv_buffer, &size);

        if (size >= sizeof (diameter_len_header_t)) {
            len = hton24(header->length);
        } else {
            diameter_len_header_t* a_header = (diameter_len_header_t*) alloca(sizeof (diameter_len_header_t));
            if (RETCODE_OK != bip_buffer_peek(&c->recv_buffer, a_header, sizeof (diameter_len_header_t))) {
                break;
            }
            len = hton24(a_header->length);
        }

        if (len > DIAMETER_MAX_MESSAGE_SIZE) {
            rx_log_write(RX_LOG_ERROR, "recv too large diameter msg from %s", get_connection_name(c));
            destroy_connection(c);
            return;
        }

        const diameter_cmd_t* cmd;
        if (size >= len) {
            cmd = (const diameter_cmd_t*) header;
        } else {
            if (bip_buffer_get_data_size(&c->recv_buffer) < len) {
                break;
            }
            diameter_cmd_t* a_cmd = (diameter_cmd_t*) alloca(len);
            if (RETCODE_OK != bip_buffer_peek(&c->recv_buffer, a_cmd, len)) {
                break;
            }
            cmd = a_cmd;
        }
        if (RETCODE_OK != process_in_diameter_message(c, cmd)) {
            return;
        }
        ++c->shm_connection->recv_messages;
        bip_buffer_free(&c->recv_buffer, len);

        rx_log_write(RX_LOG_DEBUG, "recv full diameter msg from %s"
                " (data in buffer=%ld, header.len=%u)",
                get_connection_name(c), bip_buffer_get_data_size(&c->recv_buffer), len);
    }

    if (bip_buffer_get_data_size(&c->recv_buffer)) {
        rx_log_write(RX_LOG_DEBUG,
                "recv part diameter msg from %s (len=%ld, rec_len=%lu)",
                get_connection_name(c),
                bip_buffer_get_data_size(&c->recv_buffer), recv_len);
    }
}

while (true) {
  size_t bytes_read =0;
  unsigned packet_len;
  char *buf = (char *)&packet_len;
  while (bytes_read < 4) {
    ssize_t ret = read(skt, buf+bytes_read, 4 - bytes_read);
    if (ret <= 0) {
      throw ("пока!");
    }
    bytes_read += ret;
  }
  packet_len = ntohl(packet_len);
  std::unique_ptr <unsigned char *> packet(new unsigned char [packec_len]);
  bytes_read = 0;
  while (bytes_read < packet_len) {
    ssize_t ret = read(skt, &packet[0] + bytes_read, packet_len - bytes_read);
    if (ret <= 0) {
      throw ("пока!");
    }
    bytes_read += ret;
  }
  puts("Увага! Нью пакет дід кам!");
  process_packet_and_write_another_100_lines_of_code(packet, packet_len);
}

-module(tcp_server).
-export([start/0, init/0, acceptor/1, servant/1]).

start() ->
  P = spawn_link(?MODULE, init, []),
  {ok, P}.

init() ->
  process_flag(trap_exit, true),
  {ok, Srv} = gen_tcp:listen(3000, [binary, {packet, 4}, {active, true}]),
  Acceptor = spawn_link(?MODULE, acceptor, [Srv, self()]),
  loop(Acceptor, Srv).

loop(Acceptor, Srv) ->
  receive
    {'EXIT', Acceptor, shutdown} ->
      error_logger:warning_msg("Tcp acceptor down, restarting~n"),
      loop(spawn_link(?MODULE, acceptor, [Srv]), Srv);
    {'EXIT', _From, shutdown} ->
      gen_tcp:close(Srv),
      ok;
    _ ->
      loop(Acceptor, Srv)
  end.

acceptor(Srv) ->
  case get_tcp:accept(Srv) of
    {ok, Skt} ->
      {ok, Pid} = spawn_link(?MODULE, servant, [Skt]),
      gen_tcp:controlling_process(Skt, Pid),
      acceptor(Srv);
    {error, closed} ->
      exit(normal)
  end.

servant(Skt) ->
  receive
    {tcp, Skt, Request} ->
      gen_server:call(fpga_server, {fpga_req, {raw, Request}}),
      receive
        {fpga_rep, {raw, Reply}} ->
          gen_tcp:send(Skt, Reply);
        {error, R} ->
          gen_tcp:close(Skt),
          exit(R)
      end,
      servant(Skt);
    {tcp_closed, Skt} ->
      gen_tcp:close(Skt),
      exit(normal);
    {'EXIT', _From, shutdown} ->
      exit(normal)
  end.

<<Len:32/little, _/binary>> = Message.

<<Len:32/big, _/binary>> = Message.