Sha256: 0925e96efe35de6147eed94dd5cf9ef04458ea3a15e9d30c1735d319c5d28b7f
Contents?: true
Size: 1.43 KB
Versions: 43
Compression:
Stored size: 1.43 KB
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