Multiple Servers in a Single Eventmachine Reactor

Multiple servers in a single EventMachine reactor

I tried this and it's working:

#!/usr/bin/env ruby

require 'eventmachine'

module EchoServer
  def post_init
    puts "-- someone connected to the echo server!"
  end
  def receive_data data
    send_data ">>>you sent: #{data}"
    close_connection if data =~ /quit/i
  end
  def unbind
    puts "-- someone disconnected from the echo server!"
  end
end

EventMachine::run {
  EventMachine::start_server "127.0.0.1", 8081, EchoServer
  EventMachine::start_server "127.0.0.1", 8082, EchoServer
  EventMachine::start_server "127.0.0.1", 8083, EchoServer
}

Here you get 3 echo services with different ports. (I was too lazy to implement different services.)

So, it's very easy to build a huge multi service wrapper.

---

Update

Simple code example for condition based start of a EM server:

#!/usr/bin/env ruby
# encoding: utf-8

require 'eventmachine'

module EchoServer
  def post_init
    puts "-- someone connected to the echo server!"
  end
  def receive_data data
    send_data ">>>you sent: #{data}"
    close_connection if data =~ /quit/i
  end
  def unbind
    puts "-- someone disconnected from the echo server!"
  end
end

$check_ok = false

EventMachine::run {
  puts "checker var is: #{$check_ok}"
  EventMachine::start_server "127.0.0.1", 8081, EchoServer
  EventMachine::start_server "127.0.0.1", 8082, EchoServer
  puts "echos on 8081 and 8082 started."

  # periodic timer check - every 1 sec
  EventMachine.add_periodic_timer(1) {
    if $check_ok
      EventMachine::start_server "127.0.0.1", 8083, EchoServer
      $check_ok = false
      puts "echo on 8083 started!"
    end
  }
  # timer triggered after 10 secs - only once!
  EventMachine.add_timer(10) {
    $check_ok = true
    puts "checker var is #{$check_ok} now!"
  }
}

In this example the echo server on port 8083 is started ~10 secs after app start. Try to telnet localhost 8083 before and after this timer, you'll see the effect.

You also can use values lower than 1 sec like 0.01 for every 1/100th sec checks.

This might be your starting point for your own ideas. The periodic timer is your internal loop, where you hook in your conditional checks for starting further services.

---

Good tutorial (PDF): eventmachine introduction (blog post)

How do I run Net::SSH and AMQP in the same EventMachine reactor?

Net::SSH is not asynchronous, so your EventMachine.run() is never reaching the end of the block, thus never resuming the reactor thread. This causes the AMQP code to never start. I would suggest running your SSH code within another thread.

Spec testing EventMachine-based (Reactor) Code

Sorry if this response is too pedantic, but I think you'll have a much easier time writing the tests and the library if you distinguish between your unit tests and your acceptance tests.

BDD vs. TDD

Be careful not to confuse BDD with TDD. While both are quite useful, it can lead to problems when you try to test every edge case in an acceptance test. For example, BDD is about testing what you're trying to accomplish with your service, which has more to do with what you're doing with the message queue than connecting to the queue itself. What happens when you try to connect to a non-existent message queue fits more into the realm of a unit test in my opinion. It's also worth pointing out that your service shouldn't be responsible for testing the message queue itself, since that's the responsibility of AMQP.

BDD

While I'm not sure what your service is supposed to do exactly, I would imagine your BDD tests should look something like:

1. start the service (can do this in a separate thread in the tests if you need to)
2. write something to the queue
3. wait for your service to respond
4. check the results of the service

In other words, BDD (or acceptance tests, or integration tests, however you want to think about them) can treat your app as a black box that is supposed to provide certain functionality (or behavior). The tests keep you focused on your end goal, but are more meant for ensuring one or two golden use cases, rather than the robustness of the app. For that, you need to break down into unit tests.

TDD

When you're doing TDD, let the tests guide you somewhat in terms of code organization. It's difficult to test a method that creates a new thread and runs EM inside that thread, but it's not so hard to unit test either of these individually. So, consider putting the main thread code into a separate function that you can unit test separately. Then you can stub out that method when unit testing the connect method. Also, instead of testing what happens when you try to connect to a bad server (which tests AMQP), you can test what happens when AMQP throws an error (which is your code's responsibility to handle). Here, your unit test can stub out the response of AMQP.connect to throw an exception.

How to connect to multiple WebSockets with Ruby?

Here's one way to do it.

First, you have to accept that the EM thread needs to be running. Without this thread you won't be able to process any current connections. So you just can't get around that.

Then, in order to add new URLs to the EM thread you then need some way to communicate from the main thread to the EM thread, so you can tell it to launch a new connection. This can be done with EventMachine::Channel.

So what we can build now is something like this:

@channel = EventMachine::Channel.new

Thread.new {
  EventMachine.run {
     @channel.subscribe { |url| 
       ws = Faye::...new(url)
       ...
    }
  }
}

Then in the main thread, any time you want to add a new URL to the event loop, you just use this:

def setup_socket(url)
  @channel.push(url)
end

Using EventMachine with Thin

Thin already provides and manages an EventMachine reactor, so you don't need to set one up seperately. I don't think you need to embed this code in the EventMachine.run {} block for starters.

There are a few problems with your approach here. Firstly, the returned @response variable will always be nil, because the EventMachine::HttpRequest happens asynchronously, and will not give you any data until you hit the http.callback {} block. Secondly, in each of the EventMachine::HttpRequest callbacks, you're calling EventMachine.stop. This will have the effect of stopping the thin webserver, which is probably why you're seeing the server bombing.

If you're trying to run this sort of code within a rails app, you might need to find a way to handle the calls asynchronously, so that the app doesn't hang while waiting for a long-running process to happen. A good approach I've used for this is to use Sinatra, which has an async plugin, allowing you to hold open long-running requests. You can then include this in your rails3 app using rails metal, so that requests to your async/eventmachine code are routed to sinatra.

---

Related Topics