class Fluent::Plugin::TwitterInput
Constants
- OUTPUT_FORMAT_TYPE
- TIMELINE_TYPE
Public Class Methods
new()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/in_twitter.rb, line 35 def initialize super @running = false end
Public Instance Methods
array_key_to_s(array)
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 154 def array_key_to_s(array) array.map do |v| if v.instance_of?(Hash) hash_key_to_s(v) elsif v.instance_of?(Array) array_key_to_s(v) elsif v.instance_of?(String) v.scrub('') else v end end end
configure(conf)
click to toggle source
Calls superclass method
# File lib/fluent/plugin/in_twitter.rb, line 40 def configure(conf) super @keyword = @keyword.gsub('${hashtag}', '#') unless @keyword.nil? @client = Twitter::Streaming::Client.new do |config| config.consumer_key = @consumer_key config.consumer_secret = @consumer_secret config.access_token = @access_token config.access_token_secret = @access_token_secret config.proxy = @proxy.to_h if @proxy end end
get_message(tweet)
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 104 def get_message(tweet) case @output_format when :nest record = hash_key_to_s(tweet.to_h) when :flat record = hash_flatten(tweet.to_h) when :simple record = {} record['message'] = tweet.text.scrub('') record['geo'] = tweet.geo record['place'] = tweet.place record['created_at'] = tweet.created_at.to_s record['user_name'] = tweet.user.name record['user_screen_name'] = tweet.user.screen_name record['user_profile_image_url'] = tweet.user.profile_image_url.to_s record['user_time_zone'] = tweet.user.time_zone record['user_lang'] = tweet.user.lang end router.emit(@tag, Fluent::Engine.now, record) end
handle_object(object)
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 86 def handle_object(object) return unless @running if is_message?(object) get_message(object) end end
hash_flatten(record, prefix = nil)
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 125 def hash_flatten(record, prefix = nil) record.inject({}) do |d, (k, v)| k = prefix.to_s + k.to_s if v.instance_of?(Hash) d.merge(hash_flatten(v, k + @flatten_separator)) elsif v.instance_of?(String) d.merge(k => v.scrub("")) else d.merge(k => v) end end end
hash_key_to_s(hash)
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 138 def hash_key_to_s(hash) newhash = {} hash.each do |k, v| if v.instance_of?(Hash) newhash[k.to_s] = hash_key_to_s(v) elsif v.instance_of?(Array) newhash[k.to_s] = array_key_to_s(v) elsif v.instance_of?(String) newhash[k.to_s] = v.scrub('') else newhash[k.to_s] = v end end newhash end
is_message?(tweet)
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 93 def is_message?(tweet) return false if !tweet.is_a?(Twitter::Tweet) return false if (!@lang.nil? && @lang != '') && !@lang.include?(tweet.user.lang) if @timeline == :userstream && (!@keyword.nil? && @keyword != '') pattern = NKF::nkf('-WwZ1', @keyword).gsub(/,\s?/, '|') tweet = NKF::nkf('-WwZ1', tweet.text) return false if !Regexp.new(pattern, Regexp::IGNORECASE).match(tweet) end return true end
run()
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 67 def run notice = "twitter: starting Twitter Streaming API for #{@timeline}." notice << " tag:#{@tag}" notice << " lang:#{@lang}" unless @lang.nil? notice << " keyword:#{@keyword}" unless @keyword.nil? notice << " follow:#{@follow_ids}" unless @follow_ids.nil? && !@keyword.nil? log.info notice if [:sampling, :tracking].include?(@timeline) && @keyword @client.filter(track: @keyword, &method(:handle_object)) elsif @timeline == :tracking && @follow_ids @client.filter(follow: @follow_ids, &method(:handle_object)) elsif @timeline == :sampling && @keyword.nil? && @follow_ids.nil? @client.sample(&method(:handle_object)) elsif @timeline == :userstream @client.user(&method(:handle_object)) end end
shutdown()
click to toggle source
Calls superclass method
# File lib/fluent/plugin/in_twitter.rb, line 61 def shutdown @running = false @client.close if @client.respond_to?(:close) super end
start()
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 54 def start @running = true thread_create(:in_twitter) do run end end