zeromqインストール、twitter stream APIを中継

ZeroMQ Ruby

ぜろむきゅー

wget http://www.zeromq.org/local--files/area:download/zeromq-2.0.7.tar.gz
tar -zxvf zeromq-2.0.7.tar.gz
cd zeromq-2.0.7
./configure
make
sudo make install

ソースから入れてるのはC++ binding目当て



それからgem入れる

sudo gem install zmq

http://gist.github.com/498513 とか http://gist.github.com/498702 がrubyの参考になる


じゃあtwitterのstream APIをzeromqにして吐き出すpub側を作ってみる。stream APIは1台しか接続できないので、何かHUB的な物が欲しかった。

#!/usr/bin/env ruby
require 'rubygems'
require 'net/http'
require 'uri'
require 'json'
require 'yaml'
require 'kconv'
require 'zmq'
$KCODE = 'u'

conf = YAML::load open(File.dirname(__FILE__)+'/config.yaml')

ctx = ZMQ::Context.new
sock = ctx.socket(ZMQ::PUB)
sock.bind('tcp://127.0.0.1:5000')

uri = URI.parse('http://chirpstream.twitter.com/2b/user.json')
Net::HTTP.start(uri.host, uri.port) do |http|
  req = Net::HTTP::Get.new(uri.request_uri)
  req.basic_auth(conf['user'], conf['pass'])
  http.request(req){|res|
    next if !res.chunked?
    res.read_body{|chunk|
      status = JSON.parse(chunk) rescue next
      sock.send("chirp #{chunk}")
      begin
        puts "#{status['user']['screen_name']}: #{status['text']}"
      rescue => e
        p status
        STDERR.puts e
      end
    }
  }
end

http streamで受信したchunkをそのままzmqで送る。先頭にchirpって文字列を付けておく。



sub側

#!/usr/bin/env ruby
require 'rubygems'
require 'zmq'
$KCODE = 'u'

ctx = ZMQ::Context.new
sock= ctx.socket(ZMQ::SUB)
sock.connect('tcp://127.0.0.1:5000')
sock.setsockopt(ZMQ::SUBSCRIBE, 'chirp')

loop do
  puts sock.recv()
end

subはchirpでSUBSCRIBEする。subscribeは前方一致のみらしい。

きた。
chirpstream -> zeromq


sub側を複数増やしても全員受信できるし、接続が切れたり、pubとsubどっちを先に起動しても接続を適当に処理してくれるので気楽でいい。

chirp streamをとりあえずzeromqでpubしておいて、
sub側で

  • kanasayで読み上げる
  • growlする
  • DBに溜める
  • IMに転送

とか好きにやればいいんじゃないかと思った。



saykanaしてみた。
先頭の'chirp'は忘れず外して、chunkを復元する。JSONでparseして、saykanaが漢字を読めないのでmecabで全部カタカナにして読み上げさせた。

#!/usr/bin/env ruby
require 'rubygems'
require 'zmq'
require 'json'
require 'MeCab'
$KCODE = 'u'

mecab = MeCab::Tagger.new('-Ochasen')

ctx = ZMQ::Context.new
sock= ctx.socket(ZMQ::SUB)
sock.connect('tcp://127.0.0.1:5000')
sock.setsockopt(ZMQ::SUBSCRIBE, 'chirp')

loop do
  str = sock.recv()
  chunk = str.scan(/chirp (.+)/).first.first
  status = JSON.parse(chunk) rescue next
  begin
    puts "#{status['user']['screen_name']}: #{status['text']}"
  rescue => e
    p status
    STDERR.puts e
  end
  next unless status['text']
  begin
    str = status['text']
    puts kana = mecab.parse(str).map{|i|
      i.split(/\t/)[1]
    }.delete_if{|i|
      i.to_s.size < 1
    }.join('')
    `saykana '#{kana}'`
  rescue => e
    STDERR.puts e
  end
end