Class: Mysql::FlydataJsonHandlerFactory::DatabaseFlydataJsonHandler

Inherits:
FlydataJsonHandler show all
Defined in:
lib/flydata/fluent-plugins/mysql/binlog_record_handler.rb

Constant Summary

Constants inherited from FlydataJsonHandler

FlydataJsonHandler::DB_NAME, FlydataJsonHandler::RESPECT_ORDER, FlydataJsonHandler::SEQ, FlydataJsonHandler::SRC_POS, FlydataJsonHandler::TABLE_NAME, FlydataJsonHandler::TABLE_REV, FlydataJsonHandler::TYPE, FlydataJsonHandler::V

Instance Method Summary collapse

Constructor Details

#initialize(records, record, context, type, increment_table_rev = nil) ⇒ DatabaseFlydataJsonHandler 



159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/flydata/fluent-plugins/mysql/binlog_record_handler.rb', line 159

def initialize(records, record, context, type, increment_table_rev=nil)
  super
  database = records.first[DB_NAME] || record['db_name']

  return unless acceptable_event?(type)

  position = record['next_position'] - record['event_length']

  # Add common information to each record
  records.each do |r|
    r[TYPE] = type
    r[RESPECT_ORDER] = true
    r[SRC_POS] = "#{@context.current_binlog_file}\t#{position}"
    r[V] = FlydataCore::Record::V2
  end

  # Use binlog's timestamp
  timestamp = record["timestamp"].to_i
  records.each do |row|
    emit(timestamp, row)
  end
end

Instance Method Details

#emit(timestamp, row) ⇒ Object 



182
183
184
 
# File 'lib/flydata/fluent-plugins/mysql/binlog_record_handler.rb', line 182

def emit(timestamp, row)
  Fluent::Engine.emit(@context.tag, timestamp, row)
end