class Fluent::CassandraUpdatetor

Public Instance Methods

configure(conf) click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_cassandra_update.rb, line 33
def configure(conf)
  super

  @updateValue = self.update_value
  @whereCondUpd = self.where_condition_upd
end
format(tag, time, record) click to toggle source
# File lib/fluent/plugin/out_cassandra_update.rb, line 40
def format(tag, time, record)
  record.to_msgpack
end
shutdown() click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_cassandra_update.rb, line 28
def shutdown
  super
  @session.close if @session
end
start() click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_cassandra_update.rb, line 23
def start
  super
  @session ||= get_session(self.host, self.port, self.keyspace, self.connect_timeout, self.username, self.password)
end
write(chunk) click to toggle source
# File lib/fluent/plugin/out_cassandra_update.rb, line 44
def write(chunk)
  chunk.msgpack_each { |record|

    whereCondition = prepareParameter(@whereCondUpd, record)

    @updateValue = prepareParameter(@updateValue, record)
    updateCassandra(@updateValue, whereCondition)

  }
end

Private Instance Methods

prepareParameter(strOri,record) click to toggle source
# File lib/fluent/plugin/out_cassandra_update.rb, line 72
def prepareParameter(strOri,record)
  tmpCondVal = {}
  tmpStr = nil
  count = 0

  strOri.split(":").each do |str|
    if count > 0
      tmpStr = str.gsub(/(;.*)/, '')
      tmpCondVal[tmpStr] = record[tmpStr]
    end
    count += 1
  end

  tmpCondVal.each do |k,v|
    strOri= strOri.gsub(k,v)
  end

  strOri = strOri.gsub(':','')
  strOri = strOri.gsub(';','')

  strOri
end
updateCassandra(updateVal, whereCondition) click to toggle source
# File lib/fluent/plugin/out_cassandra_update.rb, line 57
def updateCassandra(updateVal, whereCondition)

  cql = "update #{self.keyspace}.#{self.tablename} set "
  cql += updateVal + " where " + whereCondition + ";"

  print cql
  begin
    @session.execute(cql)
  rescue Exception => e
    $log.error "Cannot update record Cassandra: #{e.message}\nTrace: #{e.backtrace.to_s}"

    raise e
  end
end