Contents

require 'fluent/plugin/in_mysql_binlog'
require 'binlog'
require_relative 'dml_record_handler'
require_relative 'binlog_query_dispatcher'

module Mysql
  class BinlogRecordDispatcher
    def dispatch(event)
      method_name = "on_#{event.event_type.downcase}"
      if self.respond_to?(method_name)
        # TODO to_hash method call below can fail if event.event_type is
        # "Update_rows".  This seems to be a bug of ruby-binlog.  The bug must
        # be fixed when we support record update.
        record = Fluent::MysqlBinlogInput::BinlogUtil.to_hash(event)
        self.send(method_name, record)
      else
        # $log.trace "Unhandled type: #{record["event_type"]}"
      end
    end
  end

  class FlydataBinlogRecordDispatcher < BinlogRecordDispatcher
    def initialize(context)
      context.current_binlog_file = ""
      @context = context
      @query_dispatcher = FlydataBinlogQueryDispatcher.new(context)
      @dml_record_handler = DmlRecordHandler.new(context)
    end

    def on_rotate(record)
      @context.current_binlog_file = record["binlog_file"]
    end

    def on_write_rows(record)
      @dml_record_handler.process(record, :write_rows)
    end

    def on_update_rows(record)
      @dml_record_handler.process(record, :update_rows)
    end

    def on_delete_rows(record)
      @dml_record_handler.process(record, :delete_rows)
    end

    def on_query(record)
      @query_dispatcher.dispatch(record)
    end
  end
end

Version data entries

43 entries across 43 versions & 1 rubygems

Version	Path
flydata-0.1.10	lib/flydata/fluent-plugins/mysql/binlog_record_dispatcher.rb
flydata-0.1.9	lib/flydata/fluent-plugins/mysql/binlog_record_dispatcher.rb
flydata-0.1.8	lib/flydata/fluent-plugins/mysql/binlog_record_dispatcher.rb