WebSocket for Ruby learning

Recommended for you: Get network issues from WhatsUp Gold. Not end users.

Use the Ruby Eventmachine Websockets August 20th 2010

HTML5 has added many new features, enable the development of more convenient, more comfortable to use. Here we discuss one of the new features: WebSockets. We'll Ruby Eventmachine based gem to achieve.

First make sure that your browser supports WebSockets (such as Chrome, Safari, Firefox trunk).

Set EventMachine

Use the EventMachine for WebSockets is very simple, have a ready em-websockets gem. Start the EventMachine.run block, EventMachine:: WebSocket.start will do the rest of the living. It provides a socket object that resembles the javascript WebSocket spec with callbacks for onopen, onmessage, and onclose.

  1. require 'eventmachine'
  2. require 'em-websocket'

  3. @sockets = []
  4. EventMachine.run do
  5. EventMachine::WebSocket.start(:host => '', :port => 8080) do |socket|
  6. socket.onopen do
  7. @sockets <<socket
  8. end
  9. socket.onmessage do |mess|
  10. @sockets.each {|s| s.send mess}
  11. end
  12. socket.onclose do
  13. @sockets.delete socket
  14. end
  15. end
  16. end

In this example, the connected sockets is added to the array, broadcasting and trigger in onmessage.

The client

The client side script listens for keypresses and sends them to the websocket. On receiving a message, which is just a single letter in this case, the corresponding letter will be lit up.

  1. var socket;

  3. function animateCharacter(letter)
  4. {
  5. var upper = letter.toUpperCase();
  6. $('#character_' + upper)
  7. .stop(true,true)
  8. .animate({ opacity: 1}, 100)
  9. .animate({ opacity: .2}, 100);
  10. }

  11. function setup()
  12. {
  13. var target = $('#alphabet');
  14. for(var i = 0; i <=alphabet.length; i++)
  15. {
  16. var char = alphabet.charAt(i);
  17. target.append('<span id="character_' + char +'">' + char + '</span');
  18. }
  19. connect();

  20. $(document).keypress(function(e){
  21. var char = String.fromCharCode(e.keyCode);
  22. socket.send(char);
  23. });
  24. };

  25. function connect(){
  26. socket = new WebSocket('ws://h.jxs.me:8080');
  27. socket.onmessage = function(mess) {
  28. animateCharacter(mess.data);
  29. };

  30. };

  31. window.onload += setup();

With it all put together, WebSockets makes for a simple way to connect users together through the browser without plugins, polling, or other ugly frameworks.

Install EventMachine

Ruby gem has been provided by EventMachine, the name is eventmachine. Run the gem install eventmachine can be installed.
Remind a bit, EventMachine takes the system installed C+ + the compiler used to compile the machine, binary expansion module.

The use of EventMachine

Learning from examples, let us begin.

echo service

Echo service is a traditional UNIX service, its function is to accept input information network connected to the client terminal, then it returns a byte by byte output to client. Use EventMachine to write a very simple.

  1. #!/usr/bin/env ruby

  2. require 'rubygems'
  3. require 'eventmachine'

  4. module EchoServer
  5. def receive_data(data)
  6. send_data(data)
  7. end
  8. end

  9. EventMachine::run do
  10. host = ''
  11. port = 8080
  12. EventMachine::start_server host, port, EchoServer
  13. puts "Started EchoServer on #{host}:#{port}..."
  14. end

Running the above code will start a echo listener process on port 8080, let us step by step analysis code.

First look at the lower part of the EventMachine: Code: run call. He has opened a cycle of events, followed by a block of code, commonly used to start clients or servers, he will always exist, in this life cycle does not terminate, unless we direct call EventMachine:: stop_event_loop .

This example we use the EventMachine:: start_server to start our echo service. The first two parameters is the server and port number. Binding to, means that EchoServer services will be listening on port 8080 on any IP machine.

The third parameter is the event processor (handler). In general, the processor is a Ruby module is used to define the callback function, the module does not pollute the global name space. Here EchoServer defines only receive_data, whenever we receive data through the connection which is triggered.

Finally, the EchoServer module call in triggering receive_data after send_data, connection sends the data it will pass in front of initialization.

HTTP client

EventMachine can also be used to create clients. I only need to be replaced by the EventMachine:: EventMachine:: start_server connect. Here is an example, which is connected to a HTTP server, print out the received headers, the program is EventMachine wind oh.

  1. #!/usr/bin/env ruby

  2. require 'rubygems'
  3. require 'eventmachine'

  4. module HttpHeaders
  5. def post_init
  6. send_data "GET /\r\n\r\n"
  7. @data = ""
  8. end

  9. def receive_data(data)
  10. @data <<data
  11. end

  12. def unbind
  13. if @data =~ /[\n][\r]*[\n]/m
  14. $`.each {|line| puts ">>> #{line}" }
  15. end

  16. EventMachine::stop_event_loop
  17. end
  18. end

  19. EventMachine::run do
  20. EventMachine::connect 'microsoft.com', 80, HttpHeaders
  21. end

If you modify the'microsoft.com'to the ARGV[0], you can access the headers from the command line input.

Here we see a new call, post_init. He will call in the setting of connection. On the client side, it shows that you have successfully connected to the server, the server, suggests a new client connections.

We also use unbind calls, it in connection as trigger end closing. In this example, the server is shut down in sends all data. If the server application, this means that a client has been disconnected.

The unbind and post_init functions is complementary. Whenever the network connection is disconnected or created, former will be called. I can't understand is why not use class named like type, in order to better reflect this relationship. In any case, they are named.

This is the three main callback function. Processing network connection created when receiving data, processing, and connection off. Other basically the same category send_data, like but some function variants.

Introduction of eventmachine and how to avoid the callback confusion.

Event driven programming has become very fashionable, largely due to the node.js project grace. In fact, in the world of ruby. Through eventmachine, we have used many years event programming (eventmachine added to event IO ruby). Generally speaking, writing event driven code is complicated and confusing, but we actually wrote the code is very beautiful. You can only use special skills not much.

The biggest challenge is actually the real understanding of how to create a clean model abstract. Due to different requirements, the code structure is different, not careful planning, you will soon discover that you fall into the maze of callback (here is an image of the statement, called pasta). This article will explain some generic template, which mixed use the Twitter streaming API, Google 's language API, and a WebSocket server. There is absolutely no pasta, I promise!

After eventmachine entry on, we will discuss two general abstract. Delay in processing, the first object of class like asynchronous method invocation. The second is how to abstract code, in order to realize the multiple event trigger. Finally, we will add the WebSocket Server to demonstrate the parallel IO processing.

Eventmachine start:

The first is the installation of eventmachine: Gem install eventmachine

You can run the following code to test ruby test1.rb, terminate the program requires kill away.

  1. # test1.rb

  2. require 'rubygems'
  3. require 'eventmachine'

  4. EventMachine.run {
  5. EventMachine.add_periodic_timer(1) {
  6. puts "Hello world"
  7. }
  8. }

This program every second output a message, can not run, first check the personality problems.

How does it work? Require eventmachine, we call the EventMachine.run, it will receive a code block most parameters. Now we can ignore these, focus on the code block, not the eventmachine can not work ah (if you have a strong curiosity, can learn under reactor pattern.) In EventMachine.run, we call the add_periodic_timer, and introduced into another block of code. Here to tell eventmachine every 1 second trigger an event, and then call the code block. This is the first useful knowledge we have learned, this block of code is called callback.

You may want to use a cycle is not a simple loop {puts'hi'; sleep 1}, what you think is right. But I promise, looking back, you will know our way better.

The use of eventmachine in network IO.

Efficient IO is all about eventmachine, must understand this point. When you use the eventmachine network programming in IO, you can either directly use eventmachine, is either a lib eventmachine hook through extended (on the GitHub you will find examples of this, a lot of very good recognition, because most of them named beginning with em-). Then you need to carefully choose which gems to database, API programming.

If you choose undeserved can block reactor, is in the IO before the end of operation, eventmachine will not trigger any event. For example, if you use the library Net:: HTTP, sends a request to a URL, expected 10s response, during this period in the code above, the timer will not trigger, until the end of the operation. Have you completely lost concurrency.

We discuss the HTTP client. Eventmachine already comes with two different HTTP client, but has some problems, we recommend not using them, there is a more powerful module: em-http-request .

Installation: Gem install em-http-request let us send a HTTP request through it, have a look how it works. (Note: EM is the abbreviation of the EventMachine, we all love playing fewer words.!)
  1. require 'rubygems'
  2. require 'eventmachine'

  3. EM.run {
  4. require 'em-http'

  5. EM::HttpRequest.new('http://json-time.appspot.com/time.json').get.callback { |http|
  6. puts http.response
  7. }
  8. }

Again we're attaching a callback which is called once the request has completed. We're attaching it in a slightly different way to the timer above, which we'll discuss next.

Abstracting code that has a success or failure case

In the design of API interface, need a way to distinguish the response of success or failure. In Ruby, there are two commonly used methods. One is to return to the nil, a throwing an exception (such as ActiveRecord:: NotFound). Eventmachine provides a more elegant solution: the deferrable.

Deferrable is an object, through which you can add on the success or failure of the callback method, named callback and errback. If you want, you can view the source code code, Not too complicated.

Call HttpRequest#get in the preceding code, the returned is a deferrable (actually returns a EM:: HttpClient object, it is mix in to the EM:: Deferrable module). Of course there are more general way, is the direct use of EM:: DefaultDeferrable.

As an excuse to use a deferrable ourselves I've decided that it would be a jolly marvelous idea to look up the language of tweets as they arrive from the twitter streaming API.

Handily, the Google AJAX Language API allows you to get the language of any piece of text. It's designed to be used from the browser, but we won't let a small matter like that stop us for now (although you should if it's a real application).

When I'm trying out a new API I generally start with HTTParty (gem install httparty) since it's just so quick and easy to use. Warning: you can't use HTTParty with eventmachine since it uses Net::HTTP under the covers – this is just for testing!

  1. require 'rubygems'
  2. require 'httparty'
  3. require 'json'

  4. response = HTTParty.get("http://www.google.com/uds/GlangDetect", :query => {
  5. :v => '1.0',
  6. :q => "Sgwn i os yw google yn deall Cymraeg?"
  7. })

  8. p JSON.parse(response.body)["responseData"]

  9. # => {"isReliable"=>true, "confidence"=>0.5834181, "language"=>"cy"}

Cool, Google understands Welsh!

In this section of code will be modified to use the use em-http-request (HTTParty is the use of Net:: HTTP), we first be encapsulated into a class, take it with us then to write eventmachine version compare. Unable to determine the language nil is returned value.

  1. require 'rubygems'
  2. require 'httparty'
  3. require 'json'

  4. class LanguageDetector
  5. URL = "http://www.google.com/uds/GlangDetect"

  6. def initialize(text)
  7. @text = text
  8. end

  9. # Returns the language if it can be detected, otherwise nil
  10. def language
  11. response = HTTParty.get(URL, :query => {:v => '1.0', :q => @text})

  12. return nil unless response.code == 200

  13. info = JSON.parse(response.body)["responseData"]

  14. return info['isReliable'] ? info['language'] : nil
  15. end
  16. end

  17. puts LanguageDetector.new("Sgwn i os yw google yn deall Cymraeg?").language

It modifies the code to use em-http-request:

  1. require 'rubygems'
  2. require 'em-http'
  3. require 'json'

  4. class LanguageDetector
  5. URL = "http://www.google.com/uds/GlangDetect"

  6. include EM::Deferrable

  7. def initialize(text)
  8. request = EM::HttpRequest.new(URL).get({
  9. :query => {:v => "1.0", :q => text}
  10. })

  11. # This is called if the request completes successfully (whatever the code)
  12. request.callback {
  13. if request.response_header.status == 200
  14. info = JSON.parse(request.response)["responseData"]
  15. if info['isReliable']
  16. self.succeed(info['language'])
  17. else
  18. self.fail("Language could not be reliably determined")
  19. end
  20. else
  21. self.fail("Call to fetch language failed")
  22. end
  23. }

  24. # This is called if the request totally failed
  25. request.errback {
  26. self.fail("Error making API call")
  27. }
  28. end
  29. end

  30. EM.run {
  31. detector = LanguageDetector.new("Sgwn i os yw google yn deall Cymraeg?")
  32. detector.callback { |lang| puts "The language was #{lang}" }
  33. detector.errback { |error| puts "Error: #{error}" }
  34. }

Returns the result as follows:

  1. The language was cy

This code looks completely different. The biggest difference is that, since the introduction of the EM:: Deferrable, whatever we call success or failure, the corresponding callback or errback code will be executed.

As an exercise, you can try to modify the code, so it allows three mistakes, during the repeated calling method. This same reality extremely like through this package, we well hidden complexity, we are no longer concerned after the completion of the internal implementation details.

Abstract code, allowing multiple return multiple events.

Now we will enter the eventmachine most adept domain, contact the most exciting features.

We will construct a client, connect to the Twitter 's streaming API, when a message arrives, will trigger a series of events.

The use of Twitter 's streaming API, you only need to open a long active HTTP connection to the stream.twitter.com, and then waiting for the information into the. We will again use the em-http-request. Connect to the API, listening to all tweets, waiting for the new twitter arrived, the code is very simple:

  1. http = EventMachine::HttpRequest.new('http://stream.twitter.com/1/statuses/filter.json').post({
  2. :head => { 'Authorization' => [username, password] },
  3. :query => { "track" => "newtwitter" }
  4. })

This returns a deferrable (for requests completed trigger corresponding callback function), in fact we use another technique. We can also register to be notified every time new data is received:

  1. http.stream do |chunk|
  2. # This chunk may contain one or more tweets separated by \r\n
  3. end

Let us put together the code, monitor tweets:

  1. require 'rubygems'
  2. require 'em-http'
  3. require 'json'

  4. EM.run {
  5. username = 'yourtwitterusername'
  6. password = 'password'
  7. term = 'newtwitter'
  8. buffer = ""

  9. http = EventMachine::HttpRequest.new('http://stream.twitter.com/1/statuses/filter.json').post({
  10. :head => { 'Authorization' => [ username, password ] },
  11. :query => { "track" => term }
  12. })

  13. http.callback {
  14. unless http.response_header.status == 200
  15. puts "Call failed with response code #{http.response_header.status}"
  16. end
  17. }

  18. http.stream do |chunk|
  19. buffer += chunk
  20. while line = buffer.slice!(/.+\r\n/)
  21. tweet = JSON.parse(line)
  22. puts tweet['text']
  23. end
  24. end
  25. }

Returns the information like.:

  1. Hey @Twitter. When shall I be getting the #NewTwitter?
  2. #NewTwitter #Perfect
  3. WHAHOO WTF? #NewTwitter is
  4. Buenos das a =) Estoy sola en la office, leyendo Le Monde y probando el #NewTwitter desde FireFox, que funciona de
  5. Curiosity and boredom got the better of me...I

It works! Now say we wanted to lookup the language of each tweet using the class we built earlier. We could do this by adding further to our stream method, however this is the road to callback hell (and just imagine what it would be like if we hadn't abstracted the language detection!).

  1. http.stream do |chunk|
  2. buffer += chunk
  3. while line = buffer.slice!(/.+\r\n/)
  4. tweet = JSON.parse(line)
  5. text = tweet['text']
  6. detector = LanguageDetector.new(text)
  7. detector.callback { |lang|
  8. puts "Language: #{lang}, tweet: #{text}"
  9. }
  10. detector.errback { |error|
  11. puts "Language could not be determined for tweet: #{text}"
  12. }
  13. end
  14. end

We rewrite optimization code.

We use deferrable to draw from the code, for asynchronous processing returns success or failure. Another common technique used in eventmachine is to provide a callback onevent. Class like the interface code.:

  1. stream = TweetStream.new(username, password, term)
  2. stream.ontweet { |tweet| puts tweet }

As if by magic here is the code!

  1. require 'rubygems'
  2. require 'em-http'
  3. require 'json'

  4. class TwitterStream
  5. URL = 'http://stream.twitter.com/1/statuses/filter.json'

  6. def initialize(username, password, term)
  7. @username, @password = username, password
  8. @term = term
  9. @callbacks = []
  10. @buffer = ""
  11. listen
  12. end

  13. def ontweet(&block)
  14. @callbacks <<block
  15. end

  16. private

  17. def listen
  18. http = EventMachine::HttpRequest.new(URL).post({
  19. :head => { 'Authorization' => [ @username, @password ] },
  20. :query => { "track" => @term }
  21. })

  22. http.stream do |chunk|
  23. @buffer += chunk
  24. process_buffer
  25. end
  26. end

  27. def process_buffer
  28. while line = @buffer.slice!(/.+\r\n/)
  29. tweet = JSON.parse(line)

  30. @callbacks.each { |c| c.call(tweet['text']) }
  31. end
  32. end
  33. end

Now we can further optimize the code:

  1. EM.run {
  2. stream = TwitterStream.new('yourtwitterusername', 'pass', 'newtwitter')
  3. stream.ontweet { |tweet|
  4. LanguageDetector.new(tweet).callback { |lang|
  5. puts "New tweet in #{lang}: #{tweet}"
  6. }
  7. }
  8. }

Different IO mixed with a treatment process

Using eventmachine does not obstruct any IO operation, we have another advantage, the same processes can be mixed with different types of IO. As an example, we use WebSocket real-time delivery of tweets to the browser. Fortunately have a ready to use eventmachine WebSocket server, em-websocket (for use with the Pusher in a very kindlike).

Install: gem install em-websocket.

To start the service, the code as follows:

  1. stream.ontweet { |tweet|
  2. LanguageDetector.new(tweet).callback { |lang|
  3. puts "New tweet in #{lang}: #{tweet}"

  4. websocket_connections.each do |socket|
  5. socket.send(JSON.generate({
  6. :lang => lang,
  7. :tweet => tweet
  8. }))
  9. end
  10. }
  11. }

To perform manual, the code is very clean!
Recommended from our users: Dynamic Network Monitoring from WhatsUp Gold from IPSwitch. Free Download

Posted by Bartholomew at January 17, 2014 - 3:47 AM