Function send_engine_events

sql/sql_repl.cc:3343–3483

* Send events from binlog implemented in storage engine. Will wait for more * data to become available as needed. * * return 0 - OK * else NOK */

 *        else NOK
 */
static int send_engine_events(binlog_send_info *info, LOG_INFO* linfo)
{
  int error;
  ulong ev_offset;

  String *packet= info->packet;
  handler_binlog_reader *reader= info->engine_binlog_reader;
  DBUG_ASSERT(reader);
  while (!should_stop(info))
  {
    /* reset the transmit packet for the event read from binary log
       file */
    if (reset_transmit_packet(info, info->flags, &ev_offset, &info->errmsg))
      return 1;

    error= reader->read_log_event(packet, packet->length(),
                                  info->thd->variables.max_allowed_packet);
    if (unlikely(error) && error != LOG_READ_EOF)
    {
      set_read_error(info, error);
      return 1;
    }

    uint64_t prev_file_no= linfo->file_no.load(std::memory_order_relaxed);
    if (unlikely(prev_file_no == ~(uint64_t)0) ||
                 unlikely(reader->cur_file_no > prev_file_no))
    {
      linfo->file_no.store(reader->cur_file_no, std::memory_order_relaxed);
      (*opt_binlog_engine_hton->get_filename)(info->log_file_name,
                                              reader->cur_file_no);
    }
    linfo->pos= (my_off_t) reader->cur_file_pos;

    if (error == LOG_READ_EOF)
    {
      /**
       * check if we should wait for more data
       */
      if ((info->flags & BINLOG_DUMP_NON_BLOCK) ||
          (info->thd->variables.server_id == 0))
      {
        info->should_stop= true;
        return 0;
      }

      /**
       * flush data before waiting
       */
      if (net_flush(info->net))
      {
        info->errmsg= "failed on net_flush()";
        info->error= ER_UNKNOWN_ERROR;
        return 1;
      }

      while (!should_stop(info, true) && !reader->data_available())
      {
        struct timespec ts;