Contents

require 'flydata/fluent-plugins/mysql/alter_table_query_handler'

module Mysql
  class BinlogQueryDispatcher
    def initialize
      @handlers = []
    end

    def dispatch(record)
      normalize_query(record["query"]) do |query|
        @handlers.each do |handler|
          if (handler.pattern.match(query))
            handler.process(record.merge({"normalized_query" => query}))
            break
          end
        end
      end
    end

    private

    def normalize_query(queries)
      queries = strip_comments(queries)
      queries = queries.gsub(/\s+/, ' ') # replace multiple spaces with a space
      queries.split(';').each do |query|
        query = query.lstrip
        yield(query + ";") if block_given? && !query.empty?
      end
    end

    def strip_comments(query)
      q = query.dup
      # \/\*.*?\*\/        /* */ style comment
      # `.*?`              `resource_name`
      # '(?:\\.|.)*?'      'string'
      # "(?:\\.|.)*?"      "string"
      # --\s+.*?(?:\n|$)   -- style comment
      # #\s+.*?(?:\n|$)    # style comment
      query.scan(/(\/\*.*?\*\/|`.*?`|'(?:\\.|.)*?'|"(?:\\.|.)*?"|--\s+.*?(?:\n|$)|#.*?(?:\n|$))/m) do |m|
        comment_or_quoted = m.first
        if comment_or_quoted.start_with?("/*") || comment_or_quoted.start_with?("--") ||
           comment_or_quoted.start_with?("#")
          # comment. replace with spaces of the same length
          idx_from = $~.offset(0)[0]
          idx_to = $~.offset(0)[1]
          len = idx_to - idx_from
          q[idx_from...idx_to] = ' ' * len
        end
      end
      q
    end
  end

  class FlydataBinlogQueryDispatcher < BinlogQueryDispatcher
    def initialize(context)
      @handlers = [
        AlterTableQueryHandler.new(context),
      ]
    end
  end
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
flydata-0.3.11	lib/flydata/fluent-plugins/mysql/binlog_query_dispatcher.rb