DSLs: one interface, multiple implementations

One interface, multiple implementations is one of these design concepts I like most. It’s basically polymorphism in its pure form!

Coding a little bit a while ago, I realized how DSLs could reinforce this idea. My example here is a simple implementation for queue consumers, as a simple Ruby internal DSL:

class GitRepositoryCloner < Consumer
  queue "RepositoriesToBeCloned"
  exclusive true

  handle do |message|
    # git clone repository!
    # I'm pretty sure github has something alike
  end
end

And to enqueue a message, we could do:

GitRepositoryCloner.publish("rails")

GitRepositoryCloner is a simple consumer of the queue named RepositoriesToBeCloned. One of the times I did something similar to this, I needed support for exclusive consumers. Then, my choices were ActiveMQ as the messaging middleware together with the Stomp protocol.

Using them as an example, let’s take a look on a possible implementation for the Consumer class, using the stomp gem:

module ActiveMQ
  class Consumer

    def self.queue(name)
      @queue_name = name
    end

    def self.exclusive(bool)
      @exclusive = bool
    end

    def self.handle(&blk)
      @callback = blk
    end

    def self.listen
      broker = Stomp::Client.new(Config[:broker])
      broker.subscribe(@queue_name, 
          :'activemq.exclusive' => @exclusive) do |message|
        @callback.call(message)
        broker.acknowledge(message)
      end
    end

    def self.publish(message)
      broker = Stomp::Client.new(Config[:broker])
      broker.publish(@queue_name, message, :persistent => true)
    end

  end
end

Consumer = ActiveMQ::Consumer

The last line is where we choose the ActiveMQ::Consumer as the default implementation.

The beautiful aspect of this little internal DSL, composed only of three methods (queue, exclusive and handle), is that it defines an interface. Here, I have seen a common misconception from many developers coming from Java, C# and similar languages which have the interface construct. An interface in Object Orientation is composed of all accessible methods of an object. In other words, the interface is the object’s face to the rest of the world. We are not necessarily talking about a Java or C# interface construct.

In this sense, these three methods (queue, exclusive and handle) are the interface of the Consumer internal DSL (or class object, as you wish).

Let’s say for some reason, we would like to switch our messaging infrastructure to something else, like Resque, which Github uses and is awesome. Resque’s documentation says that things are a little bit different for Resque consumers. They must define a @queue class attribute and must have a perform class method.

As we would do with regular Java/C# interfaces, let’s make another implementation respecting the previous contract:

module Resque
  class Consumer

    def self.queue(name)
      @queue = name
    end

    def self.exclusive(bool)
      self.extend(Resque::Plugins::LockTimeout) if bool
    end

    def self.handle(&blk)
      self.send(:define_method, :perform, &@blk)
    end

    def self.listen
      raise "Not ready yet" unless self.respond_to?(:perform)
    end

    def self.publish(message)
      Resque.enqueue(self, message)
    end

  end
end

There you can see how the implementations differ. The exclusive consumer feature is provided by the LockTimeout plugin. In this case, instead of passing the activemq.exclusive parameter to the connection, we must use the Resque::Plugins::LockTimeout module, as the documentation says. Another key difference is in the message handling process. Instead of passing a handler block to the subscribe method, Resque consumers are required to define a perform method, which we are dynamically creating with some metaprogramming: Class#define_method(name).

Finally, here is how we switch our messaging backend to Resque, without any changes to the consumer classes (GitRepositoryCloner in this example):

Consumer = Resque::Consumer

That’s it: one interface, two (multiple) implementations.

3 thoughts on “DSLs: one interface, multiple implementations

  1. Polymorphism in OO is vital while playing with multiple strategies to achieve a specific goal. One can extend the example to support DI and allow composition of those strategies. Lets say, for example, I need to optionally add logging capability to the process, one can simply implement the expected interface and delegate to the proper base implementation. Nick Kallen has posted an example on DI and polymorphism usage in Ruby which I like, and I have extended it here too: .gist table { margin-bottom: 0; } class QueryEvaluator def transaction(&block) @connection_pool.with_connection do |connection| TransactionalQueryEvaluator.new(connection, @query_factory).instance_eval &block end end end class Factory def self.using(&config) Factory.new.instance_eval(&config) end def method_missing(decorator_name, *args) @built = "#{decorator_name.to_s.classify}QueryFactory".constantize.new(@built || Query, *args) end end query_factory = Factory.using { memoizing timing_out(2.seconds) stats_collecting(Stats.new) } [query_factory, ReversingQueryFactory.new(query_factory)].each do |qf| query_evaluator = QueryEvaluator.new(ConnectionPool.new(20), qf) query_evaluator.transaction do select("SELECT … FROM … FOR UPDATE …") execute("INSERT …") execute("INSERT …") end puts end # reverse implementation still suffers because its spread in the Proxy and in the # ReversingQueryFactory. everyone who can be reversed needs to extend QueryProxy view raw diff.rb hosted with ❤ by GitHub 2010 Modularity Olympics This is a contest, open to programming languages from all nations, to write modular and extensible code to solve the following problem: Implement a service that can run queries on a database. The Challenge Sounds simple right? Wrong! A programmer without control over the source-code of that service must be able to later add enhancements such as statistics collecting, timeouts, memoization, and so forth. There are a few more requirements: the “enhancements” must be specified in a configuration object which is consumed at run-time (e.g., it could be based on user-input). The enhancements are ordered (stats collecting wraps timeouts, not the other way around) but it must be possible to reverse the order of the enhancements at run-time. The enhancements must be “surgical” and not “global”. That is, it must be possible to simultaneously have two query services, one reversed and one not reversed, and even have a query service without any enhancements. Your code must be thread-safe or at least support concurrency using some technique. Please bear in mind the use of connection pools and such. Most programming contests emphasize things that are not modularity. This contest emphasizes things that are modularity. And its the olympics and don’t you want to be olympic? So go ahead and prove to the world that Design Patterns are no longer necessary, or that Haskell is the raddest thing since the movie Rad, or that Peanut Butter is better than Jelly. This is much more fun than inflamatory blog posts and angry reddit comments, yes? Yes. How to Play Fork this gist!! See my sample http://magicscalingsprinkles.wordpress.com/2010/02/16/2010-modularity-olympics/ Output Your program must produce exactly this output. Note the reversing order of the enhancements and the memoization of Query instantiation:
    Forward:
    Instantiating Query Object
    Selecting SELECT ... FROM ... FOR UPDATE ... on #<Object:0x11f6750>
    Did not timeout! Yay fast database!
    Measured select at 1.00 seconds
    Instantiating Query Object
    Executing INSERT ... on #<Object:0x11f6750>
    Did not timeout! Yay fast database!
    Measured select at 1.00 seconds
    Executing INSERT ... on #<Object:0x11f6750>
    Did not timeout! Yay fast database!
    Measured select at 1.00 seconds
    
    Backward:
    Instantiating Query Object
    Selecting SELECT ... FROM ... FOR UPDATE ... on #<Object:0x11f4ea0>
    Measured select at 1.00 seconds
    Did not timeout! Yay fast database!
    Instantiating Query Object
    Executing INSERT ... on #<Object:0x11f4ea0>
    Measured select at 1.00 seconds
    Did not timeout! Yay fast database!
    Executing INSERT ... on #<Object:0x11f4ea0>
    Measured select at 1.00 seconds
    Did not timeout! Yay fast database!
    
    view raw MODULARITY_OLYMPICS.markdown hosted with ❤ by GitHub require 'rubygems' require 'activesupport' require 'timeout' class MemoizingQueryFactory def initialize(query_factory) @memo = Hash.new do |h, k| puts "Instantiating Query Object" h[k] = query_factory.new(*k) # let's pretend this is thread-safe, kthx ruby! end end def new(connection, query_string, *args) @memo[[connection, query_string, args]] end end class QueryEvaluator def initialize(connection_pool, query_factory = Query) @query_factory = query_factory @connection_pool = connection_pool end def select(query_string, *args) @connection_pool.with_connection do |connection| @query_factory.new(connection, query_string, *args).select end end def execute(query_string, *args) @connection_pool.with_connection do |connection| @query_factory.new(connection, query_string, *args).execute end end def transaction(&block) @connection_pool.with_connection do |connection| TransactionalQueryEvaluator.new(connection, @query_factory).instance_eval &block end end end class TransactionalQueryEvaluator def initialize(connection, query_factory) @connection = connection @query_factory = query_factory end def select(query_string, *args) @query_factory.new(@connection, query_string, *args).select end def execute(query_string, *args) @query_factory.new(@connection, query_string, *args).execute end def transaction yield self end end class QueryProxy attr_accessor :query def initialize(query) @query = query end def select delegate("select") { @query.select } end def execute delegate("execute") { @query.execute } end def reverse case @query when QueryProxy reverse = @query.reverse inner = reverse.query clone = dup clone.query = inner reverse.query = clone reverse else self end end def dup clone = super clone.query = @query.dup clone end end class ReversingQueryFactory def initialize(query_factory) @query_factory = query_factory end def new(connection, query_factory, *args) @query_factory.new(connection, query_factory, *args).reverse end end class TimingOutQueryFactory def initialize(query_factory, timeout) @query_factory = query_factory @timeout = timeout end def new(connection, query_string, *args) TimingOutQuery.new(@query_factory.new(connection, query_string, *args), @timeout) end end class TimingOutQuery < QueryProxy def initialize(query, timeout) super(query) @timeout = timeout end def delegate(method) result = Timeout.timeout(@timeout.to_i) { yield } puts "Did not timeout! Yay fast database!" result end end class StatsCollectingQueryFactory def initialize(query_factory, stats) @query_factory = query_factory @stats = stats end def new(connection, query_string, *args) StatsCollectingQuery.new(@query_factory.new(connection, query_string, *args), @stats) end end class StatsCollectingQuery < QueryProxy def initialize(query, stats) super(query) @stats = stats end def delegate(method) @stats.measure(method) { yield } end end class Query def initialize(connection, query_string, *args) @connection = connection @query_string = query_string @args = args end def select sleep 1 puts "Selecting #{@query_string} on #{@connection}" [1, 2, 3] end def execute sleep 1 puts "Executing #{@query_string} on #{@connection}" 1 end end class ConnectionPool def initialize(size) @size = size end def with_connection yield Object.new end end class Stats def measure(name) result = nil bm = Benchmark.measure { result = yield } puts "Measured #{name} at #{"%.2f" % bm.real} seconds" result end end class Factory def self.using(&config) Factory.new.instance_eval(&config) end def method_missing(decorator_name, *args) @built = "#{decorator_name.to_s.classify}QueryFactory".constantize.new(@built || Query, *args) end end query_factory = Factory.using { memoizing timing_out(2.seconds) stats_collecting(Stats.new) } [query_factory, ReversingQueryFactory.new(query_factory)].each do |qf| query_evaluator = QueryEvaluator.new(ConnectionPool.new(20), qf) query_evaluator.transaction do select("SELECT … FROM … FOR UPDATE …") execute("INSERT …") execute("INSERT …") end puts end # reverse implementation suffers because its spread in the Proxy and in the # ReversingQueryFactory. everyone who can be reversed needs to extend QueryProxy view raw sample.rb hosted with ❤ by GitHub One can also extend by providing instance based methods instead of self. methods, allowing different callbacks in different parts of the system: but that is a question of single/multiple configs required in a running enviroment (in a easy way) Great example on something vital to coding in OO.

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s