@@ -1,5 +1,6 @@ # Todos for 0.5.0 +* refactor starling* to be memcache*. add aliased classes into deprecated.rb. * gemify * move all runner/invoker implementations out of workling * move backend discovery code out of workling @@ -8,7 +9,6 @@ # Todos for 1.0 * add phusion daemon starter option so that workling_client doesn't need to be started manually. -* refactor starling* to be memcache*. add aliased classes into deprecated.rb. * merb support * test on jruby * more runners: sqs, beanstalkd TODO.markdown @@ -8,7 +8,7 @@ module Workling module Clients class AmqpClient < Workling::Clients::Base - # connects to the queue server + # starts the client. def connect; @amq = MQ.new; end # disconnect from the queue server lib/workling/clients/amqp_client.rb @@ -41,7 +41,8 @@ module Workling # closes the memcache connection def close - self.connection.close + self.connection.flush_all + self.connection.reset end # implements the client job request and retrieval lib/workling/clients/memcache_queue_client.rb @@ -16,7 +16,7 @@ module Workling # set the desired invoker. this class grabs work from the job broker and executes it. mattr_accessor :invoker - @@invoker = Workling::Remote::Invokers::ThreadedPoller + @@invoker ||= Workling::Remote::Invokers::ThreadedPoller # retrieve the dispatcher or instantiate it using the defaults def self.dispatcher lib/workling/remote.rb @@ -71,6 +71,7 @@ module Workling # Setup connection to client (one per thread) connection = @client_class.new + connection.connect logger.info("** Starting client #{ connection.class } for #{clazz.name} queue") # Start dispatching those messages lib/workling/remote/invokers/threaded_poller.rb @@ -11,12 +11,16 @@ module Workling def initialize AmpqRunner.routing = Workling::Routing::ClassAndMethodRouting.new - AmpqRunner.client = Workling::Clients::AmqpClient + AmpqRunner.client = Workling::Clients::AmqpClient.new end # enqueues the job onto RabbitMQ or similar def run(clazz, method, options = {}) - AmqpClient.client.set(@@routing.queue_for(clazz, method), options) + + # neet to connect in here as opposed to the constructor, since the EM loop is + # not available there. + @connected ||= AmpqRunner.client.connect + AmpqRunner.client.request(@@routing.queue_for(clazz, method), options) return nil end lib/workling/remote/runners/ampq_runner.rb @@ -26,7 +26,7 @@ module Workling # enqueues the job onto Starling. def run(clazz, method, options = {}) - StarlingRunner.client.set(@@routing.queue_for(clazz, method), options) + StarlingRunner.client.request(@@routing.queue_for(clazz, method), options) return nil # empty. lib/workling/remote/runners/starling_runner.rb @@ -2,7 +2,8 @@ puts '=> Loading Rails...' require File.dirname(__FILE__) + '/../../../../config/environment' require File.dirname(__FILE__) + '/../lib/workling/remote' -require File.dirname(__FILE__) + '/../lib/workling/remote/invokers/poller' +require File.dirname(__FILE__) + '/../lib/workling/remote/invokers/basic_poller' +require File.dirname(__FILE__) + '/../lib/workling/remote/invokers/threaded_poller' require File.dirname(__FILE__) + '/../lib/workling/remote/invokers/eventmachine_subscriber' require File.dirname(__FILE__) + '/../lib/workling/routing/class_and_method_routing' script/listen.rb 78f744bae4f83effa6c8f9f166febd04b1cba7ac Rany Keddo rany@playtype.net http://github.com/purzelrakete/workling/commit/b46e88a66f65cdc13c3d3d6bd72c799ef5ba065d b46e88a66f65cdc13c3d3d6bd72c799ef5ba065d 2008-11-06T13:50:57-08:00 2008-11-06T13:50:57-08:00 fixes. 46b2d5bfe938d2b01cf33742cbb5dfc14d35606f Rany Keddo rany@playtype.net