client.rb

#!/usr/bin/ruby

require 'socket'
require 'rubygems'
require "xmpp4r"
require "xmpp4r/pubsub"
require "xmpp4r/pubsub/helper/servicehelper.rb"
require "xmpp4r/pubsub/helper/nodehelper.rb"
require "xmpp4r/pubsub/helper/nodehelper.rb"

module BL
  class Jabber
    def initialize(jid = nil, password = nil, hostname = nil)
      @jid = jid
      @password = password
      @hostname = hostname
    end
  
    def self.work_as(jid, password, hostname = nil, &jabber)
      jabber = self.new(jid, password, hostname)
      yield(jabber)
    end
  
    def method_missing(method, *args)
      UNIXSocket.open(AppConfig.xmpp.proxy_socket) do |socket|
        dump = Marshal.dump({:klass => self.class,
          :method => method,
          :jid => @jid,
          :password => @password,
          :hostname => @hostname,
          :args => args})
        socket.sync = true
        socket.send(dump, 0)
        Marshal.load(socket.read)
      end
      # FIXME should handle Marshal ArgumentError
    rescue Errno::ECONNREFUSED
      sleep(1) and retry
    end
  end
  
  class PubSub < Jabber
  end
end

server.rb

require File.dirname(__FILE__) + '/../config/app_config'

require 'rubygems'
require 'eventmachine'
require "xmpp4r"
require "xmpp4r/pubsub"
require "xmpp4r/pubsub/helper/servicehelper.rb"
require "xmpp4r/pubsub/helper/nodehelper.rb"
require "xmpp4r/pubsub/helper/nodehelper.rb"

module BL
  class Connections < Hash
    include Singleton
  end
  
  class JabberProxy
    def self.work_as(jid, password, hostname, &client)
      jid = Jabber::JID.new(jid)
      pubsub_service = AppConfig.xmpp.pubsub.service
      hostname = hostname || AppConfig.xmpp.hostname
      client = (Connections.instance[jid.to_s] ||= Jabber::Client.new(jid))
      if client.is_disconnected?
        client.connect(hostname)
        client.auth(password)
        client.send(Jabber::Presence.new.set_type(:available))
      end
      if self == PubSub
        client = (Connections.instance["p" + jid.to_s] ||= Jabber::PubSub::ServiceHelper.new(client, pubsub_service))
      end
      yield(client)
    end
  end
  
  class PubSub < JabberProxy
  end
end
  
EM.run do
  EM.start_unix_domain_server(AppConfig.xmpp.proxy_socket) do |connection|
    def connection.receive_data(marshal_dump)
      m = Marshal.load(marshal_dump)
      # Run method on klass
      m[:klass].__send__(:work_as, m[:jid], m[:password], m[:hostname]) do |j|
        # Jabber::Client overloads Object.send, so Object.__send__ is used
        send_data(Marshal.dump(j.__send__(m[:method], *m[:args])))
      end
    rescue => e
      # FIXME send exceptions
      puts "Rescue: #{e}"
      send_data(Marshal.dump(false))
    ensure
      close_connection_after_writing
    end
  end
end

usage.rb

node_path = "/test_node"

BL::PubSub.work_as('[email protected]/Test', 'test123') do |pubsub|    
  node_config = Jabber::PubSub::NodeConfig.new(node_path, 
    {'pubsub#deliver_payloads'    => 1,
     'pubsub#persistence'         => 1,
     'pubsub#max_items'           => 1_000_000,
     'pubsub#subscription_allow'  => 'open',
     'pubsub#publish_model'       => 'open',
     'pubsub#send_item_subscribe' => 1,
     'pubsub#access_model'        => 'open'})
  pubsub.create_collection_node(node_path, node_config)
end

BL::PubSub.work_as('[email protected]/Test', 'test456') do |pubsub|
  pubsub.subscribe_to(node_path)
  pubsub.unsubscribe_from(node_path)
end

BL::PubSub.work_as('[email protected]/Test', 'test123') do |pubsub|    
  pubsub.delete_node(node_path)
end