Uso de la gema de Couchbase para Ruby con EventMachine

Como habrán notado el nuevo gem de Ruby para Couchbase se ha lanzado recientemente. La versión 1.2.2 es principalmente una versión de mantenimiento con varias corrección de errores, pero aun así puedes probar una nueva característica experimental: la integración con EventMachine biblioteca. Esta publicación le ofrecerá una introducción rápida sobre cómo empezar a usar Couchbase Server con sus aplicaciones basadas en el modelo asíncrono de EventMachine.

La integración con EventMachine solo es (actualmente) accesible en sistemas tipo UNIX (como Linux, Solaris, BSD). Debido a que usa fibers, también requiere la versión 1.9 o posterior de MRI Ruby.

Configura tu entorno de pruebas

El primer paso es instalar la librería libcouchbase, que se encarga de todos los detalles de bajo nivel del protocolo de Couchbase. Puedes seguir el guía de instalación en la página oficial. Aquí solo replicaré los pasos necesarios para una caja GNU/Linux típica (estoy usando Debian unstable):

1. Instalar la clave PGP del repositorio:

   $ wget -O- https://packages.couchbase.com/ubuntu/couchbase.key | sudo apt-key add –

2. Configurar la fuente del repositorio. Aquí estoy usando el enlace para Ubuntu 12.04, pero en general no importa porque vamos a usar el complemento EventMachine, que está integrado en la propia gema. Los paquetes están en diferentes repositorios de paquetes construidos usando la misma base de código; la única diferencia es la versión de las bibliotecas de IO (libevent, libev) incluidos en la versión de la distribución.

   $ sudo wget -O/etc/apt/sources.list.d/couchbase.list https://packages.couchbase.com/ubuntu/couchbase-ubuntu1204.list

3. Instale los encabezados de libcouchbase, la biblioteca principal y los símbolos de depuración. Nuevamente, es posible que desee instalar herramientas de línea de comandos o uno de los backends de E/S, pero eso no es necesario para la tarea en cuestión.

   $ sudo apt-get update
   $ sudo sudo apt-get install libcouchbase-dev libcouchbase2-core libcouchbase-dbg

    

   Eso es todo. 

   Ahora necesitas instalar Servidor Couchbase, siga las instrucciones del sitio oficial. Después de la instalación, obtendrá la consola de administrador ejecutándose en https://localhost:8091 y también una API REST accesible en el mismo puerto. Realice los pasos de configuración inicial y eventualmente asignará un bucket con el nombre “default”.

4. Por último, debes instalar la gema. Es tan fácil como escribir esto en la terminal:

   $ gem install couchbase
   Construyendo extensiones nativas. Esto puede tomar un tiempo…
   Se instaló correctamente couchbase-1.2.2
   1 gema instalada
   Instalando la documentación ri para couchbase-1.2.2…
   Instalando la documentación de RDoc para couchbase-1.2.2…

Construcción de la aplicación

Para demostrar la integración, construyamos una aplicación de chat sencilla utilizando EventMachine y agreguemos el registro de todos los eventos de la misma en un bucket de Couchbase. Es sumamente fácil construir una aplicación asíncrona con EventMachine y para probarlo pondré el código fuente completo en esta publicación (también se encuentra en ejemplos/chat-em directorio de las fuentes de gemas).

class ChatServer < EM::Connection

  @@clients = []

  def post_init
    @username = nil
    enviar_datos(“*** ¿Cómo te llamas?n”)
  fin

  def receive_data(data)
    si @username
      broadcast(data.strip, @username)
    otro
      name = data.gsub(/s+|[[]]/, ”).strip[0..20]
      si nombre.empty?
        enviar_datos(“*** ¿Cómo te llamas?n”)
      otro
        @username = name
        @@clients.push(self)
        broadcast(“#{@username} se ha unido”)
        send_data(“*** ¡Hola, #{@username}!n”)
      fin
    fin
  fin

  desvincular
    @@clients.delete(self)
    broadcast(“#{@username} se ha ido”) si @username
  fin

  def broadcast(message, author = nil)
    prefijo = autor ? “” : “***”
    @@clients.each do |client|
      a menos que client == self
        client.send_data(“#{prefijo} #{mensaje}n”)
      fin
    fin
  fin

fin

EventMachine.run do
  #: presiona Control + C para detener
  Signal.trap(“INT”)  { EventMachine.stop }
  Signal.trap(“TERM”) { EventMachine.stop }

  EventMachine.start_server(“0.0.0.0”, 9999, ChatServer)
fin

Este es un servidor típico de EventMachine basado en EM::Connection. Para aquellos que no conocen el significado de estos métodos redefinidos, aquí hay un extracto del documentación oficial:

EventMachine::Connection es una clase que es instanciada por el bucle de procesamiento de EventMachine cada vez que se crea una nueva conexión. (Las nuevas conexiones pueden iniciarse localmente hacia un servidor remoto o aceptarse localmente desde un cliente remoto.) Cuando se instancia un objeto Connection, este incorpora la funcionalidad contenida en el módulo definido por el usuario especificado en las llamadas a connect o start_server. Los módulos de gestión definidos por el usuario pueden redefinir cualquiera o todos los métodos estándar definidos aquí, así como agregar código adicional arbitrario que también será incorporado.

EventMachine administra un objeto heredado de EventMachine::Connection (y que contiene el código de usuario incorporado) por cada conexión de red que esté activa en un momento dado. El bucle de eventos llamará automáticamente a los métodos en los objetos EventMachine::Connection siempre que ocurran eventos específicos en las conexiones correspondientes, como se describe a continuación.

Esta clase nunca es instanciada por el código del usuario y no publica un método de inicialización. Los métodos de instancia de EventMachine::Connection que pueden ser llamados por el bucle de eventos son: #post_init, #connection_completed, #receive_data, #unbind, #ssl_verify_peer (si se utiliza TLS), #ssl_handshake_completed

Todos los demás métodos de instancia definidos aquí son llamados únicamente por el código del usuario.

El protocolo es muy simple y está orientado a líneas. Para cada conexión, EventMachine creará una instancia de ChatServer, que primero pide el nombre del nuevo participante y luego difunde todos sus mensajes al grupo. Puedes usar tu herramienta favorita que te permita comunicarte mediante un protocolo de texto arbitrario, como telnet por ejemplo o nc. Aquí hay una muestra de sesión entre extremos.

~ $ telnet localhost 9999 ??? ~ $ nc localhost 9999
Intentando con 127.0.0.1... ??? *** ¿Cuál es tu nombre?
Conectado a localhost. ??? alice
El carácter de escape es ‘^]’. ??? *** ¡Hola, alice!
*** ¿Cuál es tu nombre? ??? *** bob se ha unido
bob ??? Hola a todos
*** ¡Hola, bob! ??? ¡hola, bob! ¿cómo estás?
Hola a todos ??? ^C
¡Hola, Bob! ¿Cómo estás? ??? ~ $
*** alice ha salido                  ???
^]                                  ???
telnet> Conexión cerrada. ???
~ $                                 ???

Ahora es el momento de agregar un poco de Couchbase. Imagina que me gustaría conservar todos los mensajes en una base de datos distribuida de la manera más eficiente posible. Couchbase es la respuesta Para hacerlo, necesito:

Implementa un método log en la clase ChatServer, el cual debe aceptar el mensaje y un autor opcional (para eventos del sistema será nil):

def log(message, author = nil)
  Couchbase.bucket.incr(“log:key”, :initial => 1) do |res|
    entrada = {
      ‘time’ => Time.now.utc,
      ‘author’ => author || “[system]”,
      ‘message’ => message
    }
    Couchbase.bucket.set(“log:#{res.value}”, entry)
  fin
fin

Luego agrego una llamada a log(mensaje, autor) en el método de difusión justo antes de iterar por todos los clientes conectados. Y envuelvo EventMachine.start_server con la función de retorno Couchbase::Bucket#on_connect, para ejecutar el servidor justo después de que el cliente se haya conectado. La ejecución del bucle resultante se verá así:

EventMachine.run do
  #: presiona Control + C para detener
  Signal.trap(“INT”)  { EventMachine.stop }
  Signal.trap(“TERM”) { EventMachine.stop }

  Couchbase.connection_options = {:async => true, :engine => :eventmachine}
  Couchbase.bucket.on_connect do |res|
    si res.success?
      EventMachine.start_server(“0.0.0.0”, 9999, ChatServer)
    otro
      muestra el mensaje “No se puede conectar a Couchbase Server: #{res.error}”
    fin
  fin
fin

¡Eso es todo por ahora! En el futuro podemos ampliar este ejemplo para usar técnicas más modernas como microsincronía y tal vez websockets. Consulta este blog para ver actualizaciones.

Puntos extra

Solo registrar puede no ser tan interesante; con Couchbase Server puedes realizar análisis simples con consultas de vistas utilizando la impresionante función incremental de Map-Reduce de Couchbase. Por ejemplo, aquí está la función Map para obtener todas las entradas en orden cronológico.

function (doc, meta) {
  if (doc.message) {
    if (doc.author == “[system]” && doc.time) {
      emit(new Date(doc.time), “*** ” + doc.message);
    } else {
      emit(new Date(doc.time), ” ” + doc.message);
    }
  }
}

Y la salida JSON.

{“total_rows”:6,”rows”:[
  {“id”:”log:1″,”key”:”2013-02-11T19:08:05.000Z”,”value”:”*** alice se ha unido”},
  {“id”:”log:2″,”key”:”2013-02-11T19:08:18.000Z”,”value”:”*** bob se ha unido”},
  {“id”:”log:3″,”key”:”2013-02-11T19:08:38.000Z”,”value”:” hola a todos”},
  {“id”:”log:4″,”key”:”2013-02-11T19:08:48.000Z”,”value”:” ¡hola, bob! ¿cómo estás?”},
  {“id”:”log:5″,”key”:”2013-02-11T19:08:58.000Z”,”value”:”*** alice ha salido”},
  {“id”:”log:6”,”key”:”2013-02-11T19:09:01.000Z”,”value”:”*** bob se ha ido"}
]}

Bien, eso es realmente todo por ahora. Disfruta de esta nueva función experimental. Será totalmente compatible en una próxima versión. Si tienes algún problema, por favor abre una incidencia en Rastreador de problemas del proyecto RCBC. Las correcciones y contribuciones siempre son bienvenidas y es de código abierto bajo una licencia Apache 2.0. Encontrarás el códigos fuente en GitHub.