@@ -12,35 +12,38 @@ module SMQueue super options = args.first @configuration = options[:configuration] + @client_thread = nil end def put(*args, &block) - EM.run { - broker = MQ.new - broker.queue(@configuration[:queue]).publish(args[0]) - EM.add_timer(0.1) { EM.stop } - } + connect + channel.publish(args[0]) end def get(*args, &block) if block_given? EM.run { - broker = MQ.new - broker.queue(@configuration[:queue]).subscribe { |header, body| + channel.subscribe { |header, body| yield ::SMQueue::Message(:headers => header, :body => body) } } else - @message = nil - EM.run { - broker = MQ.new - broker.queue(@configuration[:queue]).subscribe { |header, body| - @message = ::SMQueue::Message(:headers => header, :body => body) - EM.stop - } - } - @message + raise "TODO: Implement me" end end + + private + def connect + current_thread = Thread.current + @client_thread ||= Thread.new { + AMQP.start { + current_thread.wakeup + } + } + end + + def channel + MQ.queue(@configuration[:queue]) + end end end \ No newline at end of file lib/smqueue/adapters/amqp_adapter.rb fa791306c665508c1f81955cb3825f1f7c889388 Craig R Webster craig@barkingiguana.com http://github.com/seanohalpin/smqueue/commit/c4c0f4b46339469d9291d3e75f070c5ef4433a06 c4c0f4b46339469d9291d3e75f070c5ef4433a06 2009-11-09T09:53:40-08:00 2009-08-13T12:37:03-07:00 Don't eat all the sockets on the machine. That's just not nice. 112d7cf4f99ae1b3f5d9a625e9c231995d7720f9 Sean O'Halpin sean.ohalpin@gmail.com