Upgrade to Pro — share decks privately, control downloads, hide ads and more …

EventMachine Server Design With XMPP

EventMachine Server Design With XMPP

Sharing lessons learned building the Vines XMPP server with the Boulder Ruby group.

David Graham

April 17, 2012
Tweet

Other Decks in Programming

Transcript

  1. # test.ru app = proc do |env| sleep 20 [

    200, {'Content-Type' => 'text/html'}, ['Hello World'] ] end run app $ thin --rackup test.ru start $ curl http://localhost:3000 $ curl http://localhost:3000
  2. # test.ru app = proc do |env| sleep 20 [

    200, {'Content-Type' => 'text/html'}, ['Hello World'] ] end run app $ thin --rackup test.ru --threaded start $ curl http://localhost:3000 $ curl http://localhost:3000
  3. require 'eventmachine' module EchoServer def post_init puts "-- someone connected

    to the echo server!" end def receive_data data send_data ">>> you sent: #{data}" end end EventMachine::run do EventMachine::start_server "127.0.0.1", 8081, EchoServer puts 'running echo server on 8081' end Easy!
  4. require 'eventmachine' module EchoServer def post_init puts "-- someone connected

    to the echo server!" end def receive_data data send_data ">>> you sent: #{data}" end end EventMachine::run do EventMachine::start_server "127.0.0.1", 8081, EchoServer puts 'running echo server on 8081' end Easy! Maybe Not
  5. <stream:stream from='[email protected]' to='im.example.com' version='1.0' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'> <stream:stream from='im.example.com' id='t7AMCin9zjMNwQKDnplntZPIDEI=' to='[email protected]'

    version='1.0' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'> <stream:features> <starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'> <required/> </starttls> </stream:features>
  6. <stream:stream from='[email protected]' to='im.example.com' version='1.0' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'> <stream:stream from='im.example.com' id='t7AMCin9zjMNwQKDnplntZPIDEI=' to='[email protected]'

    version='1.0' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'> <stream:features> <starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'> <required/> </starttls> </stream:features> <starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'/>
  7. <stream:stream from='[email protected]' to='im.example.com' version='1.0' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'> <stream:stream from='im.example.com' id='t7AMCin9zjMNwQKDnplntZPIDEI=' to='[email protected]'

    version='1.0' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'> <stream:features> <starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'> <required/> </starttls> </stream:features> <starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'/> <proceed xmlns='urn:ietf:params:xml:ns:xmpp-tls'/>
  8. <message from='[email protected]/balcony' to='[email protected]'> <body>Wherefore art thou, Romeo?</body> </message> <presence from='[email protected]/balcony'>

    <show>away</show> <status>I’m not here!</status> </presence> <iq from='[email protected]/balcony' id='hu2bac18' type='get'> <query xmlns='jabber:iq:roster'/> </iq>
  9. Line Separated def post_init @parser = BufferedTokenizer.new end def receive_data(data)

    @parser.extract(data).each do |line| receive_line(line) end end
  10. 10.1. In-Order Processing An XMPP server MUST ensure in-order processing

    of the stanzas and other XML elements it receives over a given input stream from a connected client or remote server. XMPP RFC 6120
  11. Blocking Queue def post_init @nodes = Queue.new end def process_node_queue

    node = @nodes.pop # blocks process_node(node) process_node_queue end
  12. Asynchronous Queue def post_init @nodes = EM::Queue.new end def process_node_queue

    @nodes.pop do |node| process_node(node) process_node_queue end end
  13. it 'accepts valid plain auth passwords' do state = Stream::Client::Auth.new

    next_state = Stream::Client::BindRestart.new stream = MiniTest::Mock.new stream.expect(:write, nil, ['<success/>']) stream.expect(:advance, nil, [next_state]) node = "<auth mechanism='PLAIN'>#{@password}</auth>" state.node(node) assert stream.verify end Authentication Passes
  14. case @state when :start_document then start_document when :start_object then start_object

    when :start_string then start_string when :unicode_escape then unicode_escape when :start_surrogate_pair then start_surrogate when :start_negative_number then start_negative when :start_zero then start_zero when :start_float then start_float when :start_exponent then start_exponent when :start_int then start_int when :start_true then start_true when :start_false then start_false when :start_null then start_null when :end_key then end_key when :start_array then start_array when :end_value then end_value when :end_document then end_document end JSON::Stream
  15. case @state when :start_document case ch when "{" @state =

    :start_object @stack.push(:object) notify_start_document notify_start_object when "[" @state = :start_array @stack.push(:array) notify_start_document notify_start_array when WS # ignore else error("Expected object or array start") end when :start_object case ch JSON::Stream
  16. case @state when :start_document case ch when LEFT_BRACE @state =

    :start_object @stack.push(:object) notify_start_document notify_start_object when LEFT_BRACKET @state = :start_array @stack.push(:array) notify_start_document notify_start_array when WS # ignore else error("Expected object or array start") end when :start_object case ch when RIGHT_BRACE end_container(:object) when QUOTE @state = :start_string @stack.push(:key) when WS # ignore else error("Expected object key start") end when :start_string case ch when QUOTE if @stack.pop == :string @state = :end_value notify_value(@buf) else # :key @state = :end_key notify_key(@buf) end @buf = "" when BACKSLASH @state = :start_escape when CONTROL error('Control characters must be escaped') else @buf << ch end when :start_escape case ch when QUOTE, BACKSLASH, SLASH @buf << ch @state = :start_string when B @buf << "\b" @state = :start_string when F @buf << "\f" @state = :start_string when N @buf << "\n" @state = :start_string when R @buf << "\r" @state = :start_string when T @buf << "\t" @state = :start_string when U @state = :unicode_escape else error("Expected escaped character") end when :unicode_escape case ch when HEX @unicode << ch if @unicode.size == 4 codepoint = @unicode.slice!(0, 4).hex if codepoint >= 0xD800 && codepoint <= 0xDBFF error('Expected low surrogate pair half') if @stack[-1].is_a?(Fixnum) @state = :start_surrogate_pair @stack.push(codepoint) elsif codepoint >= 0xDC00 && codepoint <= 0xDFFF high = @stack.pop error('Expected high surrogate pair half') unless high.is_a?(Fixnum) pair = ((high - 0xD800) * 0x400) + (codepoint - 0xDC00) + 0x10000 @buf << pair @state = :start_string else @buf << codepoint @state = :start_string end end else error('Expected unicode escape hex digit') end when :start_surrogate_pair case ch when BACKSLASH @state = :start_surrogate_pair_u else error('Expected low surrogate pair half') end when :start_surrogate_pair_u case ch when U @state = :unicode_escape else error('Expected low surrogate pair half') end when :start_negative_number case ch when ZERO @state = :start_zero @buf << ch when DIGIT_1_9 @state = :start_int @buf << ch else error('Expected 0-9 digit') end when :start_zero case ch when POINT @state = :start_float @buf << ch when EXPONENT @state = :start_exponent @buf << ch else @state = :end_value notify_value(@buf.to_i) @buf = "" @pos -= 1 redo end when :start_float case ch when DIGIT @state = :in_float @buf << ch else error('Expected 0-9 digit') end when :in_float case ch when DIGIT @buf << ch when EXPONENT @state = :start_exponent @buf << ch else @state = :end_value notify_value(@buf.to_f) @buf = "" @pos -= 1 redo end when :start_exponent case ch when MINUS, PLUS, DIGIT @state = :in_exponent @buf << ch else error('Expected +, -, or 0-9 digit') end when :in_exponent case ch when DIGIT @buf << ch else error('Expected 0-9 digit') unless @buf =~ DIGIT_END @state = :end_value num = @buf.include?('.') ? @buf.to_f : @buf.to_i notify_value(num) @buf = "" @pos -= 1 redo end when :start_int case ch when DIGIT @buf << ch when POINT @state = :start_float @buf << ch when EXPONENT @state = :start_exponent @buf << ch else @state = :end_value notify_value(@buf.to_i) @buf = "" @pos -= 1 redo end when :start_true keyword(TRUE_KEYWORD, true, TRUE_RE, ch) when :start_false keyword(FALSE_KEYWORD, false, FALSE_RE, ch) when :start_null keyword(NULL_KEYWORD, nil, NULL_RE, ch)
  17. class Start < State def initialize(stream, success=TLS) super end def

    node(node) raise StreamErrors::NotAuthorized unless stream?(node) stream.start(node) doc = Document.new features = doc.create_element('stream:features') do |el| el << doc.create_element('starttls') do |tls| tls.default_namespace = NAMESPACES[:tls] tls << doc.create_element('required') end end stream.write(features) advance end end
  18. class Ready < State def node(node) stanza = to_stanza(node) raise

    StreamErrors::UnsupportedStanzaType unless stanza stanza.validate_to stanza.validate_from stanza.process end end
  19. # runs on thread pool operation = proc do sleep

    20 42 end # runs on the reactor thread callback = proc do |result| send_data(result) end EM.defer(operation, callback) Thread Pool
  20. user = find_user('[email protected]') user.name = 'Alice' user = save_user(user) puts

    user.name Callback Spaghetti find_user('[email protected]') do |user| user.name = 'Alice' save_user(user) do |user| puts user.name end end
  21. fiber = Fiber.new do Fiber.yield 1 Fiber.yield 2 Fiber.yield 3

    end fiber.resume #=> 1 fiber.resume #=> 2 fiber.resume #=> 3 fiber.resume #=> FiberError: dead fiber called Episode IV: A New Stack
  22. def find_user(username) fiber = Fiber.current op = proc { User.find_by_username(username)

    } cb = proc {|user| fiber.resume(user) } EM.defer(op, cb) Fiber.yield end Threads + Fibers
  23. def find_user(username) fiber = Fiber.current op = proc { User.find_by_username(username)

    } cb = proc {|user| fiber.resume(user) } EM.defer(op, cb) Fiber.yield end Boilerplate
  24. def self.defer(method) old = "_deferred_#{method}" alias_method old, method define_method method

    do |*args| fiber = Fiber.current op = proc { method(old).call(*args) } cb = proc {|result| fiber.resume(result) } EM.defer(op, cb) Fiber.yield end end Metaprogramming
  25. def find_user(username) # find user end defer :find_user def save_user(user)

    # save user end defer :save_user def find_vcard(jid) # find card end defer :find_vcard def save_vcard(jid, card) # save card end defer :save_vcard DRY
  26. def post_init @nodes = EM::Queue.new end def process_node_queue @nodes.pop do

    |node| Fiber.new do process_node(node) process_node_queue end.resume end end Fiber/Stanza