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 28 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 145 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 33 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 end end
get_message(tweet)
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 95 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 record['user_name'] = tweet.user.name record['user_screen_name'] = tweet.user.screen_name record['user_profile_image_url'] = tweet.user.profile_image_url 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 77 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 116 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 129 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 84 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 58 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 53 def shutdown @running = false super end
start()
click to toggle source
# File lib/fluent/plugin/in_twitter.rb, line 46 def start @running = true thread_create(:in_twitter) do run end end