pubsub.rb

require 'rubygems'
require 'ffi-rzmq'
require 'benchmark'
require 'json'

TESTS = 100_000

def error_check(rc)
  if ZMQ::Util.resultcode_ok?(rc)
    false
  else
    STDERR.puts "Operation failed, errno [#{ZMQ::Util.errno}] description [#{ZMQ::Util.error_string}]"
    caller(1).each { |callstack| STDERR.puts(callstack) }
    true
  end
end

@ctx = ZMQ::Context.create(1)
STDERR.puts "Failed to create a Context" unless @ctx

def publisher
  @pub_sock ||= @ctx.socket(ZMQ::PUB).tap {|pub_sock|
    error_check(pub_sock.setsockopt(ZMQ::LINGER, 1))
    rc = pub_sock.bind('tcp://192.168.36.88:2200')
    error_check(rc)    
  }
end

def publish(message)
  error_check(publisher.send_string(message))
end

def publish_benchmark(message)
  puts Benchmark.measure {
    publish('{start}')
    TESTS.times {
      publish(message)      
    }
    publish('{end}')
  }
end

def subscribe
  @sub_sock ||= @ctx.socket(ZMQ::SUB).tap {|sub_sock|
    error_check(sub_sock.setsockopt(ZMQ::LINGER, 1))
    rc = sub_sock.setsockopt(ZMQ::SUBSCRIBE,'{')
    error_check(rc)
    rc = sub_sock.connect('tcp://192.168.36.88:2200')
    error_check(rc)
  }
  count = 0
  message = ''  
  start_time = Time.now
  loop do
    @sub_sock.recv_string(message)

    if message.start_with?('{start')
      start_time = Time.now
      count = 0
    end
    count += 1
    if message.start_with?('{end')
      p count
      p Time.now-start_time
    end
  end
end

message = {action: :ad_updated, id: 'aosenuhtaosntuhasoneht'}.to_json