class Fluent::CassandraInsertor

Public Instance Methods

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

  @insertValue = self.insert_value
end
format(tag, time, record) click to toggle source
# File lib/fluent/plugin/out_cassandra_insert.rb, line 38
def format(tag, time, record)
  record.to_msgpack
end
shutdown() click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_cassandra_insert.rb, line 27
def shutdown
  super
  @session.close if @session
end
start() click to toggle source
Calls superclass method
# File lib/fluent/plugin/out_cassandra_insert.rb, line 22
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_insert.rb, line 42
def write(chunk)
  chunk.msgpack_each { |record|
    @insertValue = prepareParameter(@insertValue, record)
    insertCassandra(@insertValue)
  }
end

Private Instance Methods

insertCassandra(insertVal) click to toggle source
# File lib/fluent/plugin/out_cassandra_insert.rb, line 51
def insertCassandra(insertVal)
  colIns = []
  valIns = []
  tmpStr = nil

  insertVal.split(",").each do |str|
    tmpStr = str.split("=")
    colIns.push(tmpStr[0])
    valIns.push(tmpStr[1])
  end

  cql = "INSERT INTO #{self.keyspace}.#{self.tablename} (#{colIns.join(',')}) VALUES (#{valIns.join(',')});"

  print cql

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

    raise e
  end
end
prepareParameter(strOri,record) click to toggle source
# File lib/fluent/plugin/out_cassandra_insert.rb, line 75
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