Shahzad Bhatti Welcome to my ramblings and rants!

Introduction

I have been a Twitter user for a while, have observed or heard about downtime and scalability problems with Twitter. The scalability of Twitter has become a topic for a lot of discussions and blogs and has also offered a useful excercise to design scalable systems. A common root cause as identified from Twitter’s blogs is that the architecture is based on CMS because it was written in Ruby on Rails and that is what Rails good at. The solution to the scalability problem as pointed by other people is messaging based architecture. There’s also been a lot of blame for Twitter’s problems on Ruby and Rails because Ruby is a slow language compared to other static and dynamic languages and Rails is not built for scalability. Though, there is some truth to it, but I don’t think there are the sole bottlenecks. In fact, I am going to show a small prototype written in Ruby and Rails (partially) that integrate with the messaging middlewares, which can be scaled easily. I have been using messaging middlwares such as CORBA Event service, IBM MQ series, Websphere, Weblogic, Tibco and ActiveMQ for over ten years and have long been proponent of messaging based sysems for scalable systems [1] [2]. So, I spent a few hours to put together a prototype based on messaging middleware and Ruby on Rails to see how such system can be developed.

Design Principles

Before describing my design, I am going to review some of the design principles that I have used for building large scale systems ([1], [2]), which are:

• Coupling/Cohesion – loosely coupled and shared nothing architecture and partitioning based on functionality.
• Messaging or SEDA architecture to implement reliable and scalable services and avoid temporal dependencies.
• Resource management – good old practices of avoiding leaks of memory, database connections, etc.
• Data replication especially read-only data.
• Partition data (Sharding) – using multiple databases to partition the data.
• Avoid single point of failure.
• Bring processing closer to the data.
• Design for service failures and crashes.
• Dynamic Resources – Design service frameworks so that resources can be removed or added automatically and clients can automatically discover them. Use virtualization and horizontal scalability whenever possible.
• Smart Caching – Cache expensive operations and contents as much as possible.
• Avoid distributed transactions, use optimistic compensating transactions (See CAP principle).

Architecture & Design

Following is high level architecture for the Microblogging system:

First, I selected REST architecture as an entry point to our system for both Web UI and 3rd party applications and used messaging middleware for implementing the sevices. This gives us ease of access with REST APIs and scalability with messaging. In my implementation I chose JRuby/Rails to implement most of the code, Derby for the database and ActiveMQ for the messaging store. In addition to scalability, the messaging middleware gives you a lot of advantages from functional languages like Erlang such as immutability, message passing, fault tolerance (via persistence queues). You can even build support for versioning and hot code swapping by adding version number to each message and creating routers (See integration patterns) to direct messages to different handlers.

APIs

Following are REST APIs that will be exposed to 3rd party apps, Web and other kind of UI:

Create User

POST /users
where the user information is passed in the body in the form of parameters.

Login/Authenticate User

POST /users/userid/sessions
This will authenticate the user and create a session. Note that most of following APIs send back session-id, will be stored in the database (sharded) and will be used to retrieve user information.

Logout

HTTP-HEADER
session-id

DELETE /users/userid/sessions

Get User information

HTTP-HEADER
session-id

GET /users/userid
This API will return detailed user information

Anonymous User information

GET /users/userid
This API will return public user information

List of Followings

HTTP-HEADER
session-id

GET /followings/userid
This API will return summary of people, the user is following.

Create Followers

HTTP-HEADER
session-id

POST /followers/followerid
This API will create one-way follower relationship between the user and follower.

Enable notification for Followers

HTTP-HEADER
session-id

POST /followers/followerid/notifications

Disable notification for Followers

HTTP-HEADER
session-id

DELETE /followers/followerid/notifications

Block Followers

HTTP-HEADER
session-id

POST /followers/followerid/blocking

Unblock Followers

HTTP-HEADER
session-id

DELETE /followers/followerid/blocking

Follower Exist

HTTP-HEADER
session-id

GET /followers/followerid
This API will return 200 HTTP code if follower exist.

List of Followers

HTTP-HEADER
session-id

GET /followers
This API will return summary of people, the user is following.

Archive Messages

HTTP-HEADER
session-id

GET /messages?offset=xxx&limit=yyy&since=date
This API will return archived messages for the user, where offset and limit will be optional.

DELETE Messages

HTTP-HEADER
session-id

DELETE /messages/message-id
This API will return archived messages for the user, where offset and limit will be optional.

Send Direct Messages

HTTP-HEADER
session-id

POST /directmessages/targetuserid
This API will return send direct message to the given user.

Send Reply

HTTP-HEADER
session-id

POST /reply/message-id
This API will return reply for the given message-id and pass the contents of the message in the body (as parameters).

Direct Messages Received

HTTP-HEADER
session-id

GET /directmessages/userid?offset=xxx&limit=yyy&since=date
This API will return messages received by the user.

Replies Received

HTTP-HEADER
session-id

GET /replies?offset=xxx&limit=yyy&since=date
This API will return replies received by the user.

Update Status

HTTP-HEADER
session-id

POST /statuses
This API will update status of the user and pass the contents of the message in the body (as parameters).

Get Statuses

HTTP-HEADER
session-id

GET /statuses?offset=xxx&limit=yyy&since=date
This API will update status of the user and pass the contents of the message in the body (as parameters).

User Timeline

HTTP-HEADER
session-id

GET /timeline/username?offset=xxx&limit=yyy&since=date
This API will return timeline of the user. This API will compare given username with the authenticated username and will return detailed timeline if match, otherwise it will return public timeline.

Public Timeline

GET /timeline/username?offset=xxx&limit=yyy&since=date
This API will return public timeline of the user.

Friends Timeline

HTTP-HEADER
session-id

GET /friendstimeline?offset=xxx&limit=yyy&since=date
This API will return timeline of the friends of the user.

Request Flow

Here is an illustration of how information is flowed through different components:

Though, I am not showing request flow of all APIs, but they will follow similar pattern of flow.

Detailed Design

Domain Classes

Primary domain classes are:

• User
• Message, which has four subclasses DirectMessage, ReplyMessage, Tweet and Status for various kind of messages in the system.
• Follower – creates one-way relationship between two users, where follower can choose to be notified when the user changes his/her status.

I identify each message with special GUID and using a simple scheme to generate GUID for request-ids and message-ids, but for real project I would recommend better libraries such as UUIDTools.

Schema

Followers

  
  class CreateFollowers < ActiveRecord::Migration
    def self.up
      
      create_table :followers do |t|
        t.column :username,            :string, :limit => 16
        t.column :follower_username,   :string, :limit => 16
        t.column :relation_type,       :string, :default => 'Follower', :limit => 32
        t.column :blocked,             :boolean, :default => false
        t.column :notifications,       :boolean, :default => false
        t.column :created_at,          :datetime
        t.column :updated_at,          :datetime
        t.column :deleted_at,          :datetime
      end
      add_index :followers, :username
      add_index :followers, :follower_username
    end
  
    def self.down
      drop_table :followers
    end
  end
 
 

---

Messages

  class CreateMessages < ActiveRecord::Migration
    def self.up
      create_table :messages do |t|
        t.column :message_id,          :string, :limit => 64
        t.column :type,                :string, :limit => 32
        t.column :message_type,        :string, :default => 'Say', :limit => 32
        t.column :reply_message_id,    :string, :limit => 64
        t.column :username,            :string, :limit => 16
        t.column :channel_name,        :string, :limit => 32
        t.column :message_body,        :string, :limit => 140
        t.column :favorite,            :boolean, :default => false
        t.column :sent_at,             :datetime, :default => Time.now.utc
        t.column :created_at,          :datetime
        t.column :deleted_at,          :datetime
      end
      add_index :messages, :message_id
      add_index :messages, :username
    end
  
    def self.down
      drop_table :messages
    end
  end
 
 

---

Users

  class CreateUsers < ActiveRecord::Migration
    def self.up
      create_table :users do |t|
        t.column :username,            :string, :limit => 16
        t.column :password,            :string, :limit => 16
        t.column :name,                :string, :limit => 64
        t.column :email,               :string, :limit => 64
        t.column :time_zone_id,        :string, :limit => 32
        t.column :created_at,          :datetime
        t.column :updated_at,          :datetime
        t.column :deleted_at,          :datetime
      end
      add_index :users, :username
    end
  
    def self.down
      drop_table :users
    end
  end
 
 

---

Persistence

I used Rails’ ActiveRecord library to provide persistence, though alternatively I could have used ActiveHibernate. These libraries provide a quick way to add persistence capabilities with minimum configuration and boilerplate. This prototype is using multiple levels of partitioning, first at the service level, second at the persistence level. I am using multiple databases of Derby to store objects using a simple hashing scheme for load distribution. This prototype also shows how to connect to multiple databases in Rails, which was difficult in early implementation of Twitter.

Domain Services

The core model and services use domain driven design and applies principles of fat model and thin service (as opposed to fat servicess and anemic model). The domain services implement external REST APIs and use underlying ActiveRecord for most of the functionality.

Messaging Middleware

The REST based web services don’t invoke domain services directly, instead they use messaging middleware. In real application, I might use ESB/integration patterns such as intelligent routing to partition the system across multiple machines and send the request to the suitable queue. In this prototype, I am simply using ActiveMQ, which is fairly robust and easy to use. I am also using separate queues for different kind of operations. Another lesson I have learned in building large systems is to separate reads from the writes so that you can scale them independently and also offer different quality of services, e.g. read queues can be non-persistent, but write queues can be persistent.

Business Delegate

The REST based web services don’t interact with the messaging middleware directly, instead they use business delegates that hides all details of sending out message, creating temporary queues and receiving messages. The interface of business delegates is same as services.

Benchmark Results

Though, performance was not the objective of my prototype, but I tried to check how many requests I can process on my development machine. I chose only to benchmark messaging middlewares and not REST server because JVM uses native threads and web containers such as Tomcat uses a small sized thread pool to perform requests. Since, our architecture is heavily IO-bound, that would not scale. Alternatively, I could have build reactive or event based APIs for HTTP or use Yaws/Mochiweb as a container for REST based web sevices because creating a process in Erlang is pretty cheap. For example, Erlang process takes 300 bytes, whereas Java thread take 2M by default (though, it can be reduced to 256K on most machines). Here are results of running a simple server with embedded ActiveMQ and load test both running JRuby on my Pentium Linux machine. I used default VM size for both JRuby processes and didn’t tune any options:

What	Elapsed Time (secs)	Throughput	Invocation Times
load_test_create_users	99.041000	10.09682857311318/s	1000
load_test_bad_authenticate	79.407000	12.593348183199506/s	1000
load_test_good_authenticate	91.802000	10.893008861477503/s	1000
load_test_get_users	91.362000	10.945469671474584/s	1000
load_test_create_followers	143.560000	6.958714408811979/s	999
load_test_follower_exists	95.140000	10.50020495537709/s	999
load_test_get_followers	99.881000	10.002002391423325/s	999
load_test_get_followings	104.737000	9.538176576655392/s	999
load_test_block_follower	156.106000	6.399497779340769/s	999
load_test_unblock_follower	161.950000	6.168532449211547/s	999
load_test_follower_enable_notifications	157.637000	6.337344665142635/s	999
load_test_follower_disable_notifications	161.714000	6.1775727524053545/s	999
load_test_send_direct_message	1673.494000	5.969546350480188/s	9990
load_test_messages_sent_receive	157.832000	6.323179075798668/s	998
load_test_send_direct_message	1725.717000	5.788898179687536/s	9990
load_test_send_replies	21.546000	5.105592926919201/s	110
load_test_get_user_status	86.952000	11.500598043963544/s	1000
load_test_update_status	213.298000	4.688276498896948/s	1000
load_test_get_user_status	88.987000	11.23759650430517/s	1000
load_test_archive_messages	80.982000	12.348575563593377/s	1000
load_test_public_timeline	98.143000	10.189213685309506/s	1000
load_test_user_timeline	92.871000	10.767623902461311/s	1000
load_test_friends_timeline	140.155000	7.1349577268319555/s	1000
load_test_user_and_friends_timeline	193.257000	5.174430032905596/s	1000

I was not impressed with the results I got and I may implement similar prototype in Erlang using MochiWeb, Mnesia and Ejabber.

Summary

Though, the technologies I choose are somewhat arbitrary, the same design principles can be applied to other technology stack. Though, technologies are generaly selected due to political reasons or personal preference, it is important to consider team’s familiarity with the technology stack. For example, Twitter tried Ejabber but had problems troubleshooting becaues they were not familiar with Erlang.
Again, I used architecture principles such as stateless services, though in real life I may have to store some state which can be stored in the shareded database. I used embedded services such as embedded database server and messaging server because you can then easily start a single process on a machine and replicate machines as the load increases. I used partitioning/sharding heavily, which is key for scalability. I also used replication specially replicating messages for each follower so that the reads are fast and we don’t spend a lot of time querying the database. Though, it does add cost on the write side and adds disk requirements. I think we can control those by limiting number of messages per user per minute and maximum number of notifications. Also, we can remove old messages from the database. I also used principle of bringing computating closer to data by using embedded database.

Other Improvements

I tried to show that scalability problems are best solved from the architecture that is independent of a particular technology or language. In fact, I may implement similar design in Erlang using Mnesia as the database, Ejabber as the messaging middleware, Mochiweb for REST and OTP for implementing services. Other improvements that I would suggest be use of reverse proxy server for caching along with DMZ for security. We can also add some throttling to the reverse proxy servers. In order to have better fault tolerance, I might have multiple reverse proxy servers and use DNS round robin, though I don’t like DNS based load balance for real load distribution because it does not take server’s capacity and load into account, but it would suitable in this case. Also, I didn’t implement any caching, but we can use caching solutions such as EHCache, Tangosol, Terracotta, Memcache, etc. Though, caching highly dynamic contents may be difficult and less reusable. Also, I didn’t use any ESB (lightweight providers such as Mule, ServiceMix), but it can be used to abstract routing to the services, load balance, aggregation, replication, transformation, etc. Also, we can build support for versioning and hot code swapping by adding version number to each message and changing routing schemes. This prototype uses a simple scheme for partitioning by creating hash of business keys such as username, however it is difficult to manage when you need to add or remove servers because it requires migrating data. Another solution is to use MD5 scheme that we use at Amazon for S3, which may calculate MD5 of username and each replicated queue and select the queue whose md5-sum is higher than the username’s sum. For further fault tolerance, we could replicate data in Master/Slave fashion.

PS: I noticed “NativeException: javax.jms.JMSException: PermGen space” a couple of times running long load test, which is generally caused by reloading classes (as I have observed this for many years in Weblogic and Tomcat). I probably need to investigate this further but my guess is that JRuby is doing something dumb.

Download

Source Source Code

Appendix

Source Code Model

./model/user.rb

  ###
  # User of the system.
  # Though, in our system user can have many messages that it sent to update 
  # his/her status or other users and can also have followers and followings, but
  # we are not showing those relationships because we cannot populate them due to
  # sharding and scalability concerns.
  
 
  ###
  class User < ActiveRecord::Base
      validates_presence_of       :name
      validates_presence_of       :username,         :within => 3..16
      validates_presence_of       :password,      :within => 3..16
      validates_length_of         :email,         :within => 3..64
      validates_uniqueness_of     :email,         :case_sensitive => false
      validates_format_of         :email,         :with => /^([^@\s]+)@((?:[-a-z0-9]+\.)+[a-z]{2,})$/i
      #
      ### all messages for the same user wil be stored in the same database.
      ### also, all messages from the users he/she is following will be copied to the user's database
      ### if they lie in different databases.
      ### limiting rows returned from these relationships to 100 so that we don't overwhelm server.
      #
      has_many :statuses, :class_name => 'Message', :foreign_key => :username, :limit => 100, :order => 'sent_at desc'
      has_many :tweets, :class_name => 'Message', :foreign_key => :channel_name, :limit => 100, :order => 'sent_at desc'
  
      #
      ### We will replicate follower/following to each database and though we may not have corresponding users,
      ### but we can have usernames
      #
      def self.follower_usernames(username, offset, limit)
         Follower.followers(username, offset, limit).map {|f| f.follower_username }
      end
  
      def self.following_usernames(username, offset, limit)
         Follower.followings(username, offset, limit).map {|f| f.username }
      end
  
      def self.follower_count(username)
         Follower.followers_count(username)
      end
      def self.following_count(username)
         Follower.followers_count(username)
      end
  
      #
      ### user's timeline
      #
      def self.timeline(username, offset, limit)
         Message.find(:all, :conditions => ['username = ?', username], :offset => offset, :limit => limit)
      end
  
      def to_s
        "#{username}"
      end
  
      def to_external_xml(xml)
        xml.user(:username => self.username, :name => self.name, :email => self.email)
      end
  end
 
 

---

./model/message.rb

  #
  ###
  # Message stores any message sent by user to update his/her status, tweets
  # from other users that he/she is following, direct message sent to another
  # user or reply message sent in response to tweet or direct message.
  # Note that we use a GUID based message_id in conjunction with the database column 'id'
  # because we are using sharding the database column can be duplicate, but the GUID
  # based message_id will always be unique across all databases.
  ###
  class Message < ActiveRecord::Base
      MESSAGE_TYPE_SAY = "Say"
      MESSAGE_TYPE_SHOUT = "Shout"
  #
      validates_presence_of       :message_id,    :within => 1..64
      validates_presence_of       :username
      #
      ### channel_name will be username of the target user for now,
      ### though, in future it could be a group or a topic
      #
      validates_presence_of       :channel_name,  :within => 1..32
      validates_presence_of       :message_body,  :within => 1..140
      validates_presence_of       :message_type,  :within => 1..32
      validates_presence_of       :sent_at
      validates_uniqueness_of     :message_id
      #
      ### all messages for the same user wil be stored in the same database.
      ### also, all messages from the users he/she is following will be copied to the user's database
      ### if they lie in different databases.
      #
      belongs_to :user, :class_name => 'User', :foreign_key => :username
  
      def self.messages_for(username, offset, limit)
         Message.find(:all, :conditions => ['username = ?', username], :offset => offset, :limit => limit, :order => 'sent_at desc')
      end
  
      def self.destroy_message(id)
         message = Message.find_by_message_id(id)
         if message
            message.destroy
            message
         else
            nil
         end
      end
  
  
      def to_s
        "#{username} -> #{channel_name}: #{message_body}"
      end
  
      def to_external_xml(xml)
        xml.message(message_body, :message_id => self.message_id, :message_type => self.message_type, :from => self.username, :to => self.channel_name, :sent_at => sent_at.httpdate) 
      end
  end
 
 

---

./model/reply_message.rb

  ###
  # Message sent in response to another tweet, message or direct message.
  #
  # commented below because message could be in different database.
  # belongs_to :reply_message_id, :class_name => Message, :dependent => :destroy
  ###
  
  require 'direct_message'
  
  class ReplyMessage < DirectMessage
    validates_presence_of       :reply_message_id,    :within => 1..64
    def self.replies(username, offset, limit)
       ReplyMessage.find(:all, :conditions => ['username = ?', username], :offset => offset, :limit => limit, :order => 'sent_at desc')
    end
  end
 
 

---

./model/guid_generator.rb

  class GuidGenerator
    @@count = 1
    def self.guid(prefix)
      @@count += 1
      "#{prefix}#{@@count}#{Time.now.to_i}"
    end
    def self.next_message_id
      guid("message_id")
    end
    def self.next_request_id
      guid("request_id")
    end
  end
 
 

---

./model/follower.rb

  #
  ####
  # Follower stores one-way relation between two users. However, in order to 
  # support sharding, we don't directly connect users instead we only store
  # users' usernames.
  ###
  class Follower < ActiveRecord::Base
      RELATION_TYPE_FOLLOWER = "Follower"
      RELATION_TYPE_FRIEND = "Friend"
      RELATION_TYPE_FAMILY = "Family"
      RELATION_TYPE_COLLEAGUE = "Colleague"
      RELATION_TYPE_ACQUAINTENCE = "Acquaintance"
      RELATION_TYPE_FAN = "Fan"
      RELATION_TYPE_OTHER = "Other"
  
      validates_presence_of       :relation_type
      #validates_presence_of       :blocked
      #validates_presence_of       :notifications
  
      def self.create_follower(follower_attrs)
          Follower.create!(follower_attrs) unless exists?(follower_attrs[:username], follower_attrs[:follower_username])
      end
  
      def self.followers(username, offset, limit)
          self.find(:all, :conditions => ['username = ?', username], :offset => offset, :limit => limit)
      end
      def self.follower_count(username)
          self.count(:conditions => ['username = ?', username])
      end
      def self.following_count(username)
          self.count(:conditions => ['follower_username = ?', username])
      end
  
      def self.followings(username, offset, limit)
          self.find(:all, :conditions => ['follower_username = ?', username], :offset => offset, :limit => limit)
      end
      def self.destroy_follower(username, follower_username)
          self.get_record(username, follower_username).destroy
      end
      def self.exists?(username, follower_username)
          !self.get_record(username, follower_username).nil?
      end
      def self.enable_notifications(username, follower_username)
          self.get_record(username, follower_username).update_attributes(:notifications => true)
      end
      def self.disable_notifications(username, follower_username)
          self.get_record(username, follower_username).update_attributes(:notifications => false)
      end
      def self.block(username, follower_username)
          self.get_record(username, follower_username).update_attributes(:blocked => true)
      end
      def self.unblock(username, follower_username)
          self.get_record(username, follower_username).update_attributes(:blocked => false)
      end
      def self.get_record(username, follower_username)
          self.find(:first, :conditions => ['username = ? and follower_username = ?', username, follower_username])
      end
      def to_s
        "#{username} -> #{follower_username}"
      end
      def to_external_xml(xml)
        xml.follower(:username => self.username, :follower_username => self.follower_username)
      end
  
  end
 
 

---

./model/direct_message.rb

  ####
  # Direct message is sent directly to another user
  ####
  class DirectMessage < Message
    def to_user(user)
       self.channel_name = user.username
    end
    def self.direct_messages_sent(username, offset, limit)
       DirectMessage.find(:all, :conditions => ['username = ?', username], :offset => offset, :limit => limit, :order => 'sent_at desc')
    end
    def self.direct_messages_received(username, offset, limit)
       DirectMessage.find(:all, :conditions => ['channel_name = ?', username], :offset => offset, :limit => limit, :order => 'sent_at desc')
    end
  end
 
 

---

Source Code Services

./service/base_service.rb

  class BaseService
  protected
    def initialize(username)
      SchemaHelper.setup_schema_for(username)
      
    end
  
    def serialize(obj, format)
      if obj
        format ||= 'xml'
        case format
          when 'xml'
            obj.to_xml
          when 'yaml'
            obj.to_yaml
          when 'json'
            obj.to_json
          else
            obj
        end
      end
    end
  end
 
 

---

./service/users_service.rb

  #
  ## UserService provides basic functionality for quering/storing users
  ## It assumes connection to the right database is already setup.
  #
  class UsersService < BaseService
    #
    ### initialize service
    #
    def initialize(username)
      super
    end
  
  
    #
    ### retrieves user record
    #
    def get_user(username, options={})
      User.find_by_username(username)
    end
  
    #
    ### return usernames of people whom the user is following.
    #
    def followings(username, options={})
      offset = options[:offset] || 0
      limit = options[:limit] || 100
      User.following_usernames(username, offset, limit)
    end
  
  
    #
    ### return usernames of followers of authenticated user
    #
    def followers(username, options={})
      offset = options[:offset] || 0
      limit = options[:limit] || 100
      User.follower_usernames(username, offset, limit)
    end
  
  
    def authenticate(username, password, options={})
      User.find_by_username_and_password(username, password)
    end
  
  
    def create_user(user_attrs, options={})
      user = User.new()
      user.attributes = user_attrs
      user.save!
      user
    end
  
    def archive_messages(username, options={})
      offset = options[:offset] || 0
      limit = options[:limit] || 100
      Message.messages_for(username, offset, limit)
    end
  end
 
 

---

./service/followers_service.rb

  #
  ## FollowersService provides basic functionality for creating/quering followers
  ## It assumes all users are already authenticated before calling these methods.
  #
  class FollowersService < BaseService
    #
    ### initialize service
    #
    def initialize(username)
      super
    end
    #
    ### create follower
    #
    def create_follower(follower_attrs, options={})
      Follower.create_follower(follower_attrs)
    end
  
    #
    ### enable notifications
    #
    def enable_notifications(username, follower_username, options={})
      Follower.enable_notifications(username, follower_username)
    end
  
    #
    ### disable notifications
    #
    def disable_notifications(username, follower_username, options={})
      Follower.disable_notifications(username, follower_username)
    end
  
  
    #
    ### blocks a user
    #
    def block(username, follower_username, options={})
      Follower.block(username, follower_username)
    end
  
    #
    ### unblock user
    #
    def unblock(username, follower_username, options={})
      Follower.unblock(username, follower_username)
    end
  
  
    #
    ### delete status
    #
    def destroy_follower(username, follower_username, options={})
      Follower.destroy_follower(username, follower_username)
      true
    rescue ActiveRecord::RecordNotFound => e
      false
    end
  
  
    #
    ### get a single status
    #
    def exists?(username, follower_username, options={})
      Follower.exists?(username, follower_username)
    end
  end
 
 

---

./service/messages_service.rb

  #
  ## MessagesService provides basic functionality for quering/storing statuses/messages
  ## It assumes all users are already authenticated before calling these methods.
  #
  class MessagesService < BaseService
    #
    ### initialize service
    #
    def initialize(username)
      super
    end
  
  
    #
    ### get a single status
    #
    def get_status(message_id, options={})
      Message.find_by_message_id(message_id)
    end
  
    #
    ### update status
    #
    def update_status(message_attrs, options={})
      message = Tweet.new
      message.attributes = message_attrs
      safe_save(message)
    end
  
    #
    ### delete status
    #
    def destroy_message(message_id, options={})
      Message.destroy_message(message_id)
    end
  
  
  
    #
    ### retrieves most recent statuses for public
    #
    def public_timeline(username, options={})
      offset = options[:offset] || 0
      limit = options[:limit] || 10
      User.timeline(username, offset, limit)
    end
  
    #
    ### retrieves most recent statuses for user
    #
    def user_timeline(username, options={})
      offset = options[:offset] || 0
      limit = options[:limit] || 100
      User.timeline(username, offset, limit)
    end
  
  
    #
    ### retrieves most recent statuses for friends
    #
    def friends_timeline(username, options={})
      offset = options[:offset]
      limit = options[:limit]
      get_friends_timeline(username, offset, limit)
    end
  
  
    #
    ### retrieves most recent statuses for user and his/her friends
    #
    def user_and_friends_timeline(username, options={})
      offset = options[:offset]
      limit = options[:limit]
      messages = User.timeline(username, offset, limit)
      messages += get_friends_timeline(username, offset, limit)
    end
  
  private
  
    #
    ### retrieves most recent statuses for friends without any conversion
    #
    def get_friends_timeline(username, offset, limit)
      offset = offset || 0
      following_count = Follower.following_count(username)
      limit = limit || (following_count < 100 ? following_count / 100 + 1 : 1)
  
      following = User.following_usernames(username, 0, 200)
      messages = []
      following.each do |username|
        messages += Message.messages_for(username, offset, limit)
      end
      messages
    end
  protected
    def safe_save(message)
      old_message = Message.find_by_message_id(message.message_id)
      if old_message.nil?
         message.save!
         message
      else
         puts "Message with id #{message.message_id} already exists #{message.inspect}" 
         nil
      end
    end
  end
 
 

---

./service/direct_messages_service.rb

  #
  ## DirectMessagesService provides functionality for sending/receiving direct messages
  #
  class DirectMessagesService < MessagesService
    #
    ### initialize service
    #
    def initialize(username)
      super
    end
    #
    ### sents back 20 most recent direct messages for the user
    ### it assumes user is already authenticated.
    #
    def direct_messages_received(username, options={})
      offset = options[:offset] || 0
      limit = options[:limit] || 20
      DirectMessage.direct_messages_received(username, offset, limit)
    end
    #
    ### sents back 20 most recent direct messages for the user
    ### it assumes user is already authenticated.
    #
    def direct_messages_sent(username, options={})
      offset = options[:offset] || 0
      limit = options[:limit] || 20
      DirectMessage.direct_messages_sent(username, offset, limit)
    end
  
    #
    ### send direct message
    #
    def send_direct_message(message_attrs, options={})
      message = DirectMessage.new()
      message.attributes = message_attrs
      safe_save(message)
    end
  
    #
    ### send reply message
    #
    def send_reply(message_attrs, options={})
      message = ReplyMessage.new()
      message.attributes = message_attrs
      safe_save(message)
    end
  
    #
    ### sents back 20 most recent replies for the user
    ### it assumes user is already authenticated.
    #
    def replies(username, options={})
      offset = options[:offset] || 0
      limit = options[:limit] || 100
      ReplyMessage.replies(username, offset, limit)
    end
  
  
    #
    ### delete direct message
    #
    def destroy_direct_message(id, options={})
      DirectMessage.destroy_message(id)
    end
  end
 
 

---

Source Code Business Delegate

./delegate/base_delegate.rb

  #
  ## BaseDelegate provides client side interface for connecting to the business 
  ## service using BusinessDelegate pattern.
  #
  class BaseDelegate
  protected
    def initialize(jms_helper, read_queue_name, write_queue_name)
       @read_queue_name = read_queue_name
       @write_queue_name = write_queue_name
       @jms_helper = jms_helper 
       @jms_helper.create_producers(@read_queue_name, @write_queue_name)
    end
  
  
    def new_jms_message(properties={})
      jms_msg = @jms_helper.create_message("")    ### ActiveMQTextMessage.new
      jms_msg.setStringProperty('request_id', GuidGenerator.next_request_id)
      properties.each do |name, value|
          jms_msg.setStringProperty(name.to_s, value.to_s)
      end
      jms_msg
    end
  
  
    def send_read_request(jms_msg)
      @jms_helper.send_message(@read_queue_name, jms_msg, true).text
    end
  
  
    def send_write_request(jms_msg, reply=true)
      response =  @jms_helper.send_message(@write_queue_name, jms_msg, reply)
      if response.respond_to? :text
          response.text
      else
          response
      end
    end
  end
 
 

./delegate/users_delegate.rb

  #
  ## UserDelegate provides basic functionality for quering/storing users
  ## It assumes connection to the right database is already setup.
  #
  class UsersDelegate < BaseDelegate
    def initialize(jms_helper)
      super(jms_helper, JmsHelper::QUEUE_READ_USERS, JmsHelper::QUEUE_WRITE_USERS)
    end
    def create_user(user_attrs, options={})
        args = options.merge(user_attrs).merge(:action => 'Users.create_user')
        jms_msg = new_jms_message(args)
        send_write_request(jms_msg)
        
    end
  
    #
    ### retrieves user record
    #
    def get_user(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'Users.get_user'))
        send_read_request(jms_msg)
        
    end
  
    #
    ### return usernames of people the user is following
    #
    def followings(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'Users.followings'))
        send_read_request(jms_msg)
        
    end
  
  
    #
    ### return 100 usernames of followers of authenticated user
    #
    def followers(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'Users.followers'))
        send_read_request(jms_msg)
        
    end
  
  
    def authenticate(username, password, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :password => password, :action => 'Users.authenticate'))
        send_read_request(jms_msg)
        
    end
  
  
    def archive_messages(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'Users.archive_messages'))
        send_read_request(jms_msg)
        
    end
  end
 
 

---

./delegate/messages_delegate.rb

  #
  ## MessagesDelegate provides basic functionality for quering/storing statuses/messages
  ## It assumes all users are already authenticated before calling these methods.
  #
  class MessagesDelegate < BaseDelegate
    def initialize(jms_helper)
      super(jms_helper, JmsHelper::QUEUE_READ_MESSAGES, JmsHelper::QUEUE_WRITE_MESSAGES)
    end
    #
    ### get a single status
    #
    def get_status(message_id, options={})
        jms_msg = new_jms_message(options.merge(:message_id => message_id, :action => 'Messages.get_status'))
        send_read_request(jms_msg)
        
    end
  
    #
    ### update status
    #
    def update_status(message_attrs, options={})
        jms_msg = new_jms_message({:message_id => GuidGenerator.next_message_id, :action => 'Messages.update_status'}.merge(options).merge(message_attrs))
        send_write_request(jms_msg)
        
    end
  
    #
    ### delete status
    #
    def destroy_message(message_id, options={})
        jms_msg = new_jms_message(options.merge(:message_id => message_id, :action => 'Messages.destroy_message'))
        send_write_request(jms_msg)
        
    end
  
  
  
    #
    ### retrieves most recent statuses for public
    #
    def public_timeline(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'Messages.public_timeline'))
        send_read_request(jms_msg)
        
    end
  
    #
    ### retrieves most recent statuses for user
    #
    def user_timeline(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'Messages.user_timeline'))
        send_read_request(jms_msg)
        
    end
  
  
    #
    ### retrieves most recent statuses for friends
    #
    def friends_timeline(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'Messages.friends_timeline'))
        send_read_request(jms_msg)
        
    end
  
  
    #
    ### retrieves most recent statuses for user and his/her friends
    #
    def user_and_friends_timeline(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'Messages.user_and_friends_timeline'))
        send_read_request(jms_msg)
        
    end
  end
 
 

---

./delegate/followers_delegate.rb

  #
  ## FollowersDelegate provides basic functionality for creating/quering followers
  ## It assumes all users are already authenticated before calling these methods.
  #
  class FollowersDelegate < BaseDelegate
    def initialize(jms_helper)
      super(jms_helper, JmsHelper::QUEUE_READ_FOLLOWERS, JmsHelper::QUEUE_WRITE_FOLLOWERS)
    end
    #
    ### create follower
    #
    def create_follower(follower_attrs, options={})
        jms_msg = new_jms_message(options.merge(follower_attrs).merge(:action => 'Followers.create_follower'))
        send_write_request(jms_msg)
        
    end
  
    #
    ### enable notifications
    #
    def enable_notifications(username, follower_username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :follower_username => follower_username, :action => 'Followers.enable_notifications'))
        send_write_request(jms_msg)
        
    end
  
    #
    ### disable notifications
    #
    def disable_notifications(username, follower_username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :follower_username => follower_username, :action => 'Followers.disable_notifications'))
        send_write_request(jms_msg)
        
    end
  
  
    #
    ### blocks a user
    #
    def block(username, follower_username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :follower_username => follower_username, :action => 'Followers.block'))
        send_write_request(jms_msg)
        
    end
  
    #
    ### unblock user
    #
    def unblock(username, follower_username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :follower_username => follower_username, :action => 'Followers.unblock'))
        send_write_request(jms_msg)
        
    end
  
  
    #
    ### delete status
    #
    def destroy_follower(username, follower_username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :follower_username => follower_username, :action => 'Followers.destroy_follower'))
        send_write_request(jms_msg)
        
    end
  
  
    #
    ### get a single status
    #
    def exists?(username, follower_username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :follower_username => follower_username, :action => 'Followers.exists?'))
        send_read_request(jms_msg)
        
    end
  end
 
 

---

./delegate/direct_messages_delegate.rb

  #
  ## DirectMessagesDelegate provides functionality for sending/receiving direct messages
  #
  class DirectMessagesDelegate < MessagesDelegate
    def initialize(jms_helper)
      super(jms_helper)
    end
  
    #
    ### sents back 20 most recent direct messages for the user
    ### it assumes user is already authenticated.
    #
    def direct_messages_received(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'DirectMessages.direct_messages_received'))
        send_read_request(jms_msg)
        
    end
    #
    ### sents back 20 most recent direct messages for the user
    ### it assumes user is already authenticated.
    #
    def direct_messages_sent(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'DirectMessages.direct_messages_sent'))
        send_read_request(jms_msg)
        
    end
  
    #
    ### send direct message
    #
    def send_direct_message(message_attrs, options={})
        jms_msg = new_jms_message({:message_id => GuidGenerator.next_message_id, :action => 'DirectMessages.send_direct_message'}.merge(options).merge(message_attrs))
        send_write_request(jms_msg)
        
    end
  
    #
    ### send reply message
    #
    def send_reply(message_attrs, options={})
        jms_msg = new_jms_message({:message_id => GuidGenerator.next_message_id, :action => 'DirectMessages.send_reply'}.merge(options).merge(message_attrs))
        send_write_request(jms_msg)
        
    end
  
    #
    ### sents back 20 most recent replies for the user
    ### it assumes user is already authenticated.
    #
    def replies(username, options={})
        jms_msg = new_jms_message(options.merge(:username => username, :action => 'DirectMessages.replies'))
        send_read_request(jms_msg)
        
    end
  
  
    #
    ### delete direct message
    #
    def destroy_direct_message(message_id, options={})
        jms_msg = new_jms_message(options.merge(:message_id => message_id, :action => 'DirectMessages.destroy_direct_message'))
        send_write_request(jms_msg)
        
    end
  end
 
 

---

Source Code Message Handlers

./messaging/base_handler.rb

  #
  ## BaseHandler provides common functionality for handling messages
  #
  class BaseHandler 
    attr_reader :username
    attr_reader :message_text
    attr_reader :timings
    def initialize
      @timings = []
      @started_at = Time.now
      @host_id = Digest::MD5.hexdigest("localhost")       #lookup real host
    end
  
    def handle(jms_msg, options={})
      add_timing_begin
      msg_attrs = {}
      pnames = jms_msg.getPropertyNames() 
      for pname in pnames
          msg_attrs[pname] = msg_attrs[pname.to_sym] = jms_msg.getObjectProperty(pname)
          case pname 
          when 'username'
            @username = msg_attrs[pname]
          when 'auth_user'
            @auth_user = msg_attrs[pname]
          when 'request_id'
            @request_id = msg_attrs[pname]
          end
      end
      if @username.nil?
          raise "username missing in #{jms_msg.inspect} options #{options.inspect}"
      end
      if jms_msg.respond_to? :text
          #msg_attrs.merge!(xml_to_dict(jms_msg.text)) 
          msg_attrs[:message_text] = jms_msg.text
      end
      ##
      add_timing_for_input_xml
      response = do_handle(msg_attrs)
      ##
      add_timing_end
      response
    rescue ActiveRecord::RecordInvalid => invalid
      response = xml_error("400", "handle", invalid.record.errors)
      add_timing_end
      response
    rescue java.sql.SQLException => e
      puts "Failed to handle #{self.class.name}(#{jms_msg} due to #{e.inspect}\n\n"
      while (e) 
        puts e.backtrace << "\n\n"
        e = e.getNextException() 
      end
      response = xml_error("500", "handle", {"UNKNOWN_ERROR" => e.to_s})
      add_timing_end
      response
    rescue Exception => e
      puts "Failed to handle #{self.class.name}(#{jms_msg} due to #{e.inspect}\n\n#{e.backtrace}"
      response = xml_error("500", "handle", {"UNKNOWN_ERROR" => e.to_s})
      add_timing_end
      response
    end
  
  protected
    def username_for_service
      @auth_user || @username
    end
  
    def add_timing(what)
      @timings << {what, (Time.now.to_i-@started_at.to_i)}
    end
    def add_timing_begin
      add_timing "#{self.class.name} begin"
    end
    def add_timing_end
      add_timing "#{self.class.name} end"
    end
    def add_timing_for_db
      add_timing "#{self.class.name} connected to db"
    end
    def add_timing_for_output_xml
      add_timing "#{self.class.name} converting output xml"
    end
    def add_timing_for_input_xml
      add_timing "#{self.class.name} converting input xml"
    end
  
    def do_handle(msg_attrs, options={})
      raise "Implement handle"
    end
    def xml_to_dict(xml)
      Hash.from_xml(xml)
    end
    def xml_error(response_code, action, errors, options={})
      #logger.error "Failed to #{action} due to #{error_code} by #{username} -- #{error_message}"
      to_xml(response_code, "Error response for #{action}", options) do |xml|
        xml.errors do 
          errors.each do |error_code, error_message|
            xml.error(:error_code => error_code, :error_message => error_message)
          end
        end
      end
    end
  
    def to_xml(response_code, comment="", options={})
      buffer = options[:buffer] ||= ''
  
      xml = options[:builder] ||= Builder::XmlMarkup.new(:target => buffer, :indent => options[:indent])
      xml.instruct! unless options[:skip_instruct]
      xml.comment! "Response #{comment} in reply for #{username} as of #{Time.now.utc}"
      xml.response(:request_id => @request_id, :response_code => response_code, :host_id => @host_id, :version => '1.0') do
        yield xml if block_given?
        xml.timings do
           @timings.each do |what, duration|
              xml.timing(:what => what, :duration_millis => duration)
           end
        end
      end
      buffer
    end
  end
 
 

---

./messaging/archive_messages_handler.rb

  #
  ## ArchiveMessagesHandler sends back archive messages for the user
  #
  class ArchiveMessagesHandler < BaseHandler
  protected
    def do_handle(msg_attrs, options={})
      svc = UsersService.new(self.username_for_service)
      add_timing_for_db
      messages = svc.archive_messages(self.username, options)
      add_timing_for_output_xml
      to_xml("200", "Archived Messages", options) do |xml|
        xml.messages do
           messages.each do |message|
             message.to_external_xml(xml)
           end
        end
      end
    end
  end
 
 

---

./messaging/authenticate_handler.rb

  ## AuthenticateHandler authenticates user
  #
  class AuthenticateHandler < BaseHandler
  protected
    def do_handle(msg_attrs, options={})
      password = msg_attrs[:password]
      svc = UsersService.new(self.username_for_service)
      add_timing_for_db
      user = svc.authenticate(username, password, options)
      add_timing_for_output_xml
      if user
        to_xml("200", "Authentication", options) do |xml|
           user.to_external_xml(xml)
        end
      else
        xml_error("401", "Authentication", {"AUTH_FAILURE" => "Invalid username/password"})
      end
    end
  end
 
 

---

./messaging/block_follower_handler.rb

  ## BlockFollowerHandler blocks follower
  #
  class BlockFollowerHandler < BaseHandler
  protected
    def do_handle(msg_attrs, options={})
      svc = FollowersService.new(self.username_for_service)
      follower_username = msg_attrs[:follower_username]
      add_timing_for_db
      blocked = svc.block(self.username, follower_username, options)
      response_code = blocked ? "200" : "404"
      add_timing_for_output_xml
      to_xml(response_code, "Block Follower", options) do |xml|
        xml.follower(:username => self.username, :follower_username => follower_username)
      end
    end
  end
 
 

---

Messaging Helpers

./messaging/message_forwarder.rb

  class MessageForwarder
    include java.lang.Runnable
    include javax.jms.MessageListener
  
    def initialize(jms_helper, queue_name, handlers)
      @jms_helper = jms_helper
      @queue_name = queue_name
      @handlers = handlers
    end
  
    def run
      @consumer = @jms_helper.get_consumer(@queue_name)
      @consumer.set_message_listener(self);
      puts "Starting listener for queue #{@queue_name}"
    end
  
    def onMessage(jms_msg)
      action = jms_msg.getProperty('action')
      if action.nil?
         puts "Unknown message #{jms_msg.inspect}"
      else
         handler = @handlers[action]
         if handler.nil?
            puts "No handler for #{action} -- #{jms_msg.inspect}"
         else
            
            response = handler.new().handle(jms_msg)
            
            if jms_msg.getJMSReplyTo()
               reply_queue = jms_msg.getJMSReplyTo()
               reply_producer = @jms_helper.create_producer(reply_queue)
               reply_message = @jms_helper.create_message(response)
               reply_message.setJMSCorrelationID(jms_msg.getJMSMessageID())
               reply_producer.send(reply_message)
            end
         end
      end
    end
  
    def close
      @consumer.close();
    end
  end
 
 

---

./messaging/jms_helper.rb

  
  class JmsHelper
    #Object.module_eval("::#{i}")
    #http://java.sun.com/j2ee/1.4/docs/tutorial/doc/JMS6.html
    #
    ### A key tip for scalability is to separate your reads from writes that way you can scale them independently and also
    ### offer different quality of services, e.g. read queues can be non-persistent, but write queues can be persistent.
    ###
    #
      QUEUE_READ_MESSAGES = 'read_messages'
      QUEUE_WRITE_MESSAGES = 'write_messages'
      QUEUE_READ_FOLLOWERS = 'read_followers'
      QUEUE_WRITE_FOLLOWERS = 'write_followers'
      QUEUE_READ_USERS = 'read_users'
      QUEUE_WRITE_USERS = 'write_users'
  
      READ_MESSAGES_HANDLERS = {
  	'DirectMessages.direct_messages_received' => DirectMessagesReceivedHandler,
  	'DirectMessages.direct_messages_sent' => DirectMessagesSentHandler,
  	'DirectMessages.replies' => RepliesHandler,
  	'Messages.get_status' => GetStatusHandler,
  	'Messages.public_timeline' => PublicTimelineHandler,
  	'Messages.user_timeline' => UserTimelineHandler,
  	'Messages.friends_timeline' => FriendsTimelineHandler,
  	'Messages.user_and_friends_timeline' => UserFriendTimelineHandler,
      }
      WRITE_MESSAGES_HANDLERS = {
  	'DirectMessages.send_direct_message' => SendDirectMessageHandler,
  	'DirectMessages.send_reply' => SendReplyMessageHandler,
  	'DirectMessages.destroy_direct_message' => DestroyDirectMessageHandler,
  	'Messages.update_status' => UpdateStatusHandler,
  	'Messages.destroy_message' => DestroyMessageHandler,
      }
      READ_FOLLOWERS_HANDLERS = {
  	'Followers.exists?' => FollowerExistsHandler,
      }
      WRITE_FOLLOWERS_HANDLERS = {
  	'Followers.create_follower' => CreateFollowerHandler,
  	'Followers.enable_notifications' => EnableNotificationsHandler,
  	'Followers.disable_notifications' => DisableNotificationsHandler,
  	'Followers.block' => BlockFollowerHandler,
  	'Followers.unblock' => UnblockFollowerHandler,
  	'Followers.destroy_follower' => DestroyFollowerHandler,
      }
      READ_USERS_HANDLERS = {
  	'Users.get_user' => GetUserHandler,
  	'Users.followings' => FollowingsHandler,
  	'Users.followers' => FollowersHandler,
  	'Users.authenticate' => AuthenticateHandler,
  	'Users.archive_messages' => ArchiveMessagesHandler,
      }
      WRITE_USERS_HANDLERS = {
  	'Users.create_user' => CreateUserHandler,
      }
  
    def initialize(connection=nil)
      if connection
         @connection = connection
      else
         factory = ActiveMQConnectionFactory.new("tcp://localhost:61616")
         @connection = factory.create_connection()
         @connection.set_exception_listener(self)
         @connection.start();
      end
      @queues = {}
      @producers = {}
      @consumers = {}
    end
  
    def start_consumers
      ### Creating another instance of jms helper because it keeps private session which cannot be shared
      ### across threads, though it can share connection.
      start_forwarder(QUEUE_READ_MESSAGES, READ_MESSAGES_HANDLERS)
      start_forwarder(QUEUE_WRITE_MESSAGES, WRITE_MESSAGES_HANDLERS)
      start_forwarder(QUEUE_READ_FOLLOWERS, READ_FOLLOWERS_HANDLERS)
      start_forwarder(QUEUE_WRITE_FOLLOWERS, WRITE_FOLLOWERS_HANDLERS)
      start_forwarder(QUEUE_READ_USERS, READ_USERS_HANDLERS)
      start_forwarder(QUEUE_WRITE_USERS, WRITE_USERS_HANDLERS)
      #
      java.lang.Thread.currentThread().join
    end
    
    def onException(jmsEx)
      puts "JMS Exception occured #{jmsEx.inspect}, shutting down"
    end
  
    def close
      @connection.close()
      @session.close if @session
    end
  
    def create_message(text)
      get_session().createTextMessage(text)
    end
  
    def send_message(queue_name, jms_msg, reply=false)
      ## TODO add error_queue/dead letter queue
      if reply
        reply_queue = get_session().createTemporaryQueue() 
        jms_msg.setJMSReplyTo(reply_queue)
      end
      
      get_producer(queue_name).send(get_queue(queue_name), jms_msg)
      if reply_queue 
          reply_consumer = get_session().createConsumer(reply_queue)
          reply_consumer.receive()
      end
    end
  
    def get_producer(queue_name)
        producer = @producers[queue_name]
        if producer.nil?
            producer = @producers[queue_name] = get_session().createProducer(get_queue(queue_name))
            #producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)     # this can be changed for write queues
            #producer.setTimeToLive(60000)
        end
        producer
    end
    def get_queue(queue_name)
       queue = @queues[queue_name]
       if queue.nil?
          queue = @queues[queue_name] = get_session().create_queue(queue_name)
       end
       queue
    end
  
    def get_consumer(queue_name)
        consumer = @consumers[queue_name]
        if consumer.nil?
            consumer = @consumers[queue_name] = get_session().create_consumer(get_queue(queue_name))
        end
        consumer
    end
  
    def create_producers(*queue_names)
      for queue_name in queue_names
        queue = get_queue(queue_name)
        producer = get_producer(queue_name)
      end
    end
  
    def create_producer(queue)
       get_session().createProducer(queue)
    end
  
    def self.start_broker
      broker = BrokerService.new()
      broker.addConnector("tcp://localhost:61616")
      broker.start()
      sleep 2    # let broker startup
    end
  protected    
    def get_session
      @session ||= @connection.create_session(false, Session::AUTO_ACKNOWLEDGE)
    end
    def start_forwarder(queue_name, handlers)
      raise "null connection" if @connection.nil?
      jms_helper = JmsHelper.new(@connection)
      Thread.new(){MessageForwarder.new(jms_helper, queue_name, handlers).run}
    end
  end
  
  if $PROGRAM_NAME == __FILE__
    JmsHelper.start_broker
    JmsHelper.new.start_consumers
  end
 
 

---

I am not showing all message handlers because they look pretty much the same.

Database

./db/schema_helper.rb

  require 'rubygems'
  require 'un'
  require 'active_record'
  require 'create_followers'
  require 'create_messages'
  require 'create_users'
  
  if defined?(JRUBY_VERSION)
    gem 'activerecord-jdbc-adapter'
    require 'jdbc_adapter'
    require 'java'
  end
  
  
  class SchemaHelper
    MAX_DB = 10
    #
    def self.create_schema()
      for i in 0 ... MAX_DB
         connect_to(i)
      end
    end
  
    def self.hash_for(username)
       hash = username.hash % MAX_DB
    end
  
    def self.setup_schema_for(username)
       hash = hash_for(username)
       connect_to(hash)
    end
  
    def self.connect_to(hash)
       dbdir = "#{FileUtils.pwd}/data/db#{hash}"
       dbfile = "#{dbdir}/database.sqlite"
       #ActiveRecord::Base.establish_connection(:adapter => "sqlite", :dbfile => dbfile)
       ActiveRecord::Base.establish_connection(
                  :adapter => "jdbc", 
                  :dbfile => dbfile,
                  #:driver => "SQLite.JDBCDriver",         ####:driver => "org.sqlite.JDBC", 
                  #:url => "jdbc:sqlite:#{dbfile}"
                  :driver => "org.apache.derby.jdbc.EmbeddedDriver",
                  :url => "jdbc:derby:#{dbdir};create=true"
          )
       unless File.directory?(dbdir)
          
          
          CreateUsers.migrate(:up)
          CreateMessages.migrate(:up)
          CreateFollowers.migrate(:up)
       end
    end
    def self.test
       SchemaHelper.setup_schema_for("shahbhat")
       require 'user'
       for i in 0 ... 10 do
         SchemaHelper.connect_to(i)
         puts User.count
       end
    end
  end
 
 

---