Class: Fluent::Plugin::SQLInput::TableElement
- Inherits:
-
Object
- Object
- Fluent::Plugin::SQLInput::TableElement
- Includes:
- Configurable
- Defined in:
- lib/fluent/plugin/in_sql.rb
Instance Attribute Summary collapse
-
#log ⇒ Object
readonly
Returns the value of attribute log.
Instance Method Summary collapse
- #configure(conf) ⇒ Object
-
#emit_next_records(last_record, limit) ⇒ Object
emits next records and returns the last record of emitted records.
- #init(tag_prefix, base_model, router, log) ⇒ Object
-
#normalized_time(tv, now) ⇒ Object
Make sure we always have a Fluent::EventTime object regardless of what comes in.
Instance Attribute Details
#log ⇒ Object (readonly)
Returns the value of attribute log.
63 64 65 |
# File 'lib/fluent/plugin/in_sql.rb', line 63 def log @log end |
Instance Method Details
#configure(conf) ⇒ Object
65 66 67 |
# File 'lib/fluent/plugin/in_sql.rb', line 65 def configure(conf) super end |
#emit_next_records(last_record, limit) ⇒ Object
emits next records and returns the last record of emitted records
129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 |
# File 'lib/fluent/plugin/in_sql.rb', line 129 def emit_next_records(last_record, limit) relation = @model if last_record && last_update_value = last_record[@update_column] relation = relation.where("#{@update_column} > ?", last_update_value) end relation = relation.order("#{@update_column} ASC") relation = relation.limit(limit) if limit > 0 now = Fluent::Engine.now me = Fluent::MultiEventStream.new relation.each do |obj| record = obj.serializable_hash rescue nil if record time = if @time_column && (tv = obj.read_attribute(@time_column)) normalized_time(tv, now) else now end me.add(time, record) last_record = record end end last_record = last_record.dup if last_record # some plugin rewrites record :( @router.emit_stream(@tag, me) return last_record end |
#init(tag_prefix, base_model, router, log) ⇒ Object
69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 |
# File 'lib/fluent/plugin/in_sql.rb', line 69 def init(tag_prefix, base_model, router, log) @router = router @tag = "#{tag_prefix}.#{@tag}" if tag_prefix @log = log # creates a model for this table table_name = @table primary_key = @primary_key time_format = @time_format @model = Class.new(base_model) do self.table_name = table_name self.inheritance_column = '_never_use_' self.primary_key = primary_key if primary_key self.const_set(:TIME_FORMAT, time_format) #self.include_root_in_json = false def read_attribute_for_serialization(n) v = send(n) if v.respond_to?(:to_msgpack) v elsif v.is_a? Time v.strftime(self.class::TIME_FORMAT) else v.to_s end end end # ActiveRecord requires model class to have a name. class_name = table_name.gsub(/\./, "_").singularize.camelize base_model.const_set(class_name, @model) # Sets model_name otherwise ActiveRecord causes errors model_name = ActiveModel::Name.new(@model, nil, class_name) @model.define_singleton_method(:model_name) { model_name } # if update_column is not set, here uses primary key unless @update_column pk = @model.columns_hash[@model.primary_key] unless pk raise "Composite primary key is not supported. Set update_column parameter to <table> section." end @update_column = pk.name end end |
#normalized_time(tv, now) ⇒ Object
Make sure we always have a Fluent::EventTime object regardless of what comes in
118 119 120 121 122 123 124 125 126 |
# File 'lib/fluent/plugin/in_sql.rb', line 118 def normalized_time(tv, now) return Fluent::EventTime.from_time(tv) if tv.is_a?(Time) begin Fluent::EventTime.parse(tv.to_s) rescue log.warn "Message contains invalid timestamp, using current time instead (#{now.inspect})" now end end |